From 65032870f26a850b3ecf386bd7affdb52c2cc6ba Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 6 May 2026 04:03:27 +0530 Subject: [PATCH] feat(scheduler_gate): throttling + power-aware execution for local LLM (#1073) (#1252) Co-authored-by: Claude Opus 4.7 (1M context) --- src/openhuman/agent/triage/evaluator.rs | 13 + src/openhuman/autocomplete/core/engine.rs | 6 +- src/openhuman/config/schema/scheduler_gate.rs | 25 ++ src/openhuman/learning/reflection.rs | 9 + .../local_ai/service/public_infer.rs | 114 +++++++- .../local_ai/service/public_infer_tests.rs | 126 +++++++++ .../local_ai/service/vision_embed.rs | 9 + src/openhuman/memory/tree/jobs/worker.rs | 47 ++-- src/openhuman/scheduler_gate/gate.rs | 218 +++++++++++++-- src/openhuman/scheduler_gate/mod.rs | 4 +- src/openhuman/scheduler_gate/policy.rs | 249 +++++++++++++++++- src/openhuman/voice/postprocess.rs | 9 +- 12 files changed, 781 insertions(+), 48 deletions(-) diff --git a/src/openhuman/agent/triage/evaluator.rs b/src/openhuman/agent/triage/evaluator.rs index a03200057..23eefe998 100644 --- a/src/openhuman/agent/triage/evaluator.rs +++ b/src/openhuman/agent/triage/evaluator.rs @@ -245,6 +245,19 @@ pub async fn run_triage_with_resolved( used_local = used_local, "[triage::evaluator] dispatching {AGENT_RUN_TURN_METHOD}" ); + + // Gate the LLM round-trip on the scheduler's global single slot + // ONLY when the resolved provider is local. Cloud routes already + // have their own (forthcoming) rate-limiter and are not the + // resource that the gate exists to protect — gating cloud here + // would just delay triage for no reason. + let _gate_permit = if used_local { + log::debug!("[triage::evaluator] local route — acquiring scheduler_gate llm permit"); + crate::openhuman::scheduler_gate::wait_for_capacity().await + } else { + None + }; + let response = match request_native_global::( AGENT_RUN_TURN_METHOD, request, diff --git a/src/openhuman/autocomplete/core/engine.rs b/src/openhuman/autocomplete/core/engine.rs index c4e9f2e66..36696847b 100644 --- a/src/openhuman/autocomplete/core/engine.rs +++ b/src/openhuman/autocomplete/core/engine.rs @@ -707,8 +707,12 @@ impl AutocompleteEngine { v }; + // Interactive variant — bypasses the scheduler_gate's LLM permit + // so per-keystroke autocomplete doesn't queue behind a memory-tree + // backfill or a triage turn. See `inline_complete_interactive` + // docs in `local_ai/service/public_infer.rs`. let generated = match service - .inline_complete( + .inline_complete_interactive( &config, &context, &config.autocomplete.style_preset, diff --git a/src/openhuman/config/schema/scheduler_gate.rs b/src/openhuman/config/schema/scheduler_gate.rs index 30aaea8aa..bf402f0e8 100644 --- a/src/openhuman/config/schema/scheduler_gate.rs +++ b/src/openhuman/config/schema/scheduler_gate.rs @@ -58,6 +58,26 @@ pub struct SchedulerGateConfig { /// 60_000 (60s). #[serde(default = "default_paused_poll_ms")] pub paused_poll_ms: u64, + + /// Hard CPU ceiling (recent global usage, 0..100). When the host CPU + /// climbs above this in `auto` mode, the gate flips to + /// `Paused { CpuPressure }` rather than just `Throttled` — every + /// background LLM call is held until the host calms down. Distinct + /// from `cpu_busy_threshold_pct`, which only triggers `Throttled`. + /// Default: 95.0. + #[serde(default = "default_cpu_severe_pct")] + pub cpu_severe_pct: f32, + + /// When `true`, `auto` mode only runs background LLM work while the + /// laptop is on AC power. On battery the gate flips to + /// `Paused { OnBattery }` — no background inference at all, + /// regardless of charge level. + /// + /// Default `false` to preserve the prior behavior (battery-floor + /// based throttling). Power-conscious users who never want + /// background inference on battery can flip this on. + #[serde(default)] + pub require_ac_power: bool, } fn default_battery_floor() -> f32 { @@ -72,6 +92,9 @@ fn default_throttled_backoff_ms() -> u64 { fn default_paused_poll_ms() -> u64 { 60_000 } +fn default_cpu_severe_pct() -> f32 { + 95.0 +} impl Default for SchedulerGateConfig { fn default() -> Self { @@ -81,6 +104,8 @@ impl Default for SchedulerGateConfig { cpu_busy_threshold_pct: default_cpu_busy_threshold(), throttled_backoff_ms: default_throttled_backoff_ms(), paused_poll_ms: default_paused_poll_ms(), + cpu_severe_pct: default_cpu_severe_pct(), + require_ac_power: false, } } } diff --git a/src/openhuman/learning/reflection.rs b/src/openhuman/learning/reflection.rs index 2099fe7af..e3bd89261 100644 --- a/src/openhuman/learning/reflection.rs +++ b/src/openhuman/learning/reflection.rs @@ -155,6 +155,15 @@ impl ReflectionHook { async fn run_reflection(&self, prompt: &str) -> anyhow::Result { match self.config.reflection_source { ReflectionSource::Local => { + // Local reflection acquires the scheduler_gate LLM + // permit transitively through `service.prompt` → + // `inference_with_temperature_internal`. Cloud + // reflection skips the gate (#1073 intentionally + // gates only local routes; cloud rate limiting is + // tracked separately). + log::debug!( + "[learning::reflection] local route — gate permit acquired via LocalAiService" + ); let service = crate::openhuman::local_ai::global(&self.full_config); service .prompt(&self.full_config, prompt, Some(512), true) diff --git a/src/openhuman/local_ai/service/public_infer.rs b/src/openhuman/local_ai/service/public_infer.rs index e6e7648de..4a57f661e 100644 --- a/src/openhuman/local_ai/service/public_infer.rs +++ b/src/openhuman/local_ai/service/public_infer.rs @@ -53,6 +53,65 @@ impl LocalAiService { style_instructions: Option<&str>, style_examples: &[String], max_tokens: Option, + ) -> Result { + self.inline_complete_internal( + config, + context, + style_preset, + style_instructions, + style_examples, + max_tokens, + /* gated = */ true, + ) + .await + } + + /// Latency-sensitive sibling of [`Self::inline_complete`] that + /// **bypasses the scheduler gate's LLM permit**. + /// + /// Per-keystroke autocomplete must not block waiting for a + /// long-running memory-tree backfill or a triage turn to release + /// the global single slot. The user is at the keyboard; if the + /// background pipeline is busy we'd rather race the autocomplete + /// turn against it than show stale or empty completions for the + /// duration of the backfill. + /// + /// This is the only path inside [`LocalAiService`] that opts out of + /// the gate. Every other entry point (`inference`, `prompt`, + /// `summarize`, `inline_complete`, `vision_prompt`, `embed`) + /// acquires before talking to Ollama. + pub async fn inline_complete_interactive( + &self, + config: &Config, + context: &str, + style_preset: &str, + style_instructions: Option<&str>, + style_examples: &[String], + max_tokens: Option, + ) -> Result { + log::trace!("[local_ai] inline_complete_interactive bypasses scheduler_gate permit"); + self.inline_complete_internal( + config, + context, + style_preset, + style_instructions, + style_examples, + max_tokens, + /* gated = */ false, + ) + .await + } + + #[allow(clippy::too_many_arguments)] + async fn inline_complete_internal( + &self, + config: &Config, + context: &str, + style_preset: &str, + style_instructions: Option<&str>, + style_examples: &[String], + max_tokens: Option, + gated: bool, ) -> Result { if !config.local_ai.enabled { return Ok(String::new()); @@ -102,6 +161,7 @@ impl LocalAiService { max_tokens.or(Some(24)), true, 0.05, + gated, ) .await?; Ok(sanitize_inline_completion(&raw, context)) @@ -128,6 +188,9 @@ impl LocalAiService { return Err("messages must not be empty".to_string()); } + // Multi-turn local chat is background LLM-bound work — gate it. + let _gate_permit = crate::openhuman::scheduler_gate::wait_for_capacity().await; + tracing::debug!( message_count = messages.len(), model = %crate::openhuman::local_ai::model_ids::effective_chat_model_id(config), @@ -223,6 +286,36 @@ impl LocalAiService { .await } + /// Latency-sensitive sibling of [`Self::inference`] that **bypasses + /// the scheduler gate's LLM permit**. + /// + /// Used by user-arrival paths where the user is staring at the + /// output (push-to-talk dictation cleanup, in particular). If we + /// queue these behind a long-running memory backfill, the user + /// experiences a frozen UI; better to race the call against + /// background work and accept the contention than to silently + /// degrade interactivity. + /// + /// Sibling to [`Self::inline_complete_interactive`] for autocomplete. + /// Every other entry point (`inference`, `prompt`, `summarize`, + /// `inline_complete`, `vision_prompt`, `embed`, `chat_with_history`) + /// remains gated. + pub(crate) async fn inference_interactive( + &self, + config: &Config, + system: &str, + prompt: &str, + max_tokens: Option, + no_think: bool, + ) -> Result { + log::trace!("[local_ai] inference_interactive bypasses scheduler_gate permit"); + self.inference_with_temperature_internal( + config, system, prompt, max_tokens, no_think, 0.2, /* allow_empty = */ false, + /* gated = */ false, + ) + .await + } + pub(crate) async fn inference_with_temperature( &self, config: &Config, @@ -239,11 +332,13 @@ impl LocalAiService { max_tokens, no_think, temperature, - false, + /* allow_empty = */ false, + /* gated = */ true, ) .await } + #[allow(clippy::too_many_arguments)] async fn inference_with_temperature_allow_empty( &self, config: &Config, @@ -252,6 +347,7 @@ impl LocalAiService { max_tokens: Option, no_think: bool, temperature: f32, + gated: bool, ) -> Result { self.inference_with_temperature_internal( config, @@ -260,7 +356,8 @@ impl LocalAiService { max_tokens, no_think, temperature, - true, + /* allow_empty = */ true, + gated, ) .await } @@ -275,11 +372,24 @@ impl LocalAiService { no_think: bool, temperature: f32, allow_empty: bool, + gated: bool, ) -> Result { if !matches!(self.status.lock().state.as_str(), "ready") { self.bootstrap(config).await; } + // Cooperative throttle + global single-slot acquisition for + // background LLM-bound work. Drop happens at end of scope so + // post-processing (status writes, logging) does NOT hold the + // permit any longer than necessary. Interactive autocomplete + // skips this via `gated = false` from + // `inline_complete_interactive`. + let _gate_permit = if gated { + crate::openhuman::scheduler_gate::wait_for_capacity().await + } else { + None + }; + let started = std::time::Instant::now(); let model_id = model_ids::effective_chat_model_id(config); diff --git a/src/openhuman/local_ai/service/public_infer_tests.rs b/src/openhuman/local_ai/service/public_infer_tests.rs index dbb6488fd..49a9d9575 100644 --- a/src/openhuman/local_ai/service/public_infer_tests.rs +++ b/src/openhuman/local_ai/service/public_infer_tests.rs @@ -162,3 +162,129 @@ async fn inline_complete_disabled_returns_empty_string() { .unwrap(); assert!(out.is_empty()); } + +#[tokio::test] +async fn inline_complete_interactive_disabled_returns_empty_string() { + // Interactive variant must match the gated variant on the + // disabled short-circuit so the autocomplete UX is identical. + let mut config = Config::default(); + config.local_ai.enabled = false; + let service = LocalAiService::new(&config); + let out = service + .inline_complete_interactive(&config, "ctx", "casual", None, &[], None) + .await + .unwrap(); + assert!(out.is_empty()); +} + +/// Interactive autocomplete (`inline_complete_interactive`) MUST NOT +/// block on a held LLM permit. Hold the global slot, race the +/// interactive variant against a tight deadline; if it queued behind +/// the permit it would deadlock or time out. +#[tokio::test] +async fn inline_complete_interactive_does_not_block_on_held_permit() { + let _guard = crate::openhuman::local_ai::LOCAL_AI_TEST_MUTEX + .lock() + .expect("local ai test mutex"); + + // Hold the global LLM permit for the duration of the test. + let _held = crate::openhuman::scheduler_gate::gate::try_acquire_llm_permit() + .expect("test must start with a free permit; previous test leaked one"); + + let app = Router::new().route( + "/api/generate", + post(|Json(_body): Json| async move { + Json(json!({ + "model": "test", + "response": "ip", + "done": true + })) + }), + ); + let base = spawn_mock(app).await; + unsafe { + std::env::set_var("OPENHUMAN_OLLAMA_BASE_URL", &base); + } + + let config = enabled_config(); + let service = ready_service(&config); + + // Tight 2s deadline — comfortably above mock RTT, well below any + // policy-paused-poll backoff. If the interactive call goes through + // the gate it'll never finish. + let result = tokio::time::timeout( + std::time::Duration::from_secs(2), + service.inline_complete_interactive(&config, "ctx", "casual", None, &[], Some(8)), + ) + .await; + + unsafe { + std::env::remove_var("OPENHUMAN_OLLAMA_BASE_URL"); + } + + let inner = result.expect("interactive variant must NOT block on held permit"); + assert!( + inner.is_ok(), + "interactive call should have completed: {inner:?}" + ); +} + +/// Counterpart: the gated `inline_complete` (and `prompt`/`summarize`) +/// MUST queue behind a held permit. We assert this with a try-style +/// race: spawn the gated call, give it time to enter the wait, then +/// confirm it hasn't completed. We then drop the permit and verify +/// the call resolves. +#[tokio::test] +async fn gated_inline_complete_blocks_on_held_permit() { + let _guard = crate::openhuman::local_ai::LOCAL_AI_TEST_MUTEX + .lock() + .expect("local ai test mutex"); + + let held = crate::openhuman::scheduler_gate::gate::try_acquire_llm_permit() + .expect("test must start with a free permit"); + + let app = Router::new().route( + "/api/generate", + post(|Json(_body): Json| async move { + Json(json!({ + "model": "test", + "response": "x", + "done": true + })) + }), + ); + let base = spawn_mock(app).await; + unsafe { + std::env::set_var("OPENHUMAN_OLLAMA_BASE_URL", &base); + } + + let config = enabled_config(); + let service = std::sync::Arc::new(ready_service(&config)); + let svc = service.clone(); + let cfg = config.clone(); + + let join = tokio::spawn(async move { + svc.inline_complete(&cfg, "ctx", "casual", None, &[], Some(8)) + .await + }); + + // Give the spawned task a chance to enter `wait_for_capacity`. + tokio::time::sleep(std::time::Duration::from_millis(80)).await; + assert!( + !join.is_finished(), + "gated inline_complete must block while permit is held" + ); + + // Release the permit; the gated call should now resolve. + drop(held); + let resolved = tokio::time::timeout(std::time::Duration::from_secs(2), join) + .await + .expect("gated call must resolve once permit is released") + .expect("join") + .expect("ollama call"); + assert!(!resolved.is_empty() || resolved.is_empty()); // sanity — value depends on sanitiser + + unsafe { + std::env::remove_var("OPENHUMAN_OLLAMA_BASE_URL"); + } +} diff --git a/src/openhuman/local_ai/service/vision_embed.rs b/src/openhuman/local_ai/service/vision_embed.rs index cf503e386..de00c9fa3 100644 --- a/src/openhuman/local_ai/service/vision_embed.rs +++ b/src/openhuman/local_ai/service/vision_embed.rs @@ -47,6 +47,10 @@ impl LocalAiService { return Err("no valid image payloads were provided".to_string()); } + // Vision generation is background LLM-bound work; gate it through + // the scheduler's global LLM permit. + let _gate_permit = crate::openhuman::scheduler_gate::wait_for_capacity().await; + let body = OllamaGenerateRequest { model: vision_model, prompt: prompt.trim().to_string(), @@ -147,6 +151,11 @@ impl LocalAiService { self.ensure_ollama_model_available(&embedding_model, "embedding") .await?; + // Embeds are bge-m3 calls (8K context, ~1.3 GB resident) — the + // single concurrent embed that has historically crashed the + // user's laptop when stacked with other Ollama work. Gate it. + let _gate_permit = crate::openhuman::scheduler_gate::wait_for_capacity().await; + let response = self .http .post(format!("{}/api/embed", ollama_base_url())) diff --git a/src/openhuman/memory/tree/jobs/worker.rs b/src/openhuman/memory/tree/jobs/worker.rs index d5f20ee02..6f563968d 100644 --- a/src/openhuman/memory/tree/jobs/worker.rs +++ b/src/openhuman/memory/tree/jobs/worker.rs @@ -1,12 +1,18 @@ //! Worker pool: claims jobs from `mem_tree_jobs`, dispatches them through -//! [`handlers::handle_job`], and settles the row. A small global semaphore -//! caps concurrent LLM-bound work; non-LLM jobs run unrestricted. +//! [`handlers::handle_job`], and settles the row. +//! +//! Concurrency control for LLM-bound work is delegated to +//! [`crate::openhuman::scheduler_gate`] — its global single-slot +//! semaphore (`LlmPermit`) is the one source of truth across this +//! worker, voice cleanup, autocomplete, triage, and reflection. The +//! worker itself just calls `wait_for_capacity()`; non-LLM jobs +//! (`AppendBuffer`, `FlushStale`) run without acquiring a permit. use std::sync::{Arc, OnceLock}; use std::time::Duration; use anyhow::Result; -use tokio::sync::{Notify, Semaphore}; +use tokio::sync::Notify; use crate::openhuman::config::Config; use crate::openhuman::memory::tree::jobs::handlers; @@ -14,12 +20,6 @@ use crate::openhuman::memory::tree::jobs::store::{ claim_next, mark_done, mark_failed, recover_stale_locks, DEFAULT_LOCK_DURATION_MS, }; -// Held at 1 to keep concurrent bge-m3 embed calls (8K context, ~1.3 GB -// resident each) from saturating local RAM. The cloud-chat path itself -// would be fine at higher concurrency, but every worker also runs a -// bge-m3 embed step, and 3 concurrent embeds at 8K context have -// crashed the laptop in practice. See feedback memory -// `feedback_local_llm_load.md`. const WORKER_COUNT: usize = 1; const POLL_INTERVAL: Duration = Duration::from_secs(5); @@ -47,18 +47,16 @@ pub fn start(config: Config) { let notify = WORKER_NOTIFY .get_or_init(|| Arc::new(Notify::new())) .clone(); - let llm_slots = Arc::new(Semaphore::new(1)); if let Err(err) = recover_stale_locks(&config) { log::warn!("[memory_tree::jobs] recover_stale_locks failed at startup: {err:#}"); } for idx in 0..WORKER_COUNT { let notify = notify.clone(); - let llm_slots = llm_slots.clone(); let cfg = config.clone(); tokio::spawn(async move { loop { - match run_once_with_semaphore(&cfg, llm_slots.clone()).await { + match run_once(&cfg).await { Ok(true) => continue, Ok(false) => { tokio::select! { @@ -80,14 +78,8 @@ pub fn start(config: Config) { } /// Claim and run a single job. Returns `true` when work was processed, -/// `false` when no eligible row was available. Test entry point — the -/// production worker loop calls [`run_once_with_semaphore`] directly. +/// `false` when no eligible row was available. pub async fn run_once(config: &Config) -> Result { - let llm_slots = Arc::new(Semaphore::new(1)); - run_once_with_semaphore(config, llm_slots).await -} - -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 @@ -99,19 +91,28 @@ async fn run_once_with_semaphore(config: &Config, llm_slots: Arc) -> // 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; + // + // For LLM-bound jobs the returned `LlmPermit` reserves the global + // single slot for the lifetime of `handle_job`. Non-LLM jobs + // (`AppendBuffer`, `FlushStale`) drop the permit before the + // handler runs so they don't block the slot. + let gate_permit = crate::openhuman::scheduler_gate::wait_for_capacity().await; let Some(job) = claim_next(config, DEFAULT_LOCK_DURATION_MS)? else { return Ok(false); }; - let permit = if job.kind.is_llm_bound() { - Some(llm_slots.acquire().await?) + let llm_permit = if job.kind.is_llm_bound() { + gate_permit } else { + // Non-LLM jobs don't need the global slot; release it so an + // LLM-bound caller waiting elsewhere in the process can run. + drop(gate_permit); None }; + let result = handlers::handle_job(config, &job).await; - drop(permit); + drop(llm_permit); match result { Ok(()) => { diff --git a/src/openhuman/scheduler_gate/gate.rs b/src/openhuman/scheduler_gate/gate.rs index 94798753f..826ca5545 100644 --- a/src/openhuman/scheduler_gate/gate.rs +++ b/src/openhuman/scheduler_gate/gate.rs @@ -8,11 +8,52 @@ 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. The cloud path +/// itself is bandwidth-bound but the gate also fronts triage and +/// reflection turns that fan out to `agent.run_turn`, where every local +/// route loads the same Ollama model — so a single global slot is the +/// safest contract until #1064 (per-toolkit triage toggle) and the +/// cloud-side rate limiter ship. +/// +/// See `feedback_local_llm_load.md` — backfills with multiple +/// simultaneous Ollama requests have crashed the user's laptop twice. +const LLM_SLOTS: usize = 1; + +static LLM_PERMITS: OnceLock> = OnceLock::new(); + +fn llm_permits() -> &'static Arc { + 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, @@ -110,35 +151,182 @@ pub fn current_signals() -> Signals { }) } -/// Cooperatively block a worker until the host is ready for LLM-bound work. +/// 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. /// -/// * **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. +/// 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. /// -/// 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() { +/// * **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 { 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, + 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, - Policy::Throttled => { - tokio::time::sleep(Duration::from_millis(throttled_ms)).await; - return; + Policy::Aggressive | Policy::Normal => { + return acquire_llm_permit_inner().await; } - Policy::Paused => { + 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 { + 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 { + 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); + } +} diff --git a/src/openhuman/scheduler_gate/mod.rs b/src/openhuman/scheduler_gate/mod.rs index c4f449a23..e317d7f78 100644 --- a/src/openhuman/scheduler_gate/mod.rs +++ b/src/openhuman/scheduler_gate/mod.rs @@ -25,6 +25,6 @@ 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 gate::{current_policy, current_signals, init_global, wait_for_capacity, LlmPermit}; +pub use policy::{PauseReason, Policy}; pub use signals::Signals; diff --git a/src/openhuman/scheduler_gate/policy.rs b/src/openhuman/scheduler_gate/policy.rs index 5ccbb87e7..71dd7875c 100644 --- a/src/openhuman/scheduler_gate/policy.rs +++ b/src/openhuman/scheduler_gate/policy.rs @@ -3,13 +3,48 @@ use crate::openhuman::config::SchedulerGateConfig; use crate::openhuman::scheduler_gate::signals::Signals; +/// Why the gate is currently paused. Carried by [`Policy::Paused`] so +/// downstream consumers (UI, logging, observability) can surface a +/// specific user-facing reason instead of a generic "paused" label. +/// +/// New variants will land alongside #1073's full power-aware work +/// (`OnBattery`, `CpuPressure`); `UserDisabled` covers the existing +/// `SchedulerGateMode::Off` path and `Unknown` is the safe fallback for +/// callers that don't have specific context yet. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PauseReason { + /// User explicitly turned the gate off in config. + UserDisabled, + /// Host on battery and gate's power-aware mode kicked in (#1073). + OnBattery, + /// CPU pressure exceeded the gate threshold (#1073). + CpuPressure, + /// Pause reason not yet classified — placeholder while #1073 is in flight. + Unknown, +} + +impl PauseReason { + pub fn as_str(self) -> &'static str { + match self { + Self::UserDisabled => "user_disabled", + Self::OnBattery => "on_battery", + Self::CpuPressure => "cpu_pressure", + Self::Unknown => "unknown", + } + } +} + /// 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, + /// Gate paused. The `reason` is rendered to users in the memory-sync + /// status UI (#1136) and recorded in observability. + Paused { + reason: PauseReason, + }, } impl Policy { @@ -18,7 +53,17 @@ impl Policy { Self::Aggressive => "aggressive", Self::Normal => "normal", Self::Throttled => "throttled", - Self::Paused => "paused", + Self::Paused { .. } => "paused", + } + } + + /// `Some(reason)` when paused, `None` otherwise. Convenience for + /// callers that only need the reason and don't want to pattern-match + /// the whole enum (UI badges, log line construction). + pub fn pause_reason(self) -> Option { + match self { + Self::Paused { reason } => Some(reason), + _ => None, } } } @@ -31,7 +76,11 @@ pub fn decide(signals: &Signals, cfg: &SchedulerGateConfig) -> Policy { use crate::openhuman::config::SchedulerGateMode; match cfg.mode { - SchedulerGateMode::Off => return Policy::Paused, + SchedulerGateMode::Off => { + return Policy::Paused { + reason: PauseReason::UserDisabled, + } + } SchedulerGateMode::AlwaysOn => return Policy::Aggressive, SchedulerGateMode::Auto => {} } @@ -46,6 +95,39 @@ pub fn decide(signals: &Signals, cfg: &SchedulerGateConfig) -> Policy { // 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 cpu_severe = cfg.cpu_severe_pct.clamp(0.0, 100.0); + + // ── Pause checks come BEFORE the throttle gate — these are the + // "stand down completely" signals. Hierarchy: + // 1. user policy (`require_ac_power` on battery) + // 2. host on fire (CPU severely pegged) + + // (1) Power-aware stand-down. Only consult `on_ac_power` when the + // user explicitly opts in — many desktops report `false` here + // because they have no battery + no AC sensor, and we don't + // want to silently disable background work for them. + if cfg.require_ac_power && !signals.on_ac_power { + log::debug!( + "[scheduler_gate] policy decision: paused on_battery (require_ac_power=true, on_ac={})", + signals.on_ac_power + ); + return Policy::Paused { + reason: PauseReason::OnBattery, + }; + } + + // (2) Hard CPU ceiling — at >= cpu_severe_pct the host is unusable; + // a Throttled 30s backoff is not enough, hold every job. + if signals.cpu_usage_pct >= cpu_severe { + log::debug!( + "[scheduler_gate] policy decision: paused cpu_pressure (cpu={:.1}% >= severe={:.1}%)", + signals.cpu_usage_pct, + cpu_severe, + ); + return Policy::Paused { + reason: PauseReason::CpuPressure, + }; + } let battery_ok = signals.on_ac_power || signals @@ -74,6 +156,8 @@ mod tests { cpu_busy_threshold_pct: 70.0, throttled_backoff_ms: 30_000, paused_poll_ms: 60_000, + cpu_severe_pct: 95.0, + require_ac_power: false, } } @@ -92,7 +176,36 @@ mod tests { &signals(true, None, 5.0, true), &cfg(SchedulerGateMode::Off), ); - assert_eq!(p, Policy::Paused); + assert_eq!( + p, + Policy::Paused { + reason: PauseReason::UserDisabled + } + ); + } + + #[test] + fn pause_reason_helper_returns_user_disabled_for_off_mode() { + let p = decide( + &signals(true, None, 5.0, false), + &cfg(SchedulerGateMode::Off), + ); + assert_eq!(p.pause_reason(), Some(PauseReason::UserDisabled)); + } + + #[test] + fn pause_reason_helper_returns_none_for_non_paused() { + assert_eq!(Policy::Aggressive.pause_reason(), None); + assert_eq!(Policy::Normal.pause_reason(), None); + assert_eq!(Policy::Throttled.pause_reason(), None); + } + + #[test] + fn pause_reason_as_str_round_trips_each_variant() { + assert_eq!(PauseReason::UserDisabled.as_str(), "user_disabled"); + assert_eq!(PauseReason::OnBattery.as_str(), "on_battery"); + assert_eq!(PauseReason::CpuPressure.as_str(), "cpu_pressure"); + assert_eq!(PauseReason::Unknown.as_str(), "unknown"); } #[test] @@ -166,8 +279,11 @@ mod tests { #[test] fn out_of_range_cpu_threshold_is_clamped() { // 200.0 clamped to 100.0 — nothing above it, never throttles on CPU. + // Also push `cpu_severe_pct` to its max so the new pause-on-severe + // arm doesn't trip first. let mut c = cfg(SchedulerGateMode::Auto); c.cpu_busy_threshold_pct = 200.0; + c.cpu_severe_pct = 100.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. @@ -185,4 +301,129 @@ mod tests { ); assert_eq!(p, Policy::Normal); } + + // ── Power-aware require_ac_power gate (#1073) ───────────────────── + + #[test] + fn require_ac_power_pauses_on_battery() { + let mut c = cfg(SchedulerGateMode::Auto); + c.require_ac_power = true; + // On battery, even with healthy charge + low CPU. + let p = decide(&signals(false, Some(0.95), 10.0, false), &c); + assert_eq!( + p, + Policy::Paused { + reason: PauseReason::OnBattery + } + ); + } + + #[test] + fn require_ac_power_normal_when_plugged_in() { + let mut c = cfg(SchedulerGateMode::Auto); + c.require_ac_power = true; + // Plugged in with headroom — should still run. + let p = decide(&signals(true, Some(0.90), 10.0, false), &c); + assert_eq!(p, Policy::Normal); + } + + #[test] + fn require_ac_power_off_preserves_legacy_behavior_on_battery() { + // Default `require_ac_power = false` and a fresh battery means + // the legacy path runs: battery >= floor ⇒ Normal. + let mut c = cfg(SchedulerGateMode::Auto); + c.require_ac_power = false; + let p = decide(&signals(false, Some(0.95), 10.0, false), &c); + assert_eq!(p, Policy::Normal); + } + + #[test] + fn require_ac_power_pause_resumes_when_back_on_ac() { + // Pause → re-evaluate after plugging in → Normal. + let mut c = cfg(SchedulerGateMode::Auto); + c.require_ac_power = true; + let s_battery = signals(false, Some(0.40), 5.0, false); + let s_ac = signals(true, Some(0.45), 5.0, false); + + let p1 = decide(&s_battery, &c); + assert!(matches!( + p1, + Policy::Paused { + reason: PauseReason::OnBattery + } + )); + let p2 = decide(&s_ac, &c); + assert_eq!(p2, Policy::Normal); + } + + // ── Hard CPU ceiling (#1073) ────────────────────────────────────── + + #[test] + fn cpu_severe_pauses_on_pressure() { + let mut c = cfg(SchedulerGateMode::Auto); + c.cpu_severe_pct = 90.0; + // CPU above severe ceiling, plugged in. + let p = decide(&signals(true, None, 96.0, false), &c); + assert_eq!( + p, + Policy::Paused { + reason: PauseReason::CpuPressure + } + ); + } + + #[test] + fn cpu_just_below_severe_throttles_not_pauses() { + let mut c = cfg(SchedulerGateMode::Auto); + c.cpu_busy_threshold_pct = 70.0; + c.cpu_severe_pct = 95.0; + // CPU above busy but below severe → Throttled, not Paused. + let p = decide(&signals(true, None, 80.0, false), &c); + assert_eq!(p, Policy::Throttled); + } + + #[test] + fn cpu_severe_recovers_to_normal() { + let mut c = cfg(SchedulerGateMode::Auto); + c.cpu_severe_pct = 90.0; + let s_pegged = signals(true, None, 99.0, false); + let s_idle = signals(true, None, 5.0, false); + assert!(matches!( + decide(&s_pegged, &c), + Policy::Paused { + reason: PauseReason::CpuPressure + } + )); + assert_eq!(decide(&s_idle, &c), Policy::Normal); + } + + #[test] + fn out_of_range_cpu_severe_pct_is_clamped() { + // 200.0 clamped to 100.0 — only true 100% CPU triggers pause. + let mut c = cfg(SchedulerGateMode::Auto); + c.cpu_severe_pct = 200.0; + let p = decide(&signals(true, None, 99.9, false), &c); + // 99.9 < 100.0 (clamped), so we don't hit the pause arm and + // fall through to Throttled (cpu_busy_threshold=70). + assert_eq!(p, Policy::Throttled); + // Negative clamps to 0.0 — any positive CPU usage pauses. + c.cpu_severe_pct = -10.0; + let p = decide(&signals(true, None, 0.5, false), &c); + assert_eq!( + p, + Policy::Paused { + reason: PauseReason::CpuPressure + } + ); + } + + #[test] + fn server_mode_overrides_pause_signals() { + // Even on battery + CPU pegged, server mode stays Aggressive. + let mut c = cfg(SchedulerGateMode::Auto); + c.require_ac_power = true; + c.cpu_severe_pct = 50.0; + let p = decide(&signals(false, None, 99.0, true), &c); + assert_eq!(p, Policy::Aggressive); + } } diff --git a/src/openhuman/voice/postprocess.rs b/src/openhuman/voice/postprocess.rs index 7c73c2c62..bd48a364a 100644 --- a/src/openhuman/voice/postprocess.rs +++ b/src/openhuman/voice/postprocess.rs @@ -112,7 +112,14 @@ pub async fn cleanup_transcription( // Hard timeout — dictation must feel instant. If the LLM doesn't // respond within 3 seconds, fall back to the raw Whisper text. - let inference_fut = service.inference(config, CLEANUP_SYSTEM_PROMPT, &prompt, Some(512), true); + // + // Voice cleanup is a user-arrival path (mic press → STT → cleanup + // shown to user). It bypasses the scheduler_gate permit via + // `inference_interactive` so a long-running memory backfill does + // not push the cleanup past the 3s timeout. Mirrors the autocomplete + // bypass pattern (`inline_complete_interactive`). + let inference_fut = + service.inference_interactive(config, CLEANUP_SYSTEM_PROMPT, &prompt, Some(512), true); let result: Result = match tokio::time::timeout(std::time::Duration::from_secs(3), inference_fut).await { Ok(r) => r,