Bus-backed scheduler gate + bounded manual-override member (#126)

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)]

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);

crates/tinymemory-core/src/scheduler_gate.rs

@@ -70,11 +70,54 @@ pub fn scheduler_gate() -> Option<Arc<dyn SchedulerGate>> {
 }
 
 /// 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<Option<std::time::Instant>> = RwLock::new(None);
+
+fn manual_override_active() -> bool {
+    MANUAL_OVERRIDE_UNTIL
+        .read()
+        .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.
+///
+/// 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) {
+    // 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();
+}
+
 /// The resume handle. When ungated this is a `Notify` nobody ever fires, so a
 /// `select!` on it simply never takes that arm.
 #[must_use]
@@ -95,3 +138,22 @@ pub async fn wait_for_capacity() -> Option<Box<dyn Send>> {
         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();
+    }
+}

crates/tinymemory-module/src/host.rs

@@ -260,6 +260,165 @@ 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<tinymemory_core::scheduler_gate::Policy>,
+    notify: Arc<tokio::sync::Notify>,
+}
+
+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();
+        }
+    }
+
+    /// 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<Self> {
+        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<String>)>("SchedulerPolicy", ())
+                    .await
+            }
+            Err(error) => Err(error),
+        };
+        match reply {
+            Ok((tier, reason)) => {
+                self.store_policy(wire_to_policy(&tier, reason.as_deref()));
+                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(std::sync::PoisonError::into_inner)
+    }
+
+    fn resume_notify(&self) -> Arc<tokio::sync::Notify> {
+        Arc::clone(&self.notify)
+    }
+
+    async fn wait_for_capacity(&self) -> Option<Box<dyn Send>> {
+        // 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,
+        "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,
+            },
+        },
+        // "normal" lands here with every unknown tier, deliberately in one
+        // arm: an unknown tier degrades to the pre-gate behaviour.
+        _ => Policy::Normal,
+    }
+}
+
 #[derive(Debug)]
 pub(crate) struct UnservedSchedulerGate;
 
@@ -309,18 +468,34 @@ 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() {
-    tinymemory_core::scheduler_gate::set_scheduler_gate(Arc::new(UnservedSchedulerGate));
+/// 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<tinybus::Connection>) {
+    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"
     );
 }