feat(storage): background work visits every agent's storage scope

crates/openhuman-cli/Cargo.toml

@@ -104,6 +104,12 @@ path = "../../tests/storage_flows_e2e.rs"
 name = "storage_secrets_e2e"
 path = "../../tests/storage_secrets_e2e.rs"
 
+[[test]]
+# Its own binary: it boots a core and installs a storage backend into the
+# process-wide slot.
+name = "storage_scope_e2e"
+path = "../../tests/storage_scope_e2e.rs"
+
 [[test]]
 name = "embedded_server_shutdown_e2e"
 path = "../../tests/embedded_server_shutdown_e2e.rs"

crates/openhuman-core/src/agent/tinyagents/reaper.rs

@@ -17,8 +17,16 @@ use tinyagents_session::transcript::import::ops::open_session_stores;
 /// Skipped on a storage backend other processes may share (MongoDB): there a
 /// non-terminal status can belong to a run another replica is still driving,
 /// and cancelling it would hide that run from active and late-attach status.
+///
+/// With a storage backend every agent keeps its own status store, so the
+/// sweep visits `local` and then each known agent (`crate::storage::agents`).
 pub(crate) async fn reap_orphaned_runs(workspace: &Path) -> usize {
-    reap_unless_shared(workspace, crate::storage::installed_is_shared()).await
+    let shared = crate::storage::installed_is_shared();
+    crate::storage::agents::for_each_scope("run reaper", || reap_unless_shared(workspace, shared))
+        .await
+        .into_iter()
+        .map(|(_, reaped)| reaped)
+        .sum()
 }
 
 async fn reap_unless_shared(workspace: &Path, shared: bool) -> usize {

crates/openhuman-core/src/core/runtime/context.rs

@@ -372,7 +372,7 @@ impl CoreContext {
             }
         };
         let agent = self.agent.derive(&mut overlay);
-        Arc::new(CoreContext {
+        crate::storage::agents::registered(Arc::new(CoreContext {
             host_kind: self.host_kind,
             workspace_binding: RwLock::new(shared_binding),
             domains,
@@ -383,7 +383,7 @@ impl CoreContext {
             turn_origin: self.turn_origin.clone(),
             session_agent: overlay.session_agent.or_else(|| self.session_agent.clone()),
             agent,
-        })
+        }))
     }
 
     /// The agent a host session store scopes work under this context to, if
@@ -689,6 +689,8 @@ pub async fn init_stores(cfg: &crate::config::Config, domains: crate::core::runt
     }
 }
 
+#[path = "context_for_agent.rs"]
+mod for_agent;
 #[path = "context_turn_origin.rs"]
 mod turn_origin_scope;
 

crates/openhuman-core/src/core/runtime/context_for_agent.rs

@@ -0,0 +1,31 @@
+use super::*;
+
+impl CoreContext {
+    /// This context, acting for `agent`: the same configuration, workspace
+    /// binding, domains and transport, with the session agent replaced.
+    ///
+    /// For background work that visits an agent's records when no live
+    /// context of that agent is at hand (`crate::storage::agents`): the agent
+    /// was registered by an earlier process, or its host dropped it. A live
+    /// agent context is always preferred, since it carries the agent's own
+    /// configuration.
+    pub fn for_agent(self: &Arc<Self>, agent: &str) -> Arc<CoreContext> {
+        let binding = self
+            .workspace_binding
+            .read()
+            .map(|handle| Arc::clone(&*handle))
+            .unwrap_or_else(|poisoned| Arc::clone(&*poisoned.into_inner()));
+        Arc::new(CoreContext {
+            host_kind: self.host_kind,
+            workspace_binding: RwLock::new(binding),
+            domains: self.domains,
+            tool_groups: self.tool_groups.clone(),
+            embedder_config: self.embedder_config.clone(),
+            user_skill_roots: self.user_skill_roots,
+            backend_transport: self.backend_transport.clone(),
+            turn_origin: self.turn_origin.clone(),
+            session_agent: Some(agent.to_string()),
+            agent: Default::default(),
+        })
+    }
+}

crates/openhuman-core/src/flows/ops/run_management.rs

@@ -270,6 +270,32 @@ pub async fn sweep_expired_parked_runs(config: &Config) -> usize {
 /// resumable — only `pending_approval` is). Best-effort by construction: a store
 /// error is logged and the sweep returns what it managed.
 pub async fn sweep_orphaned_running_runs_on_boot(config: &Config) -> usize {
+    // `local`, then every agent that keeps its runs in its own storage scope
+    // (`crate::storage::agents`).
+    //
+    // On a backend other processes share (MongoDB), a `running` row below the
+    // boot floor can belong to a run another replica is still driving, and
+    // sweeping it would drop that run's checkpoint — so the agent scopes are
+    // left alone there, as the agent run reaper does.
+    if crate::storage::installed_is_shared() {
+        tracing::info!(target: "flows", "[flows] boot sweep: agent scopes skipped, the storage backend is shared");
+        if crate::core::runtime::mode::is_saas() {
+            // No `local` scope to fall back on: every row belongs to an agent.
+            return 0;
+        }
+        return sweep_orphaned_running_runs_in_scope(config).await;
+    }
+    crate::storage::agents::for_each_scope("flows boot sweep", || {
+        sweep_orphaned_running_runs_in_scope(config)
+    })
+    .await
+    .into_iter()
+    .map(|(_, swept)| swept)
+    .sum()
+}
+
+/// [`sweep_orphaned_running_runs_on_boot`] for the current storage scope.
+async fn sweep_orphaned_running_runs_in_scope(config: &Config) -> usize {
     let now_str = Utc::now().to_rfc3339();
     const REASON: &str =
         "Run interrupted by an app restart — no live run was executing this row after boot.";

crates/openhuman-core/src/flows/ops/triggers.rs

@@ -244,6 +244,29 @@ fn log_webhook_trigger_deferred(flow: &Flow, enabled: bool) {
 /// was lost some other way) gets its schedule re-registered on the next
 /// boot without the user having to toggle it off and on.
 pub async fn reconcile_schedule_triggers_on_boot(config: &Config) -> Result<(), String> {
+    // `local`, then every agent that keeps its flows in its own storage scope
+    // (`crate::storage::agents`); each scope's flows get their cron jobs in
+    // that scope, where the scheduler visits them.
+    let mut errors = Vec::new();
+    for (agent, result) in
+        crate::storage::agents::for_each_scope("flows schedule reconcile", || {
+            reconcile_schedule_triggers_in_scope(config)
+        })
+        .await
+    {
+        if let Err(error) = result {
+            errors.push(format!("{}: {error}", agent.as_deref().unwrap_or("local")));
+        }
+    }
+    if errors.is_empty() {
+        Ok(())
+    } else {
+        Err(errors.join("; "))
+    }
+}
+
+/// [`reconcile_schedule_triggers_on_boot`] for the current storage scope.
+async fn reconcile_schedule_triggers_in_scope(config: &Config) -> Result<(), String> {
     let (flows, skipped) = store::list_enabled_flows(config).map_err(|e| e.to_string())?;
     if skipped > 0 {
         // R-M4: a corrupt/unmigratable row must not abort boot reconciliation

crates/openhuman-core/src/integrations/task_sources/periodic.rs

@@ -43,10 +43,21 @@ fn last_poll_map() -> &'static LastPollMap {
     LAST_POLL_AT.get_or_init(|| Mutex::new(HashMap::new()))
 }
 
+/// The poll-tracking key for `source_id` in the current storage scope: the
+/// same source id in two agents' scopes is two different sources.
+fn poll_key(source_id: &str) -> String {
+    let agent = crate::core::runtime::CoreContext::current()
+        .and_then(|context| context.session_agent().map(str::to_string));
+    match agent {
+        Some(agent) => format!("{agent}\u{0}{source_id}"),
+        None => source_id.to_string(),
+    }
+}
+
 /// Record a successful (or attempted) poll for a source id.
 fn record_poll(source_id: &str) {
     if let Ok(mut map) = last_poll_map().lock() {
-        map.insert(source_id.to_string(), Instant::now());
+        map.insert(poll_key(source_id), Instant::now());
     }
 }
 
@@ -57,7 +68,7 @@ fn is_due(source: &TaskSource) -> bool {
         Ok(map) => map,
         Err(poisoned) => poisoned.into_inner(),
     };
-    match map.get(&source.id) {
+    match map.get(&poll_key(&source.id)) {
         Some(when) => when.elapsed() >= Duration::from_secs(interval_secs),
         None => true, // never polled this run — fire immediately
     }
@@ -70,7 +81,7 @@ pub fn start_periodic_poll() {
         tracing::debug!("[task_sources:periodic] scheduler already running, skipping start");
         return;
     }
-    tokio::spawn(async move {
+    crate::core::runtime::spawn_scoped(async move {
         tracing::info!(
             tick_seconds = TICK_SECONDS,
             "[task_sources:periodic] scheduler starting"
@@ -87,8 +98,14 @@ async fn run_loop() {
     ticker.tick().await;
     loop {
         ticker.tick().await;
-        if let Err(e) = run_one_tick().await {
-            tracing::warn!(error = %e, "[task_sources:periodic] tick failed (continuing)");
+        // `local`, then every agent that keeps its sources in its own storage
+        // scope (`crate::storage::agents`).
+        for (agent, result) in
+            crate::storage::agents::for_each_scope("task_sources", run_one_tick).await
+        {
+            if let Err(e) = result {
+                tracing::warn!(error = %e, agent = agent.as_deref().unwrap_or("local"), "[task_sources:periodic] tick failed (continuing)");
+            }
         }
     }
 }

crates/openhuman-core/src/integrations/task_sources/periodic_tests.rs

@@ -206,3 +206,24 @@ async fn manual_fetch_still_returns_the_explanatory_error() {
     assert_eq!(outcome.fetched, 0);
     assert_eq!(outcome.routed, 0);
 }
+
+#[tokio::test]
+async fn poll_timestamps_are_kept_per_agent_scope() {
+    use crate::core::runtime::{ContextOverlay, CoreContext, DomainSet};
+    let agent = |name: &str| {
+        CoreContext::for_test(DomainSet::full(), None).derive_with(
+            ContextOverlay::new(
+                crate::config::Config::default(),
+                DomainSet::full(),
+                Default::default(),
+            )
+            .session_agent(name),
+        )
+    };
+    let s = source("ts-per-agent-scope-xyz", 1800);
+    CoreContext::scope(agent("ts-agent-a"), async { record_poll(&s.id) }).await;
+    let due_for_a = CoreContext::scope(agent("ts-agent-a"), async { is_due(&s) }).await;
+    let due_for_b = CoreContext::scope(agent("ts-agent-b"), async { is_due(&s) }).await;
+    assert!(!due_for_a, "agent a just polled it");
+    assert!(due_for_b, "agent b's same-id source is a different source");
+}

crates/openhuman-core/src/security/devices/bus.rs

@@ -84,7 +84,25 @@ impl EventHandler<DomainEvent> for DeviceTunnelSubscriber {
                 channel_id,
                 payload_b64,
             } => {
-                handle_tunnel_frame(channel_id, payload_b64).await;
+                // Handle the frame as the agent the device belongs to, so its
+                // pairing record and the RPCs it sends land in that agent's
+                // scope (`super::owner`).
+                let pending = PENDING_SESSIONS
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner)
+                    .get(channel_id.as_str())
+                    .cloned();
+                let Ok(owner) = super::owner::owner_of(channel_id, pending.as_ref()).await else {
+                    log::warn!(
+                        "[devices/bus] dropping tunnel frame channel_id={channel_id}: owner unresolved"
+                    );
+                    return;
+                };
+                crate::storage::agents::within_agent(
+                    owner.as_deref(),
+                    handle_tunnel_frame(channel_id, payload_b64),
+                )
+                .await;
             }
             _ => {}
         }
@@ -361,6 +379,11 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) {
             &session_token_hash,
         ) {
             Ok(device) => {
+                super::owner::remember(
+                    channel_id,
+                    crate::core::runtime::CoreContext::current()
+                        .and_then(|context| context.session_agent().map(str::to_string)),
+                );
                 log::info!(
                     "[devices/bus] device persisted channel_id={} label={}",
                     device.channel_id,

crates/openhuman-core/src/security/devices/mod.rs

@@ -5,6 +5,7 @@
 
 pub mod bus;
 pub mod crypto;
+mod owner;
 pub mod rpc;
 pub mod schemas;
 pub mod store;

crates/openhuman-core/src/security/devices/owner.rs

@@ -0,0 +1,99 @@
+//! Which agent a paired device belongs to.
+//!
+//! A device paired from inside an agent's context is stored in that agent's
+//! storage scope (`crate::storage`), and the RPCs it sends through the tunnel
+//! must run as that agent — not as the process default, which would read and
+//! write somebody else's records. The tunnel subscriber runs outside any
+//! agent, so it asks here, per frame:
+//!
+//! 1. the pending pairing session, which recorded the agent that started it;
+//! 2. the owners this process has already resolved;
+//! 3. otherwise each storage scope, for a device paired by an earlier process
+//!    (`crate::storage::agents::for_each_scope`).
+//!
+//! `None` means the `local` scope (a single-user host, or no backend).
+
+use std::collections::HashMap;
+use std::sync::{LazyLock, Mutex};
+
+use super::types::PairingSession;
+
+/// Channels whose owner this process has resolved: `Some(agent)`, or `None`
+/// for `local`.
+static OWNERS: LazyLock<Mutex<HashMap<String, Option<String>>>> = LazyLock::new(Default::default);
+
+/// Records that `channel_id` belongs to `agent` (`None` = `local`).
+pub(super) fn remember(channel_id: &str, agent: Option<String>) {
+    OWNERS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .insert(channel_id.to_string(), agent);
+}
+
+fn cached(channel_id: &str) -> Option<Option<String>> {
+    OWNERS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .get(channel_id)
+        .cloned()
+}
+
+/// The agent `channel_id` belongs to, resolved as described above. A channel
+/// no scope knows yet (a handshake still in flight) is `local`.
+///
+/// # Errors
+///
+/// [`OwnerLookupFailed`] when no scope claimed the channel and at least one
+/// could not be searched (its configuration would not load): the device may
+/// well belong to that scope, so the caller must refuse the frame instead of
+/// running it as `local`.
+pub(super) async fn owner_of(
+    channel_id: &str,
+    pending: Option<&PairingSession>,
+) -> Result<Option<String>, OwnerLookupFailed> {
+    if let Some(session) = pending {
+        return Ok(session.agent.clone());
+    }
+    if let Some(owner) = cached(channel_id) {
+        return Ok(owner);
+    }
+    // The configuration is loaded inside each scope: in SaaS mode loading it
+    // needs an acting agent, which this tunnel task does not have.
+    let found = crate::storage::agents::for_each_scope("device owner", || async {
+        let Ok(config) = crate::config::rpc::load_config_with_timeout().await else {
+            return None;
+        };
+        Some(
+            super::store::get_device(&config, channel_id)
+                .ok()
+                .flatten()
+                .is_some(),
+        )
+    })
+    .await;
+    if let Some(owner) = found
+        .iter()
+        .find_map(|(agent, has_device)| (*has_device == Some(true)).then(|| agent.clone()))
+    {
+        log::debug!(
+            "[devices/owner] channel_id={channel_id} belongs to agent={}",
+            owner.as_deref().unwrap_or("local")
+        );
+        remember(channel_id, owner.clone());
+        return Ok(owner);
+    }
+    if found.iter().any(|(_, has_device)| has_device.is_none()) {
+        log::warn!("[devices/owner] channel_id={channel_id} could not be searched in every scope");
+        return Err(OwnerLookupFailed);
+    }
+    Ok(None)
+}
+
+/// A scope's configuration would not load, so the owner of a channel could not
+/// be ruled out there.
+#[derive(Debug, PartialEq, Eq)]
+pub(super) struct OwnerLookupFailed;
+
+#[cfg(test)]
+#[path = "owner_tests.rs"]
+mod tests;