diff --git a/Cargo.lock b/Cargo.lock index 275d7eef5..41c0a11e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1166,7 +1166,7 @@ dependencies = [ "jni 0.21.1", "js-sys", "libc", - "mach2", + "mach2 0.4.3", "ndk", "ndk-context", "oboe", @@ -3280,6 +3280,12 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +[[package]] +name = "lazycell" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" + [[package]] name = "leb128fmt" version = "0.1.0" @@ -3453,6 +3459,15 @@ dependencies = [ "libc", ] +[[package]] +name = "mach2" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a1b95cd5421ec55b445b5ae102f5ea0e768de1f82bd3001e11f426c269c3aea" +dependencies = [ + "libc", +] + [[package]] name = "macroific" version = "2.0.0" @@ -4609,6 +4624,7 @@ dependencies = [ "sha2 0.10.9", "shellexpand", "socketioxide", + "starship-battery", "sysinfo", "tar", "tempfile", @@ -5014,6 +5030,19 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" +[[package]] +name = "plist" +version = "1.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "740ebea15c5d1428f910cd1a5f52cebf8d25006245ed8ade92702f4943d91e07" +dependencies = [ + "base64", + "indexmap", + "quick-xml", + "serde", + "time", +] + [[package]] name = "png" version = "0.18.1" @@ -5327,6 +5356,15 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3" +[[package]] +name = "quick-xml" +version = "0.38.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c" +dependencies = [ + "memchr", +] + [[package]] name = "quinn" version = "0.11.9" @@ -6630,6 +6668,24 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "starship-battery" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd0efc2c44c92705be724265a0c758e3b7c120ea63817d2d684bab86fbeced9a" +dependencies = [ + "cfg-if", + "core-foundation 0.10.1", + "lazycell", + "libc", + "mach2 0.5.0", + "nix", + "num-traits", + "plist", + "uom", + "windows-sys 0.61.2", +] + [[package]] name = "stop-token" version = "0.7.0" @@ -7479,6 +7535,16 @@ version = "0.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6d49784317cd0d1ee7ec5c716dd598ec5b4483ea832a2dced265471cc0f690ae" +[[package]] +name = "uom" +version = "0.37.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd5cfe7d84f6774726717f358a37f5bca8fca273bed4de40604ad129d1107b49" +dependencies = [ + "num-traits", + "typenum", +] + [[package]] name = "ureq" version = "3.3.0" diff --git a/Cargo.toml b/Cargo.toml index a6c771898..761e5d447 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -117,6 +117,11 @@ enigo = "0.3" arboard = "3" rdev = "0.5" fs2 = "0.4" +# Cross-platform battery probe for the scheduler gate. Maintained fork of +# the abandoned `battery` crate; same `use battery::*;` API surface. Used +# only by `openhuman::scheduler_gate::signals` to decide when to throttle +# background LLM work on laptops. +starship-battery = "0.10" matrix-sdk = { version = "0.16", optional = true, default-features = false, features = ["e2e-encryption", "rustls-tls", "markdown"] } fantoccini = { version = "0.22.0", optional = true, default-features = false, features = ["rustls-tls"] } diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index e31ac14c9..27924d99c 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -858,6 +858,11 @@ fn register_domain_subscribers( } crate::openhuman::composio::register_composio_trigger_subscriber(); crate::openhuman::composio::start_periodic_sync(); + // Initialise the scheduler gate before any background AI workers + // start so they observe a real policy on their first iteration + // (otherwise they fall back to `Policy::Normal` and miss the + // initial throttle decision on battery-powered hosts). + crate::openhuman::scheduler_gate::init_global(&config); crate::openhuman::memory::tree::jobs::start(config.clone()); // Restart requests go through a subscriber so every trigger path shares diff --git a/src/openhuman/composio/periodic.rs b/src/openhuman/composio/periodic.rs index c36c263f6..cc6ed6264 100644 --- a/src/openhuman/composio/periodic.rs +++ b/src/openhuman/composio/periodic.rs @@ -8,7 +8,7 @@ //! //! Design notes: //! -//! * One global tick (60s) drives every provider — we don't spawn a +//! * One global tick (5min) drives every provider — we don't spawn a //! task per connection, because the number of connections per user //! is small and a single tick keeps the bookkeeping trivial. //! * Per-connection state (last sync timestamp) lives in a @@ -36,7 +36,14 @@ use super::providers::{get_provider, ProviderContext, SyncReason}; /// How often the scheduler wakes up to look for due syncs. Independent /// from per-provider `sync_interval_secs` — this just bounds how long /// past a provider's interval we might fire. -const TICK_SECONDS: u64 = 60; +/// +/// 5 min trades a little staleness for noticeably less foreground load: +/// each tick triggers an HTTP fetch + DB write per due connection, and +/// for users with several connected providers the old 60s cadence kept +/// the laptop visibly busy. Per-provider `sync_interval_secs` still +/// caps the *minimum* delay between actual syncs — this only loosens +/// the upper bound. +const TICK_SECONDS: u64 = 300; /// Process-wide guard so the scheduler is only started once even /// when both `start_channels` and `bootstrap_skill_runtime` call into @@ -228,7 +235,7 @@ mod tests { fn tick_seconds_is_sane_default() { // Sanity check: don't accidentally ship a 1-second tick. assert!(TICK_SECONDS >= 30); - assert!(TICK_SECONDS <= 300); + assert!(TICK_SECONDS <= 600); } #[test] diff --git a/src/openhuman/config/mod.rs b/src/openhuman/config/mod.rs index d1f320f9f..8c74a2df0 100644 --- a/src/openhuman/config/mod.rs +++ b/src/openhuman/config/mod.rs @@ -32,10 +32,11 @@ pub use schema::{ LocalAiConfig, MatrixConfig, MemoryConfig, ModelRouteConfig, MultimodalConfig, ObservabilityConfig, ProxyConfig, ProxyScope, ReflectionSource, ReliabilityConfig, ResourceLimitsConfig, RuntimeConfig, SandboxBackend, SandboxConfig, SchedulerConfig, - ScreenIntelligenceConfig, SecretsConfig, SecurityConfig, SlackConfig, StorageConfig, - StorageProviderConfig, StorageProviderSection, StreamMode, TelegramConfig, UpdateConfig, - VoiceActivationMode, VoiceServerConfig, WebSearchConfig, WebhookConfig, DEFAULT_MODEL, - MODEL_AGENTIC_V1, MODEL_CODING_V1, MODEL_REASONING_V1, + SchedulerGateConfig, SchedulerGateMode, ScreenIntelligenceConfig, SecretsConfig, + SecurityConfig, SlackConfig, StorageConfig, StorageProviderConfig, StorageProviderSection, + StreamMode, TelegramConfig, UpdateConfig, VoiceActivationMode, VoiceServerConfig, + WebSearchConfig, WebhookConfig, DEFAULT_MODEL, MODEL_AGENTIC_V1, MODEL_CODING_V1, + MODEL_REASONING_V1, }; pub use schema::{ clear_active_user, default_root_openhuman_dir, pre_login_user_dir, read_active_user_id, diff --git a/src/openhuman/config/schema/mod.rs b/src/openhuman/config/schema/mod.rs index b3e8f3e08..b1afc91d8 100644 --- a/src/openhuman/config/schema/mod.rs +++ b/src/openhuman/config/schema/mod.rs @@ -24,6 +24,7 @@ mod observability; mod proxy; mod routes; mod runtime; +mod scheduler_gate; mod storage_memory; mod tools; mod update; @@ -53,6 +54,7 @@ pub use proxy::{ }; pub use routes::{EmbeddingRouteConfig, ModelRouteConfig}; pub use runtime::{DockerRuntimeConfig, ReliabilityConfig, RuntimeConfig, SchedulerConfig}; +pub use scheduler_gate::{SchedulerGateConfig, SchedulerGateMode}; pub use storage_memory::{ MemoryConfig, MemoryTreeConfig, StorageConfig, StorageProviderConfig, StorageProviderSection, }; diff --git a/src/openhuman/config/schema/scheduler_gate.rs b/src/openhuman/config/schema/scheduler_gate.rs new file mode 100644 index 000000000..30aaea8aa --- /dev/null +++ b/src/openhuman/config/schema/scheduler_gate.rs @@ -0,0 +1,86 @@ +//! Scheduler-gate configuration — controls when background AI work runs. +//! +//! Consumed by [`crate::openhuman::scheduler_gate`]. + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, JsonSchema, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SchedulerGateMode { + /// Decide based on power + CPU + deployment-mode signals. + Auto, + /// Always run background AI flat-out (server / power-user setting). + AlwaysOn, + /// Never run background AI. User can still trigger work explicitly. + Off, +} + +impl SchedulerGateMode { + pub fn as_str(self) -> &'static str { + match self { + Self::Auto => "auto", + Self::AlwaysOn => "always_on", + Self::Off => "off", + } + } +} + +impl Default for SchedulerGateMode { + fn default() -> Self { + Self::Auto + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct SchedulerGateConfig { + /// Top-level mode — `auto` (default), `always_on`, or `off`. + #[serde(default)] + pub mode: SchedulerGateMode, + + /// Battery charge floor in `auto` mode, 0.0..=1.0. Below this and not on + /// AC, the gate throttles. Default: 0.80. + #[serde(default = "default_battery_floor")] + pub battery_floor: f32, + + /// CPU busy threshold (recent global usage, 0..100). Above this, the gate + /// throttles even when plugged in. Default: 70.0 (i.e. <30% headroom). + #[serde(default = "default_cpu_busy_threshold")] + pub cpu_busy_threshold_pct: f32, + + /// In `Throttled` mode, sleep this many ms before each LLM-bound job to + /// serialise workers and let the host catch up. Default: 30_000 (30s). + #[serde(default = "default_throttled_backoff_ms")] + pub throttled_backoff_ms: u64, + + /// In `Paused` mode, re-check the policy every this many ms so workers + /// resume promptly when the user toggles the gate back on. Default: + /// 60_000 (60s). + #[serde(default = "default_paused_poll_ms")] + pub paused_poll_ms: u64, +} + +fn default_battery_floor() -> f32 { + 0.80 +} +fn default_cpu_busy_threshold() -> f32 { + 70.0 +} +fn default_throttled_backoff_ms() -> u64 { + 30_000 +} +fn default_paused_poll_ms() -> u64 { + 60_000 +} + +impl Default for SchedulerGateConfig { + fn default() -> Self { + Self { + mode: SchedulerGateMode::default(), + battery_floor: default_battery_floor(), + cpu_busy_threshold_pct: default_cpu_busy_threshold(), + throttled_backoff_ms: default_throttled_backoff_ms(), + paused_poll_ms: default_paused_poll_ms(), + } + } +} diff --git a/src/openhuman/config/schema/types.rs b/src/openhuman/config/schema/types.rs index 35638e20c..d5f0bca83 100644 --- a/src/openhuman/config/schema/types.rs +++ b/src/openhuman/config/schema/types.rs @@ -54,6 +54,13 @@ pub struct Config { #[serde(default)] pub scheduler: SchedulerConfig, + /// Background-AI scheduler gate — throttles memory-tree digests, + /// embeddings, and other LLM-bound background work based on power + /// state, CPU pressure, and deployment mode. See + /// [`crate::openhuman::scheduler_gate`]. + #[serde(default)] + pub scheduler_gate: SchedulerGateConfig, + #[serde(default)] pub agent: AgentConfig, @@ -251,6 +258,7 @@ impl Default for Config { autocomplete: AutocompleteConfig::default(), reliability: ReliabilityConfig::default(), scheduler: SchedulerConfig::default(), + scheduler_gate: SchedulerGateConfig::default(), agent: AgentConfig::default(), context: ContextConfig::default(), model_routes: Vec::new(), diff --git a/src/openhuman/memory/tree/jobs/worker.rs b/src/openhuman/memory/tree/jobs/worker.rs index 76b101db0..58c9aa7a4 100644 --- a/src/openhuman/memory/tree/jobs/worker.rs +++ b/src/openhuman/memory/tree/jobs/worker.rs @@ -73,6 +73,19 @@ pub async fn run_once(config: &Config) -> Result { } async fn run_once_with_semaphore(config: &Config, llm_slots: Arc) -> Result { + // Cooperative throttle BEFORE `claim_next()`. Holding the DB claim + // across an awaited `wait_for_capacity()` would let `Paused` mode + // sit on the row past `DEFAULT_LOCK_DURATION_MS`, after which + // `recover_stale_locks()` would requeue it for another worker to + // pick up — duplicating side effects. Throttling here means + // non-LLM jobs (AppendBuffer/FlushStale) also experience the same + // gate delay, but that's fine: in Throttled mode the host is + // already overloaded and a 30s breather between any DB-write batch + // is welcome; in Paused mode the user has explicitly asked us to + // stand down. Returns immediately in Aggressive/Normal so plugged-in + // desktops with headroom pay zero cost. + crate::openhuman::scheduler_gate::wait_for_capacity().await; + let Some(job) = claim_next(config, DEFAULT_LOCK_DURATION_MS)? else { return Ok(false); }; diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index 69e9a6b5f..9de4461f3 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -47,6 +47,7 @@ pub mod providers; pub mod redirect_links; pub mod referral; pub mod routing; +pub mod scheduler_gate; pub mod screen_intelligence; pub mod security; pub mod service; diff --git a/src/openhuman/scheduler_gate/gate.rs b/src/openhuman/scheduler_gate/gate.rs new file mode 100644 index 000000000..94798753f --- /dev/null +++ b/src/openhuman/scheduler_gate/gate.rs @@ -0,0 +1,144 @@ +//! 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 crate::openhuman::config::{Config, SchedulerGateConfig}; +use crate::openhuman::scheduler_gate::policy::{decide, Policy}; +use crate::openhuman::scheduler_gate::signals::Signals; + +struct State { + cfg: SchedulerGateConfig, + signals: Signals, + policy: Policy, +} + +static STATE: OnceLock>> = 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 worker until the host is ready for LLM-bound work. +/// +/// * **Aggressive / Normal** — returns immediately. +/// * **Throttled** — sleeps `throttled_backoff_ms` so concurrent workers +/// serialise themselves and the host catches its breath between jobs. +/// * **Paused** — polls every `paused_poll_ms` until the policy changes. +/// +/// Designed so existing semaphore-bounded worker pools can keep their pool +/// size and just gain a per-job throttle in front of the existing +/// `semaphore.acquire()` call. +pub async fn wait_for_capacity() { + 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 => return, + }; + match policy { + Policy::Aggressive | Policy::Normal => return, + Policy::Throttled => { + tokio::time::sleep(Duration::from_millis(throttled_ms)).await; + return; + } + Policy::Paused => { + tokio::time::sleep(Duration::from_millis(paused_ms)).await; + // re-evaluate; user may have toggled the gate back on. + } + } + } +} diff --git a/src/openhuman/scheduler_gate/mod.rs b/src/openhuman/scheduler_gate/mod.rs new file mode 100644 index 000000000..c4f449a23 --- /dev/null +++ b/src/openhuman/scheduler_gate/mod.rs @@ -0,0 +1,30 @@ +//! Scheduler gate — gates background AI work on host conditions. +//! +//! Background AI tasks (memory-tree digests, embeddings, summarisation) used +//! to run flat-out and made the host visibly lag, especially on battery. +//! This module exposes a single decision point — [`current_policy`] — that +//! background workers consult before spending CPU/GPU on LLM-bound work. +//! +//! Signals (refreshed every 30s in a background sampler): +//! * power state — on AC, or battery >= 80% +//! * CPU usage — recent global usage; <70% means "idle enough" +//! * deployment mode — server/container hosts run flat-out +//! +//! Resulting [`Policy`]: +//! * [`Policy::Aggressive`] — server-mode; bypass throttles entirely +//! * [`Policy::Normal`] — desktop with headroom; run as scheduled +//! * [`Policy::Throttled`] — busy or on battery; serialise + slow down +//! * [`Policy::Paused`] — user opted out; defer indefinitely +//! +//! Cooperative throttling: callers `await gate::wait_for_capacity()` before +//! each unit of LLM-bound work. The future resolves immediately in +//! Aggressive/Normal, sleeps in Throttled, and re-polls in Paused so the +//! caller resumes the moment the user toggles the gate back on. + +pub mod gate; +pub mod policy; +pub mod signals; + +pub use gate::{current_policy, current_signals, init_global, wait_for_capacity}; +pub use policy::Policy; +pub use signals::Signals; diff --git a/src/openhuman/scheduler_gate/policy.rs b/src/openhuman/scheduler_gate/policy.rs new file mode 100644 index 000000000..5ccbb87e7 --- /dev/null +++ b/src/openhuman/scheduler_gate/policy.rs @@ -0,0 +1,188 @@ +//! Decision logic — turn raw [`Signals`] + user config into a [`Policy`]. + +use crate::openhuman::config::SchedulerGateConfig; +use crate::openhuman::scheduler_gate::signals::Signals; + +/// Background-AI scheduling tier. See module docs in `mod.rs` for semantics. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Policy { + Aggressive, + Normal, + Throttled, + Paused, +} + +impl Policy { + pub fn as_str(self) -> &'static str { + match self { + Self::Aggressive => "aggressive", + Self::Normal => "normal", + Self::Throttled => "throttled", + Self::Paused => "paused", + } + } +} + +/// Compute the current [`Policy`] from sampled signals + user config. +/// +/// Order of evaluation matters — explicit user overrides win first, then +/// deployment mode, then dynamic host signals. +pub fn decide(signals: &Signals, cfg: &SchedulerGateConfig) -> Policy { + use crate::openhuman::config::SchedulerGateMode; + + match cfg.mode { + SchedulerGateMode::Off => return Policy::Paused, + SchedulerGateMode::AlwaysOn => return Policy::Aggressive, + SchedulerGateMode::Auto => {} + } + + if signals.server_mode { + return Policy::Aggressive; + } + + // Clamp config-supplied thresholds so a malformed config.toml (e.g. + // `battery_floor = 1.5` or a negative cpu threshold) can't silently + // disable / force throttling — the field is `f32` and serde won't + // reject out-of-domain values for us. + let battery_floor = cfg.battery_floor.clamp(0.0, 1.0); + let cpu_threshold = cfg.cpu_busy_threshold_pct.clamp(0.0, 100.0); + + let battery_ok = signals.on_ac_power + || signals + .battery_charge + .map(|c| c >= battery_floor) + .unwrap_or(true); // no battery present == treat as plugged in + + let cpu_ok = signals.cpu_usage_pct <= cpu_threshold; + + if battery_ok && cpu_ok { + Policy::Normal + } else { + Policy::Throttled + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::config::{SchedulerGateConfig, SchedulerGateMode}; + + fn cfg(mode: SchedulerGateMode) -> SchedulerGateConfig { + SchedulerGateConfig { + mode, + battery_floor: 0.8, + cpu_busy_threshold_pct: 70.0, + throttled_backoff_ms: 30_000, + paused_poll_ms: 60_000, + } + } + + fn signals(on_ac: bool, charge: Option, cpu: f32, server: bool) -> Signals { + Signals { + on_ac_power: on_ac, + battery_charge: charge, + cpu_usage_pct: cpu, + server_mode: server, + } + } + + #[test] + fn off_mode_pauses() { + let p = decide( + &signals(true, None, 5.0, true), + &cfg(SchedulerGateMode::Off), + ); + assert_eq!(p, Policy::Paused); + } + + #[test] + fn always_on_overrides_signals() { + // discharging laptop at 10% with 99% CPU — still Aggressive. + let p = decide( + &signals(false, Some(0.10), 99.0, false), + &cfg(SchedulerGateMode::AlwaysOn), + ); + assert_eq!(p, Policy::Aggressive); + } + + #[test] + fn server_mode_is_aggressive() { + let p = decide( + &signals(false, None, 50.0, true), + &cfg(SchedulerGateMode::Auto), + ); + assert_eq!(p, Policy::Aggressive); + } + + #[test] + fn plugged_in_idle_is_normal() { + let p = decide( + &signals(true, Some(0.45), 20.0, false), + &cfg(SchedulerGateMode::Auto), + ); + assert_eq!(p, Policy::Normal); + } + + #[test] + fn battery_above_floor_is_normal() { + let p = decide( + &signals(false, Some(0.85), 20.0, false), + &cfg(SchedulerGateMode::Auto), + ); + assert_eq!(p, Policy::Normal); + } + + #[test] + fn battery_below_floor_throttles() { + let p = decide( + &signals(false, Some(0.30), 20.0, false), + &cfg(SchedulerGateMode::Auto), + ); + assert_eq!(p, Policy::Throttled); + } + + #[test] + fn busy_cpu_throttles_even_when_plugged_in() { + let p = decide( + &signals(true, Some(0.95), 90.0, false), + &cfg(SchedulerGateMode::Auto), + ); + assert_eq!(p, Policy::Throttled); + } + + #[test] + fn out_of_range_battery_floor_is_clamped() { + // 1.5 clamped to 1.0 — with charge < 1.0 on battery, must throttle. + let mut c = cfg(SchedulerGateMode::Auto); + c.battery_floor = 1.5; + let p = decide(&signals(false, Some(0.99), 10.0, false), &c); + assert_eq!(p, Policy::Throttled); + // -1.0 clamped to 0.0 — any non-zero charge passes the floor. + c.battery_floor = -1.0; + let p = decide(&signals(false, Some(0.05), 10.0, false), &c); + assert_eq!(p, Policy::Normal); + } + + #[test] + fn out_of_range_cpu_threshold_is_clamped() { + // 200.0 clamped to 100.0 — nothing above it, never throttles on CPU. + let mut c = cfg(SchedulerGateMode::Auto); + c.cpu_busy_threshold_pct = 200.0; + let p = decide(&signals(true, None, 99.0, false), &c); + assert_eq!(p, Policy::Normal); + // -10.0 clamped to 0.0 — any positive CPU usage throttles. + c.cpu_busy_threshold_pct = -10.0; + let p = decide(&signals(true, None, 5.0, false), &c); + assert_eq!(p, Policy::Throttled); + } + + #[test] + fn no_battery_treated_as_plugged_in() { + // Desktop / server with no battery sensor — treat as AC. + let p = decide( + &signals(false, None, 20.0, false), + &cfg(SchedulerGateMode::Auto), + ); + assert_eq!(p, Policy::Normal); + } +} diff --git a/src/openhuman/scheduler_gate/signals.rs b/src/openhuman/scheduler_gate/signals.rs new file mode 100644 index 000000000..09d340757 --- /dev/null +++ b/src/openhuman/scheduler_gate/signals.rs @@ -0,0 +1,156 @@ +//! Host signals: power state, CPU pressure, deployment mode. +//! +//! Sampled on a 30s cadence by [`crate::openhuman::scheduler_gate::gate`]; this +//! file just captures one snapshot at a time. + +use std::path::Path; +use std::sync::Mutex; +use std::time::Duration; + +use once_cell::sync::Lazy; +use sysinfo::System; + +#[derive(Debug, Clone, Copy)] +pub struct Signals { + pub on_ac_power: bool, + /// 0.0..=1.0, or `None` when no battery sensor is present (most servers). + pub battery_charge: Option, + /// Recent global CPU usage, 0..100. + pub cpu_usage_pct: f32, + pub server_mode: bool, +} + +impl Signals { + /// Sample once. Cheap (~ms-scale) — safe to call from a 30s background task. + pub fn sample() -> Self { + let (on_ac, charge) = sample_power(); + let cpu_usage_pct = sample_cpu(); + let server_mode = detect_server_mode(charge.is_none()); + Self { + on_ac_power: on_ac, + battery_charge: charge, + cpu_usage_pct, + server_mode, + } + } +} + +// ---- power --------------------------------------------------------------- + +fn sample_power() -> (bool, Option) { + // Env overrides win — useful for CI, container hosts that misreport, + // and manual debugging of the throttle path on a desktop. Only + // explicit truthy/falsy tokens count: garbage values yield None so + // the real probe still gets to answer (vs. silently coercing to + // "on battery" and triggering throttling on every misconfigured host). + let env_on_ac = std::env::var("OPENHUMAN_ON_AC_POWER").ok().and_then(|v| { + match v.to_ascii_lowercase().as_str() { + "1" | "true" | "yes" => Some(true), + "0" | "false" | "no" => Some(false), + _ => None, + } + }); + let env_charge = std::env::var("OPENHUMAN_BATTERY_CHARGE") + .ok() + .and_then(|v| v.parse::().ok()) + .map(|v| v.clamp(0.0, 1.0)); + if let (Some(ac), Some(c)) = (env_on_ac, env_charge) { + return (ac, Some(c)); + } + + match probe_battery() { + Ok(probe) => ( + env_on_ac.unwrap_or(probe.on_ac), + env_charge.or(probe.charge), + ), + Err(err) => { + // Probe failure on Linux often just means no /sys/class/power_supply + // entries (server, container) — treat as "plugged in, no battery" + // which yields Normal/Aggressive, not Throttled. Log once at debug + // because this fires every 30s on the sampler tick. + log::debug!("[scheduler_gate] battery probe failed: {err:#}"); + (env_on_ac.unwrap_or(true), env_charge) + } + } +} + +struct BatteryProbe { + on_ac: bool, + charge: Option, +} + +fn probe_battery() -> Result { + let manager = starship_battery::Manager::new()?; + let mut any = false; + let mut on_ac = true; // if all batteries report Charging/Full, we're on AC. + let mut total: f32 = 0.0; + let mut count: f32 = 0.0; + for maybe in manager.batteries()? { + let battery = maybe?; + any = true; + // Discharging is the only state that conclusively means "on battery". + // Unknown / Empty / Full / Charging all imply the AC adapter is + // present (or at minimum that the OS isn't draining the pack). + if matches!(battery.state(), starship_battery::State::Discharging) { + on_ac = false; + } + total += battery.state_of_charge().value; + count += 1.0; + } + let charge = if any && count > 0.0 { + Some((total / count).clamp(0.0, 1.0)) + } else { + None + }; + Ok(BatteryProbe { on_ac, charge }) +} + +// ---- cpu ----------------------------------------------------------------- + +static CPU_SYS: Lazy> = Lazy::new(|| Mutex::new(System::new())); + +fn sample_cpu() -> f32 { + // Two refreshes spaced ~MINIMUM_CPU_UPDATE_INTERVAL apart give sysinfo + // a real delta to compute usage from. The interval is small enough to + // run on the 30s sampler tick without noticeable cost. + let mut sys = match CPU_SYS.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + sys.refresh_cpu_usage(); + std::thread::sleep(Duration::from_millis( + sysinfo::MINIMUM_CPU_UPDATE_INTERVAL.as_millis() as u64 + 50, + )); + sys.refresh_cpu_usage(); + sys.global_cpu_usage() +} + +// ---- deployment mode ----------------------------------------------------- + +fn detect_server_mode(no_battery: bool) -> bool { + if let Ok(v) = std::env::var("OPENHUMAN_DEPLOYMENT") { + if v.eq_ignore_ascii_case("server") { + return true; + } + if matches!(v.to_ascii_lowercase().as_str(), "desktop" | "laptop") { + return false; + } + } + if std::env::var("KUBERNETES_SERVICE_HOST").is_ok() { + return true; + } + if Path::new("/.dockerenv").exists() { + return true; + } + // Heuristic of last resort: a Linux box with no battery and no display + // server set is almost certainly a server. We *don't* infer server-mode + // from "no battery" alone — desktops have no battery either. + if cfg!(target_os = "linux") + && no_battery + && std::env::var("DISPLAY").is_err() + && std::env::var("WAYLAND_DISPLAY").is_err() + { + return true; + } + false +}