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..d21a184b 100644 --- a/crates/tinymemory-core/src/scheduler_gate.rs +++ b/crates/tinymemory-core/src/scheduler_gate.rs @@ -70,11 +70,54 @@ 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) +} + +/// 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> { 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 74130f92..7c76619f 100644 --- a/crates/tinymemory-module/src/host.rs +++ b/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, + notify: Arc, +} + +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 { + 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)) => { + 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 { + 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, + "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) { + 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/host_test.rs b/crates/tinymemory-module/src/host_test.rs index b22bb34e..6b665060 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()); @@ -307,3 +307,102 @@ async fn fire_and_forget_notification_tolerates_an_absent_host() { }); tokio::task::yield_now().await; } + +// ── 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}; + 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 = gate_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 = gate_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(); +} diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 542a9bc7..ae8d76c1 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, @@ -275,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. @@ -405,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 @@ -698,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 97c3036e..9ccbd0ad 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -2217,6 +2217,28 @@ 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 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, 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. + 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. 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(); +} 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]