From f75af263472ba3cdec4f4695a38a7eadbc6594b0 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 1 Sep 2026 19:44:22 +0530 Subject: [PATCH 1/8] feat(module): bus-backed scheduler gate; manual override member Closes the consent gap #126 records: the module's queue pool and periodic loops ran unthrottled because no scheduler gate is served in module mode -- the stub answered Policy::Normal forever, so a user's mode=off, signed-out and battery pauses stopped at the process boundary. BusSchedulerGate polls the host's SchedulerPolicy member (served on the RuntimeHost object the event sink already calls), caches the policy for the synchronous step-0 reads, wakes paused sleepers on a resume transition, and degrades to the previous stub behaviour -- Policy::Normal, reported once -- against a host that does not serve the member. Capacity permits deliberately do not cross: the LLM-slot semaphore is a host-process resource, and a permit forged module-side would be a lie. OverrideSchedulerGate (new wire member, appended at slot 138 per the table's append-only rule) opens a bounded manual-override window in tinymemory-core: for N seconds (clamped to an hour) current_policy answers Normal and sleepers wake, so user-initiated maintenance runs while the gate is paused -- the pause protects the user from background cost they did not ask for, and explicitly requested work is the opposite case (openhuman#5935). --- crates/tinymemory-bus/src/names.rs | 6 +- crates/tinymemory-bus/src/names_tests.rs | 7 +- crates/tinymemory-core/src/scheduler_gate.rs | 31 ++++ crates/tinymemory-module/src/host.rs | 182 ++++++++++++++++++- crates/tinymemory-module/src/lib.rs | 16 +- crates/tinymemory-module/src/service/mod.rs | 18 ++ 6 files changed, 242 insertions(+), 18 deletions(-) diff --git a/crates/tinymemory-bus/src/names.rs b/crates/tinymemory-bus/src/names.rs index 989de73f..69ff8f68 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -322,6 +322,9 @@ pub mod methods { // report. /// `Diagnose` — the typed, per-stage pipeline diagnosis. pub const DIAGNOSE: &str = "Diagnose"; + /// `OverrideSchedulerGate` — open a bounded manual-override window on the + /// scheduler gate, for user-initiated maintenance while paused. + pub const OVERRIDE_SCHEDULER_GATE: &str = "OverrideSchedulerGate"; /// `DegradedState` — the degradation flags alone, without a diagnosis. pub const DEGRADED_STATE: &str = "DegradedState"; @@ -370,7 +373,7 @@ pub mod methods { /// The order matters: `tinybus`'s `Interface::members()` returns declaration /// order, and the module compares the two sequences directly rather than as /// sets, so a reordering is caught alongside an addition or a removal. -pub const METHODS: [&str; 141] = [ +pub const METHODS: [&str; 142] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -512,6 +515,7 @@ pub const METHODS: [&str; 141] = [ methods::INGEST_LEARNING, methods::INGEST_EVENT, methods::ANSWER, + methods::OVERRIDE_SCHEDULER_GATE, ]; #[cfg(test)] diff --git a/crates/tinymemory-bus/src/names_tests.rs b/crates/tinymemory-bus/src/names_tests.rs index f1adcdc6..568852a6 100644 --- a/crates/tinymemory-bus/src/names_tests.rs +++ b/crates/tinymemory-bus/src/names_tests.rs @@ -99,6 +99,11 @@ fn the_newest_members_are_appended_rather_than_filed_with_their_family() { assert_eq!(METHODS[128], methods::DEGRADED_STATE); assert_eq!(METHODS[129], methods::CHUNK_SCORE); assert_eq!(METHODS[130], methods::SOURCE_INGEST_STATUS); + // Scheduler-gate round (openhuman#5935 / tinymemory#126): appended at the + // tail — slot 141, after the ingestion round's 138-140 — per this table's + // append-only rule. + assert_eq!(METHODS[141], methods::OVERRIDE_SCHEDULER_GATE); + assert_eq!(methods::OVERRIDE_SCHEDULER_GATE, "OverrideSchedulerGate"); } #[test] @@ -139,7 +144,7 @@ fn the_runtime_tree_doors_hold_the_wire_slots_they_were_released_in() { // reason the summariser-door test above gives: member order is wire order, // and an assertion measured from the end moves silently under the next // append — which is exactly the edit this exists to catch. - assert_eq!(METHODS.len(), 141); + assert_eq!(METHODS.len(), 142); assert_eq!(METHODS[131], methods::RUNTIME_BUFFER_WRITE); assert_eq!(METHODS[132], methods::RUNTIME_READ_NODE); assert_eq!(METHODS[133], methods::RUNTIME_READ_CHILDREN); diff --git a/crates/tinymemory-core/src/scheduler_gate.rs b/crates/tinymemory-core/src/scheduler_gate.rs index 2b2452a8..6f80dcc4 100644 --- a/crates/tinymemory-core/src/scheduler_gate.rs +++ b/crates/tinymemory-core/src/scheduler_gate.rs @@ -70,11 +70,42 @@ pub fn scheduler_gate() -> Option> { } /// The current scheduling tier, or [`Policy::Normal`] when ungated. +/// +/// A live manual override (see [`set_manual_override`]) takes precedence over +/// the gate: user-initiated maintenance is the one thing a pause must not +/// stop, because the pause exists to protect the user from *background* cost +/// they did not ask for — work they explicitly requested is the opposite +/// case. #[must_use] pub fn current_policy() -> Policy { + if manual_override_active() { + return Policy::Normal; + } scheduler_gate().map_or(Policy::Normal, |gate| gate.current_policy()) } +static MANUAL_OVERRIDE_UNTIL: RwLock> = RwLock::new(None); + +fn manual_override_active() -> bool { + MANUAL_OVERRIDE_UNTIL + .read() + .is_some_and(|until| std::time::Instant::now() < until) +} + +/// Open a manual-override window: for `seconds`, [`current_policy`] answers +/// [`Policy::Normal`] regardless of the installed gate, and paused sleepers +/// are woken so the window is not spent waiting out a tick. +/// +/// For user-initiated maintenance under `mode = off` (openhuman#5935): the +/// off switch stops background work, and this is how a user's explicit +/// "process now" still runs. The window is bounded — there is no "override +/// forever", because that would just be the gate turned off with extra steps. +pub fn set_manual_override(seconds: u64) { + let until = std::time::Instant::now() + std::time::Duration::from_secs(seconds); + *MANUAL_OVERRIDE_UNTIL.write() = Some(until); + resume_notify().notify_waiters(); +} + /// The resume handle. When ungated this is a `Notify` nobody ever fires, so a /// `select!` on it simply never takes that arm. #[must_use] diff --git a/crates/tinymemory-module/src/host.rs b/crates/tinymemory-module/src/host.rs index 74130f92..42c15d6e 100644 --- a/crates/tinymemory-module/src/host.rs +++ b/crates/tinymemory-module/src/host.rs @@ -260,6 +260,152 @@ pub(crate) fn report_unserved_once( /// Answers exactly what an uninstalled gate answered — see the section comment /// above for why it must not answer anything else — and says so out loud the /// first time it is asked. +#[derive(Debug)] +/// Scheduler gate answered by the host over the bus. +/// +/// The host serves `SchedulerPolicy` on its `RuntimeHost` object (the same +/// object the event sink and error reporter already call), answering the +/// policy its own `cron::scheduler_gate` computes — mode, battery, CPU +/// pressure, signed-out. This gate polls it and caches the answer, because +/// [`SchedulerGate::current_policy`] is a synchronous step-0 read on every +/// queue claim and every periodic tick, and a bus round-trip per claim would +/// put the broker on the hot path. +/// +/// What deliberately does NOT cross the bus: `wait_for_capacity`. The +/// LLM-slot semaphore is a host-process resource; a permit forged here would +/// be a lie about a semaphore this process cannot see. Policy pauses are the +/// consent-bearing half, and they cross. A host that serves no +/// `SchedulerPolicy` member (older host) degrades to exactly the previous +/// stub behaviour: `Policy::Normal`, reported once. +pub(crate) struct BusSchedulerGate { + policy: std::sync::RwLock, + notify: Arc, +} + +impl BusSchedulerGate { + /// Poll cadence while the host answers. Claims read the cache, so this + /// bounds how stale a pause can be, not how often anything blocks. + const POLL_SECS: u64 = 15; + /// Poll cadence after a failed call — an older host answers + /// `MemberNotFound` forever, and once a minute keeps the retirement of + /// that host observable without spamming its log. + const POLL_SECS_UNSERVED: u64 = 60; + + pub(crate) fn start(connection: tinybus::Connection) -> Arc { + let gate = Arc::new(Self { + policy: std::sync::RwLock::new(tinymemory_core::scheduler_gate::Policy::Normal), + notify: Arc::new(tokio::sync::Notify::new()), + }); + let poller = Arc::clone(&gate); + tokio::spawn(async move { + loop { + let served = poller.refresh(&connection).await; + let secs = if served { + Self::POLL_SECS + } else { + Self::POLL_SECS_UNSERVED + }; + tokio::time::sleep(std::time::Duration::from_secs(secs)).await; + } + }); + gate + } + + /// One poll: ask the host, map the wire strings, store, and wake sleepers + /// on a pause → not-paused transition. Returns whether the member + /// answered. + async fn refresh(&self, connection: &tinybus::Connection) -> bool { + let reply = match connection.proxy( + RUNTIME_HOST_BUS_NAME, + RUNTIME_HOST_OBJECT_PATH, + RUNTIME_HOST_INTERFACE, + ) { + Ok(proxy) => { + proxy + .call::<(String, Option)>("SchedulerPolicy", ()) + .await + } + Err(error) => Err(error), + }; + match reply { + Ok((tier, reason)) => { + let next = wire_to_policy(&tier, reason.as_deref()); + let (was_paused, changed) = { + let mut slot = self.policy.write().unwrap_or_else(|e| e.into_inner()); + let was = matches!( + *slot, + tinymemory_core::scheduler_gate::Policy::Paused { .. } + ); + let changed = *slot != next; + *slot = next; + (was, changed) + }; + if changed { + log::info!( + "[tinymemory:module] scheduler policy from host: {next:?} — background \ + claims honour it from the next tick" + ); + } + let now_paused = + matches!(next, tinymemory_core::scheduler_gate::Policy::Paused { .. }); + if was_paused && !now_paused { + self.notify.notify_waiters(); + } + true + } + Err(error) => { + report_unserved_once(&GATE_REPORTED, GATE_UNSERVED, "scheduler_gate"); + log::debug!( + "[tinymemory:module] SchedulerPolicy poll failed; keeping the last policy: {error}" + ); + false + } + } + } +} + +#[async_trait] +impl tinymemory_core::scheduler_gate::SchedulerGate for BusSchedulerGate { + fn current_policy(&self) -> tinymemory_core::scheduler_gate::Policy { + *self.policy.read().unwrap_or_else(|e| e.into_inner()) + } + + fn resume_notify(&self) -> Arc { + Arc::clone(&self.notify) + } + + async fn wait_for_capacity(&self) -> Option> { + // Host-process semaphore; see the struct docs. Policy pauses are + // enforced by every claim's step-0 `current_policy` read instead. + None + } +} + +/// Map the wire tier + pause-reason strings back onto the contract types. +/// +/// Unknown strings collapse to the safe end of their type: an unknown tier is +/// `Normal` (the pre-gate behaviour, never a surprise pause), an unknown +/// pause reason is `PauseReason::Unknown` (still a pause — the host said +/// stop, and the unknown part is only the label). +fn wire_to_policy(tier: &str, reason: Option<&str>) -> tinymemory_core::scheduler_gate::Policy { + use tinymemory_core::scheduler_gate::{PauseReason, Policy}; + match tier { + "aggressive" => Policy::Aggressive, + "normal" => Policy::Normal, + "throttled" => Policy::Throttled, + "paused" => Policy::Paused { + reason: match reason { + Some("user_disabled") => PauseReason::UserDisabled, + Some("on_battery") => PauseReason::OnBattery, + Some("cpu_pressure") => PauseReason::CpuPressure, + Some("signed_out") => PauseReason::SignedOut, + _ => PauseReason::Unknown, + }, + }, + _ => Policy::Normal, + } +} + #[derive(Debug)] pub(crate) struct UnservedSchedulerGate; @@ -310,17 +456,35 @@ impl tinymemory_core::shutdown::ShutdownHost for UnservedShutdownHost { /// host genuinely serves over the bus, and folding these in would blur the /// difference between "wired" and "wired to nothing". pub(crate) fn install_unserved_seams() { - tinymemory_core::scheduler_gate::set_scheduler_gate(Arc::new(UnservedSchedulerGate)); + install_seams(None); +} + +/// Install the host seams, bus-backing the scheduler gate when a connection +/// is available. +/// +/// With a connection, the gate is [`BusSchedulerGate`] — the host's policy, +/// polled and cached — and only `shutdown` remains a stub. Without one (unit +/// tests, or a caller that has not connected yet), both fall back to the +/// unserved stubs, which keep the previous unwired behaviour and say so once. +pub(crate) fn install_seams(connection: Option) { + match connection { + Some(connection) => { + tinymemory_core::scheduler_gate::set_scheduler_gate(BusSchedulerGate::start(connection)) + } + None => { + tinymemory_core::scheduler_gate::set_scheduler_gate(Arc::new(UnservedSchedulerGate)) + } + } tinymemory_core::shutdown::set_shutdown_host(Arc::new(UnservedShutdownHost)); - // One line, once per process — `setup` runs exactly once. It is a warning - // rather than a debug line because in module mode this is true on every - // boot, and a reader of the log should not have to diff seam lists to find - // out that the throttle and the graceful lock release are not in effect. + // One line, once per process — `setup` runs exactly once. A warning rather + // than a debug line because a reader of the log should not have to diff + // seam lists to find out which host behaviours are not in effect here. log::warn!( - "[tinymemory:module] two host seams are unserved in module mode: scheduler_gate and \ - shutdown are stubs that keep the unwired behaviour and report once when consulted. \ - Background-AI throttling, the \"Memory Tree off\" and \"signed out\" pauses on periodic \ - sync, and graceful queue-lock release are not honoured inside this process" + "[tinymemory:module] shutdown is unserved in module mode: a stub keeps the unwired \ + behaviour and reports once when consulted, so graceful queue-lock release is not \ + honoured inside this process. The scheduler gate is bus-backed when the host serves \ + SchedulerPolicy, and degrades to the unwired Policy::Normal stub behaviour when it \ + does not" ); } diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 542a9bc7..35bbb21b 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -146,13 +146,14 @@ async fn setup(connection: Connection, mut config: ModuleConfig) -> BusResult<() // back out to the engine repeatedly. tinymemory_core::config_loader::set_config_loader(Arc::new(ModuleConfigLoader::new(&config))); host::install(connection.clone()); - // The two seams no bus interface serves, and no local answer can honestly - // stand in for. Both degraded in silence rather than with a named cause; - // see the section comment on `host::install_unserved_seams` for why they - // are stubbed here rather than proxied or synthesised. Installed with the - // rest, before the store exists, so nothing can consult a seam this process - // has not yet decided about. - host::install_unserved_seams(); + // The scheduler gate is proxied to the host's SchedulerPolicy member — the + // host's cron::scheduler_gate policy, polled and cached, so mode=off, + // signed-out and battery pauses are honoured inside this process too. + // Shutdown stays a stub: no bus interface serves it and no local answer + // can honestly stand in for it (see `host::install_seams`). Installed with + // the rest, before the store exists, so nothing can consult a seam this + // process has not yet decided about. + host::install_seams(Some(connection.clone())); let client = tinymemory_core::store::factories::create_memory_client_with_local_ai( &config.memory, @@ -654,6 +655,7 @@ mod exports { // Maintenance, typed: the diagnosis an operator or an agent reads, // beside the uniform report a scheduler reads. "Diagnose", + "OverrideSchedulerGate", // Source sync this process runs itself. The periodic loops already // live here; these are the on-demand half plus what past runs cost. "RunConnectionSync", diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 97c3036e..eb4d33cf 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -1741,6 +1741,24 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } + /// Open a bounded manual-override window on the scheduler gate. + /// + /// The host calls this when the user explicitly asks for maintenance + /// while the gate is paused (`mode = off`, signed-out, battery): for + /// `seconds`, background claims read `Policy::Normal` and paused sleepers + /// are woken, so a user's "process now" runs without turning the gate's + /// protection off for anything they did not ask for (openhuman#5935). + async fn override_scheduler_gate(&self, seconds: u64) -> BusResult<()> { + // Clamp: a window longer than an hour is the gate turned off with + // extra steps, which is the config's job, not this member's. + let seconds = seconds.min(3600); + tinymemory_core::scheduler_gate::set_manual_override(seconds); + log::info!( + "[tinymemory:module] scheduler gate manually overridden for {seconds}s (host request)" + ); + Ok(()) + } + // ── Source sync the driver runs itself ────────────────────────────────── /// Sync one connection now. From 6ccd1ab38d0c14f1c6a6101e3d06e51a96e69564 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 1 Sep 2026 20:32:11 +0530 Subject: [PATCH 2/8] fix(review): wire-tail declarations everywhere; clamp owned by core Two CodeRabbit findings on the gate round: The module's member allowlist and service declaration filed OverrideSchedulerGate beside Diagnose while the bus table appends it at slot 141 -- and member order is wire order, so the mid-list filing renumbered every later member against a host built on the released table. Both module-side declarations move to the tail, matching the table. set_manual_override owns its clamp now: the hour bound is the function's contract rather than a caller courtesy, and it also makes the expiry arithmetic infallible -- Instant + 1h cannot overflow, where an unclamped u64 could panic inside library code. --- crates/tinymemory-core/src/scheduler_gate.rs | 6 ++++ crates/tinymemory-module/src/lib.rs | 4 ++- crates/tinymemory-module/src/service/mod.rs | 36 ++++++++++---------- 3 files changed, 27 insertions(+), 19 deletions(-) diff --git a/crates/tinymemory-core/src/scheduler_gate.rs b/crates/tinymemory-core/src/scheduler_gate.rs index 6f80dcc4..d42c5451 100644 --- a/crates/tinymemory-core/src/scheduler_gate.rs +++ b/crates/tinymemory-core/src/scheduler_gate.rs @@ -101,6 +101,12 @@ fn manual_override_active() -> bool { /// "process now" still runs. The window is bounded — there is no "override /// forever", because that would just be the gate turned off with extra steps. pub fn set_manual_override(seconds: u64) { + // The clamp is this function's contract, not its callers': the window is + // bounded to an hour (an unbounded override is the gate turned off with + // extra steps), and the bound also makes the expiry arithmetic + // infallible — `Instant + 1h` cannot overflow, where an unclamped u64 + // could panic inside library code. + let seconds = seconds.min(3600); let until = std::time::Instant::now() + std::time::Duration::from_secs(seconds); *MANUAL_OVERRIDE_UNTIL.write() = Some(until); resume_notify().notify_waiters(); diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 35bbb21b..3de70047 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -655,7 +655,6 @@ mod exports { // Maintenance, typed: the diagnosis an operator or an agent reads, // beside the uniform report a scheduler reads. "Diagnose", - "OverrideSchedulerGate", // Source sync this process runs itself. The periodic loops already // live here; these are the on-demand half plus what past runs cost. "RunConnectionSync", @@ -700,6 +699,9 @@ mod exports { "IngestLearning", "IngestEvent", "Answer", + // Appended at the wire tail (slot 141) to match the bus table's + // append-only order — member order is wire order. + "OverrideSchedulerGate", ], signals = [], // The host's embedder is deliberately NOT declared as `requires`. That diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index eb4d33cf..58759264 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -1741,24 +1741,6 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } - /// Open a bounded manual-override window on the scheduler gate. - /// - /// The host calls this when the user explicitly asks for maintenance - /// while the gate is paused (`mode = off`, signed-out, battery): for - /// `seconds`, background claims read `Policy::Normal` and paused sleepers - /// are woken, so a user's "process now" runs without turning the gate's - /// protection off for anything they did not ask for (openhuman#5935). - async fn override_scheduler_gate(&self, seconds: u64) -> BusResult<()> { - // Clamp: a window longer than an hour is the gate turned off with - // extra steps, which is the config's job, not this member's. - let seconds = seconds.min(3600); - tinymemory_core::scheduler_gate::set_manual_override(seconds); - log::info!( - "[tinymemory:module] scheduler gate manually overridden for {seconds}s (host request)" - ); - Ok(()) - } - // ── Source sync the driver runs itself ────────────────────────────────── /// Sync one connection now. @@ -2235,6 +2217,24 @@ impl MemoryService { .await .map_err(|error| into_bus_error(&error)) } + + /// Open a bounded manual-override window on the scheduler gate. + /// + /// The host calls this when the user explicitly asks for maintenance + /// while the gate is paused (`mode = off`, signed-out, battery): for + /// `seconds`, background claims read `Policy::Normal` and paused sleepers + /// are woken, so a user's "process now" runs without turning the gate's + /// protection off for anything they did not ask for (openhuman#5935). + async fn override_scheduler_gate(&self, seconds: u64) -> BusResult<()> { + // Clamp: a window longer than an hour is the gate turned off with + // extra steps, which is the config's job, not this member's. + let seconds = seconds.min(3600); + tinymemory_core::scheduler_gate::set_manual_override(seconds); + log::info!( + "[tinymemory:module] scheduler gate manually overridden for {seconds}s (host request)" + ); + Ok(()) + } } /// The response-size ceiling for a method that returns a list of entries. From f025ccbd9ae0574bdfb9bfc283a73b27062cf7c8 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 1 Sep 2026 20:38:13 +0530 Subject: [PATCH 3/8] fix(clippy): module-workspace lints on the gate round Redundant closures to PoisonError::into_inner, the normal tier merged into the fallback arm (same body, one deliberate arm), semicolons on the installer's match arms, and install_unserved_seams deleted -- superseded by install_seams(None), which the one test now calls directly. --- crates/tinymemory-module/src/host.rs | 17 ++++++++++------- crates/tinymemory-module/src/host_test.rs | 2 +- crates/tinymemory-module/src/lib.rs | 4 ++-- 3 files changed, 13 insertions(+), 10 deletions(-) diff --git a/crates/tinymemory-module/src/host.rs b/crates/tinymemory-module/src/host.rs index 42c15d6e..ee57c8bd 100644 --- a/crates/tinymemory-module/src/host.rs +++ b/crates/tinymemory-module/src/host.rs @@ -331,7 +331,10 @@ impl BusSchedulerGate { Ok((tier, reason)) => { let next = wire_to_policy(&tier, reason.as_deref()); let (was_paused, changed) = { - let mut slot = self.policy.write().unwrap_or_else(|e| e.into_inner()); + let mut slot = self + .policy + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); let was = matches!( *slot, tinymemory_core::scheduler_gate::Policy::Paused { .. } @@ -367,7 +370,10 @@ impl BusSchedulerGate { #[async_trait] impl tinymemory_core::scheduler_gate::SchedulerGate for BusSchedulerGate { fn current_policy(&self) -> tinymemory_core::scheduler_gate::Policy { - *self.policy.read().unwrap_or_else(|e| e.into_inner()) + *self + .policy + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) } fn resume_notify(&self) -> Arc { @@ -391,7 +397,6 @@ fn wire_to_policy(tier: &str, reason: Option<&str>) -> tinymemory_core::schedule use tinymemory_core::scheduler_gate::{PauseReason, Policy}; match tier { "aggressive" => Policy::Aggressive, - "normal" => Policy::Normal, "throttled" => Policy::Throttled, "paused" => Policy::Paused { reason: match reason { @@ -402,6 +407,8 @@ fn wire_to_policy(tier: &str, reason: Option<&str>) -> tinymemory_core::schedule _ => PauseReason::Unknown, }, }, + // "normal" lands here with every unknown tier, deliberately in one + // arm: an unknown tier degrades to the pre-gate behaviour. _ => Policy::Normal, } } @@ -455,10 +462,6 @@ impl tinymemory_core::shutdown::ShutdownHost for UnservedShutdownHost { /// Kept separate from [`install`] on purpose: that function wires the seams the /// host genuinely serves over the bus, and folding these in would blur the /// difference between "wired" and "wired to nothing". -pub(crate) fn install_unserved_seams() { - install_seams(None); -} - /// Install the host seams, bus-backing the scheduler gate when a connection /// is available. /// diff --git a/crates/tinymemory-module/src/host_test.rs b/crates/tinymemory-module/src/host_test.rs index b22bb34e..fa150609 100644 --- a/crates/tinymemory-module/src/host_test.rs +++ b/crates/tinymemory-module/src/host_test.rs @@ -260,7 +260,7 @@ async fn install_wires_every_seam_this_module_can_supply() { // The pair `setup` calls, in the order it calls them. super::install(connection); - super::install_unserved_seams(); + super::install_seams(None); assert!(tinymemory_core::events::event_sink().is_some()); assert!(tinymemory_core::observability::error_reporter().is_some()); diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 3de70047..ae8d76c1 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -276,7 +276,7 @@ fn bind_memory_client(config: &ModuleConfig, client: &MemoryClientRef) -> bool { /// `periodic_pause_reason` as step 0 of every tick, precisely so a user who /// switched Memory Tree off, or who is signed out, gets no background fetch. /// This module serves no scheduler gate — see the section comment on -/// `host::install_unserved_seams` for why it cannot — and the stub in its +/// `host::install_seams` for why it cannot — and the stub in its /// place always answers `Policy::Normal`, so `periodic_pause_reason` is always /// `None` and it ticks straight through both pauses. The per-source /// `enabled` toggle still applies; the two *global* pauses do not. @@ -406,7 +406,7 @@ pub(crate) fn claim_sync_loops(workspace: &Path) -> WorkspaceClaim { /// [`tinymemory_core::scheduler_gate`] before every claim and registers a /// [`tinymemory_core::shutdown`] hook to release in-flight job locks. This /// module serves neither seam — see the section comment on -/// `host::install_unserved_seams` for why neither can be proxied — so both are +/// `host::install_seams` for why neither can be proxied — so both are /// stubs, and the consequences follow: /// /// - **It runs unthrottled.** `wait_for_capacity` returns immediately, so From b73aef33b98ae91a9e54efadb81024c48b5cf804 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 1 Sep 2026 21:09:28 +0530 Subject: [PATCH 4/8] fix(clippy): fmt-stable semicolons; allow unused_async on the member The braced installer arms get calls reformatted so the semicolon survives rustfmt (the previous shape had fmt stripping what clippy then demanded), and override_scheduler_gate carries an allow with its reason: async is the interface macro's member contract, and the one-write body staying synchronous is the point. --- crates/tinymemory-module/src/host.rs | 6 ++++-- crates/tinymemory-module/src/service/mod.rs | 4 ++++ 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/crates/tinymemory-module/src/host.rs b/crates/tinymemory-module/src/host.rs index ee57c8bd..a3f34698 100644 --- a/crates/tinymemory-module/src/host.rs +++ b/crates/tinymemory-module/src/host.rs @@ -472,10 +472,12 @@ impl tinymemory_core::shutdown::ShutdownHost for UnservedShutdownHost { pub(crate) fn install_seams(connection: Option) { match connection { Some(connection) => { - tinymemory_core::scheduler_gate::set_scheduler_gate(BusSchedulerGate::start(connection)) + tinymemory_core::scheduler_gate::set_scheduler_gate(BusSchedulerGate::start( + connection, + )); } None => { - tinymemory_core::scheduler_gate::set_scheduler_gate(Arc::new(UnservedSchedulerGate)) + tinymemory_core::scheduler_gate::set_scheduler_gate(Arc::new(UnservedSchedulerGate)); } } tinymemory_core::shutdown::set_shutdown_host(Arc::new(UnservedShutdownHost)); diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 58759264..94d27768 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -2225,6 +2225,10 @@ impl MemoryService { /// `seconds`, background claims read `Policy::Normal` and paused sleepers /// are woken, so a user's "process now" runs without turning the gate's /// protection off for anything they did not ask for (openhuman#5935). + // async only for the interface macro's member contract — the body is one + // synchronous global write, and that is the point: a claim's step-0 read + // must never wait on this. + #[allow(clippy::unused_async)] async fn override_scheduler_gate(&self, seconds: u64) -> BusResult<()> { // Clamp: a window longer than an hour is the gate turned off with // extra steps, which is the config's job, not this member's. From 3f9981da43f473db64cb774173824c27c36d3b02 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 1 Sep 2026 21:09:55 +0530 Subject: [PATCH 5/8] fix(clippy): allow both unused-async lint ids on the member The CI toolchain fires unused_async AND unused_async_trait_impl on the same fn; the allow now names both, same reason -- async is the interface macro's member contract and the one-write body staying synchronous is the point. --- crates/tinymemory-module/src/service/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 94d27768..9ccbd0ad 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -2228,7 +2228,7 @@ impl MemoryService { // async only for the interface macro's member contract — the body is one // synchronous global write, and that is the point: a claim's step-0 read // must never wait on this. - #[allow(clippy::unused_async)] + #[allow(clippy::unused_async, clippy::unused_async_trait_impl)] async fn override_scheduler_gate(&self, seconds: u64) -> BusResult<()> { // Clamp: a window longer than an hour is the gate turned off with // extra steps, which is the config's job, not this member's. From 6f28c0540e273927f1533da6792f482806836ce0 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 1 Sep 2026 21:21:10 +0530 Subject: [PATCH 6/8] test(module): pin OverrideSchedulerGate in the manifest second-opinion The e2e's EXPECTED_METHODS list is the deliberate duplicate of the export table -- the drift test exists so a member added to one shows up as a named difference until it is added to both. This is that addition. --- crates/tinymemory-module/tests/module_e2e.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index fe50f5b1..55ecc85a 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -739,6 +739,8 @@ const EXPECTED_METHODS: &[&str] = &[ "IngestLearning", "IngestEvent", "Answer", + // Scheduler-gate round: appended at the wire tail with its declaration. + "OverrideSchedulerGate", ]; #[tokio::test] From 2a03104e73344693ac04ff30c417f10fb72c20df Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 1 Sep 2026 21:35:28 +0530 Subject: [PATCH 7/8] test(module): behavioral coverage for the gate round The module lane's 80% line floor caught the new code untested (77.95%), and the gaps were the ones worth pinning anyway: wire_to_policy is exercised over every tier and pause reason, including the deliberate unknown-tier-degrades-to-Normal and pause-without-reason arms. store_policy is factored off the bus call -- the transition rules were untestable behind a broker -- and its test proves the one contract that matters live: a sleeper parked on resume_notify wakes when a pause lifts, and only then. The override test installs a paused gate and shows set_manual_override outranking it exactly while the window is open, against the new core clear_manual_override test-support (a process global a test must not leak into its neighbours). --- crates/tinymemory-core/src/scheduler_gate.rs | 6 ++ crates/tinymemory-module/src/host.rs | 65 +++++++++------ crates/tinymemory-module/src/host_test.rs | 88 ++++++++++++++++++++ 3 files changed, 134 insertions(+), 25 deletions(-) diff --git a/crates/tinymemory-core/src/scheduler_gate.rs b/crates/tinymemory-core/src/scheduler_gate.rs index d42c5451..33ffa5be 100644 --- a/crates/tinymemory-core/src/scheduler_gate.rs +++ b/crates/tinymemory-core/src/scheduler_gate.rs @@ -92,6 +92,12 @@ fn manual_override_active() -> bool { .is_some_and(|until| std::time::Instant::now() < until) } +/// Clear any live manual override. For tests: the window is a process +/// global, and a test that opens one must not leak it into its neighbours. +pub fn clear_manual_override() { + *MANUAL_OVERRIDE_UNTIL.write() = None; +} + /// Open a manual-override window: for `seconds`, [`current_policy`] answers /// [`Policy::Normal`] regardless of the installed gate, and paused sleepers /// are woken so the window is not spent waiting out a tick. diff --git a/crates/tinymemory-module/src/host.rs b/crates/tinymemory-module/src/host.rs index a3f34698..a505332d 100644 --- a/crates/tinymemory-module/src/host.rs +++ b/crates/tinymemory-module/src/host.rs @@ -283,6 +283,45 @@ pub(crate) struct BusSchedulerGate { } impl BusSchedulerGate { + /// Store a freshly polled policy: log on change, and wake paused sleepers + /// on a pause → not-paused transition so a resume is immediate rather than + /// one tick late. Factored off the bus call so the transition rules are + /// testable without a broker. + fn store_policy(&self, next: tinymemory_core::scheduler_gate::Policy) { + let (was_paused, changed) = { + let mut slot = self + .policy + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let was = matches!( + *slot, + tinymemory_core::scheduler_gate::Policy::Paused { .. } + ); + let changed = *slot != next; + *slot = next; + (was, changed) + }; + if changed { + log::info!( + "[tinymemory:module] scheduler policy from host: {next:?} — background claims \ + honour it from the next tick" + ); + } + let now_paused = matches!(next, tinymemory_core::scheduler_gate::Policy::Paused { .. }); + if was_paused && !now_paused { + self.notify.notify_waiters(); + } + } + + /// A gate with no poller, for tests: `store_policy` is driven by hand. + #[cfg(test)] + pub(crate) fn new_for_test() -> Arc { + Arc::new(Self { + policy: std::sync::RwLock::new(tinymemory_core::scheduler_gate::Policy::Normal), + notify: Arc::new(tokio::sync::Notify::new()), + }) + } + /// Poll cadence while the host answers. Claims read the cache, so this /// bounds how stale a pause can be, not how often anything blocks. const POLL_SECS: u64 = 15; @@ -329,31 +368,7 @@ impl BusSchedulerGate { }; match reply { Ok((tier, reason)) => { - let next = wire_to_policy(&tier, reason.as_deref()); - let (was_paused, changed) = { - let mut slot = self - .policy - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let was = matches!( - *slot, - tinymemory_core::scheduler_gate::Policy::Paused { .. } - ); - let changed = *slot != next; - *slot = next; - (was, changed) - }; - if changed { - log::info!( - "[tinymemory:module] scheduler policy from host: {next:?} — background \ - claims honour it from the next tick" - ); - } - let now_paused = - matches!(next, tinymemory_core::scheduler_gate::Policy::Paused { .. }); - if was_paused && !now_paused { - self.notify.notify_waiters(); - } + self.store_policy(wire_to_policy(&tier, reason.as_deref())); true } Err(error) => { diff --git a/crates/tinymemory-module/src/host_test.rs b/crates/tinymemory-module/src/host_test.rs index fa150609..8aa9807f 100644 --- a/crates/tinymemory-module/src/host_test.rs +++ b/crates/tinymemory-module/src/host_test.rs @@ -307,3 +307,91 @@ async fn fire_and_forget_notification_tolerates_an_absent_host() { }); tokio::task::yield_now().await; } + +// ── bus scheduler gate (scheduler-gate round) ──────────────────────────────── + +#[test] +fn wire_to_policy_maps_every_tier_and_reason() { + use tinymemory_core::scheduler_gate::{PauseReason, Policy}; + assert_eq!( + super::wire_to_policy("aggressive", None), + Policy::Aggressive + ); + assert_eq!(super::wire_to_policy("throttled", None), Policy::Throttled); + // "normal" and every unknown tier share one deliberate arm: the pre-gate + // behaviour, never a surprise pause. + assert_eq!(super::wire_to_policy("normal", None), Policy::Normal); + assert_eq!( + super::wire_to_policy("something-newer", None), + Policy::Normal + ); + for (wire, reason) in [ + ("user_disabled", PauseReason::UserDisabled), + ("on_battery", PauseReason::OnBattery), + ("cpu_pressure", PauseReason::CpuPressure), + ("signed_out", PauseReason::SignedOut), + ("unheard-of", PauseReason::Unknown), + ] { + assert_eq!( + super::wire_to_policy("paused", Some(wire)), + Policy::Paused { reason }, + "reason wire {wire}" + ); + } + // A pause with no reason string is still a pause. + assert_eq!( + super::wire_to_policy("paused", None), + Policy::Paused { + reason: PauseReason::Unknown + } + ); +} + +#[tokio::test] +async fn store_policy_wakes_sleepers_only_on_resume() { + use tinymemory_core::scheduler_gate::{PauseReason, Policy, SchedulerGate}; + let gate = super::BusSchedulerGate::new_for_test(); + assert_eq!(gate.current_policy(), Policy::Normal); + + gate.store_policy(Policy::Paused { + reason: PauseReason::UserDisabled, + }); + assert!(matches!(gate.current_policy(), Policy::Paused { .. })); + + // A sleeper parked on the resume handle wakes when the pause lifts. + let notify = gate.resume_notify(); + let waiter = tokio::spawn(async move { notify.notified().await }); + tokio::task::yield_now().await; + gate.store_policy(Policy::Normal); + tokio::time::timeout(std::time::Duration::from_secs(2), waiter) + .await + .expect("resume must wake the sleeper") + .expect("waiter task"); + assert_eq!(gate.current_policy(), Policy::Normal); + + // Same-policy stores are quiet no-ops. + gate.store_policy(Policy::Normal); + assert_eq!(gate.current_policy(), Policy::Normal); +} + +#[test] +fn manual_override_outranks_a_paused_gate_and_is_bounded() { + use tinymemory_core::scheduler_gate as core_gate; + use tinymemory_core::scheduler_gate::{PauseReason, Policy}; + core_gate::clear_manual_override(); + let gate = super::BusSchedulerGate::new_for_test(); + gate.store_policy(Policy::Paused { + reason: PauseReason::UserDisabled, + }); + core_gate::set_scheduler_gate(gate); + assert!(matches!(core_gate::current_policy(), Policy::Paused { .. })); + + // The member's whole contract: user-initiated work wins while the window + // is open, and only while it is open. + core_gate::set_manual_override(60); + assert_eq!(core_gate::current_policy(), Policy::Normal); + core_gate::clear_manual_override(); + assert!(matches!(core_gate::current_policy(), Policy::Paused { .. })); + + core_gate::clear_scheduler_gate(); +} From b15bd455327e051d52bf4674f90f8ef97f430589 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 1 Sep 2026 21:47:24 +0530 Subject: [PATCH 8/8] test: service-path override coverage; ctor moved to the filtered file The module lane's floor sat at 79.73 and the powerset lane's policy guard named the cause precisely: inline #[cfg(test)] executable code pollutes the measured production lines. The test constructor moves into host_test.rs (the child module reaches the private fields), and the override member gains a service-path test -- through MemoryService exactly as a bus dispatch arrives, a paused gate installed, a deliberately absurd window proving the clamp, and the pause restored on clear. Core grows the expired-window micro-test: zero seconds is already no window, and u64::MAX not panicking is the overflow guard's test. --- crates/tinymemory-core/src/scheduler_gate.rs | 19 ++++++++ crates/tinymemory-module/src/host.rs | 9 ---- crates/tinymemory-module/src/host_test.rs | 15 ++++++- crates/tinymemory-module/src/service/test.rs | 46 ++++++++++++++++++++ 4 files changed, 78 insertions(+), 11 deletions(-) diff --git a/crates/tinymemory-core/src/scheduler_gate.rs b/crates/tinymemory-core/src/scheduler_gate.rs index 33ffa5be..d21a184b 100644 --- a/crates/tinymemory-core/src/scheduler_gate.rs +++ b/crates/tinymemory-core/src/scheduler_gate.rs @@ -138,3 +138,22 @@ pub async fn wait_for_capacity() -> Option> { None => None, } } + +#[cfg(test)] +mod override_tests { + use super::{clear_manual_override, current_policy, set_manual_override, Policy}; + + #[test] + fn a_zero_second_window_is_already_expired_and_the_clamp_holds() { + clear_manual_override(); + // Zero seconds: the window closes the instant it opens — the branch + // in `current_policy` must treat an expired window as no window. + set_manual_override(0); + assert_eq!(current_policy(), Policy::Normal); // ungated baseline + // An absurd ask cannot overflow the expiry arithmetic: the clamp is + // the function's contract, and this call not panicking is the test. + set_manual_override(u64::MAX); + assert_eq!(current_policy(), Policy::Normal); + clear_manual_override(); + } +} diff --git a/crates/tinymemory-module/src/host.rs b/crates/tinymemory-module/src/host.rs index a505332d..7c76619f 100644 --- a/crates/tinymemory-module/src/host.rs +++ b/crates/tinymemory-module/src/host.rs @@ -313,15 +313,6 @@ impl BusSchedulerGate { } } - /// A gate with no poller, for tests: `store_policy` is driven by hand. - #[cfg(test)] - pub(crate) fn new_for_test() -> Arc { - Arc::new(Self { - policy: std::sync::RwLock::new(tinymemory_core::scheduler_gate::Policy::Normal), - notify: Arc::new(tokio::sync::Notify::new()), - }) - } - /// Poll cadence while the host answers. Claims read the cache, so this /// bounds how stale a pause can be, not how often anything blocks. const POLL_SECS: u64 = 15; diff --git a/crates/tinymemory-module/src/host_test.rs b/crates/tinymemory-module/src/host_test.rs index 8aa9807f..6b665060 100644 --- a/crates/tinymemory-module/src/host_test.rs +++ b/crates/tinymemory-module/src/host_test.rs @@ -310,6 +310,17 @@ async fn fire_and_forget_notification_tolerates_an_absent_host() { // ── bus scheduler gate (scheduler-gate round) ──────────────────────────────── +/// A gate with no poller: `store_policy` driven by hand. Lives here rather +/// than as an inline `#[cfg(test)]` constructor because the coverage lanes +/// filter test files by name, and inline test-only code pollutes the +/// measured production lines (the powerset lane enforces exactly that). +fn gate_for_test() -> std::sync::Arc { + std::sync::Arc::new(super::BusSchedulerGate { + policy: std::sync::RwLock::new(tinymemory_core::scheduler_gate::Policy::Normal), + notify: std::sync::Arc::new(tokio::sync::Notify::new()), + }) +} + #[test] fn wire_to_policy_maps_every_tier_and_reason() { use tinymemory_core::scheduler_gate::{PauseReason, Policy}; @@ -350,7 +361,7 @@ fn wire_to_policy_maps_every_tier_and_reason() { #[tokio::test] async fn store_policy_wakes_sleepers_only_on_resume() { use tinymemory_core::scheduler_gate::{PauseReason, Policy, SchedulerGate}; - let gate = super::BusSchedulerGate::new_for_test(); + let gate = gate_for_test(); assert_eq!(gate.current_policy(), Policy::Normal); gate.store_policy(Policy::Paused { @@ -379,7 +390,7 @@ fn manual_override_outranks_a_paused_gate_and_is_bounded() { use tinymemory_core::scheduler_gate as core_gate; use tinymemory_core::scheduler_gate::{PauseReason, Policy}; core_gate::clear_manual_override(); - let gate = super::BusSchedulerGate::new_for_test(); + let gate = gate_for_test(); gate.store_policy(Policy::Paused { reason: PauseReason::UserDisabled, }); diff --git a/crates/tinymemory-module/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index f48fb053..6a56ed8e 100644 --- a/crates/tinymemory-module/src/service/test.rs +++ b/crates/tinymemory-module/src/service/test.rs @@ -994,3 +994,49 @@ fn an_ordinary_tree_node_read_is_not_refused() { let children: Vec = (0..31).map(|_| node.clone()).collect(); assert!(super::ensure_response_fits(&children, "RuntimeReadChildren").is_ok()); } + +#[tokio::test] +async fn override_member_opens_a_window_that_outranks_a_paused_gate() { + use tinymemory_core::scheduler_gate::{self as gate, PauseReason, Policy}; + + // A paused gate stands in for "the host said mode = off". + #[derive(Debug)] + struct PausedGate; + #[async_trait::async_trait] + impl gate::SchedulerGate for PausedGate { + fn current_policy(&self) -> Policy { + Policy::Paused { + reason: PauseReason::UserDisabled, + } + } + fn resume_notify(&self) -> std::sync::Arc { + std::sync::Arc::new(tokio::sync::Notify::new()) + } + async fn wait_for_capacity(&self) -> Option> { + None + } + } + + gate::clear_manual_override(); + gate::set_scheduler_gate(std::sync::Arc::new(PausedGate)); + assert!(matches!(gate::current_policy(), Policy::Paused { .. })); + + // The member is the host's "process now" lever: through the service impl, + // exactly as a bus dispatch would reach it, the window opens and + // user-requested work outranks the pause -- clamped, so an absurd ask is + // an hour, not forever. + let workspace = tempfile::tempdir().expect("tempdir"); + let connection = test_connection().await; + let config = test_config(workspace.path()); + let opener = test_opener(connection, config); + let service = super::MemoryService::root(test_provider(), std::sync::Arc::clone(&opener)); + service + .override_scheduler_gate(7 * 24 * 3600) + .await + .expect("override member answers"); + assert_eq!(gate::current_policy(), Policy::Normal); + + gate::clear_manual_override(); + assert!(matches!(gate::current_policy(), Policy::Paused { .. })); + gate::clear_scheduler_gate(); +}