diff --git a/crates/openhuman-core/src/core/runtime/agent_scope.rs b/crates/openhuman-core/src/core/runtime/agent_scope.rs
index 4f1e1a934fc..b10b551b8c2 100644
--- a/crates/openhuman-core/src/core/runtime/agent_scope.rs
+++ b/crates/openhuman-core/src/core/runtime/agent_scope.rs
@@ -93,15 +93,21 @@ pub struct AgentContextRegistry;
 impl AgentContextRegistry {
     /// Records `ctx` as the context of `agent_id`, replacing any dead entry.
     pub fn register(agent_id: &str, ctx: &Arc<CoreContext>) {
-        let mut map = AGENT_CONTEXTS
-            .write()
-            .unwrap_or_else(PoisonError::into_inner);
-        map.retain(|_, weak| weak.strong_count() > 0);
-        map.insert(agent_id.to_string(), Arc::downgrade(ctx));
-        log::debug!(
-            "[core-context] agent context registered agent={agent_id} live={}",
-            map.len()
-        );
+        {
+            let mut map = AGENT_CONTEXTS
+                .write()
+                .unwrap_or_else(PoisonError::into_inner);
+            map.retain(|_, weak| weak.strong_count() > 0);
+            map.insert(agent_id.to_string(), Arc::downgrade(ctx));
+            log::debug!(
+                "[core-context] agent context registered agent={agent_id} live={}",
+                map.len()
+            );
+        }
+        // With a storage backend, remember the agent there so a restarted
+        // process still visits its records (`crate::storage::agents`). After
+        // the lock is released: recording waits on the storage bridge.
+        crate::storage::agents::record(agent_id);
     }
 
     /// Removes `agent_id`'s entry when it still points at `ctx`.
diff --git a/crates/openhuman-core/src/core/runtime/context.rs b/crates/openhuman-core/src/core/runtime/context.rs
index b29a51dc69d..77cf5e46513 100644
--- a/crates/openhuman-core/src/core/runtime/context.rs
+++ b/crates/openhuman-core/src/core/runtime/context.rs
@@ -372,7 +372,7 @@ impl CoreContext {
             }
         };
         let agent = self.agent.derive(&mut overlay);
-        crate::storage::agents::registered(Arc::new(CoreContext {
+        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,11 +689,12 @@ 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;
 
+#[path = "context_for_agent.rs"]
+mod for_agent_scope;
+
 #[path = "context_agent.rs"]
 mod agent_parts;
 
diff --git a/crates/openhuman-core/src/core/runtime/context_for_agent.rs b/crates/openhuman-core/src/core/runtime/context_for_agent.rs
index 8f07ab2f7db..4b7f3bf0755 100644
--- a/crates/openhuman-core/src/core/runtime/context_for_agent.rs
+++ b/crates/openhuman-core/src/core/runtime/context_for_agent.rs
@@ -8,13 +8,23 @@ impl CoreContext {
     /// 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.
+    /// configuration, policy and tools.
     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()));
+        // The agent parts a derived context for `agent` would get: this
+        // context's policy, approval switch and definitions, and state slots
+        // of its own.
+        let mut overlay = ContextOverlay::new(
+            crate::config::Config::default(),
+            self.domains,
+            self.tool_groups.clone(),
+        )
+        .session_agent(agent);
+        let parts = self.agent.derive(&mut overlay);
         Arc::new(CoreContext {
             host_kind: self.host_kind,
             workspace_binding: RwLock::new(binding),
@@ -25,7 +35,7 @@ impl CoreContext {
             backend_transport: self.backend_transport.clone(),
             turn_origin: self.turn_origin.clone(),
             session_agent: Some(agent.to_string()),
-            agent: Default::default(),
+            agent: parts,
         })
     }
 }
diff --git a/crates/openhuman-core/src/cron/completion_owner.rs b/crates/openhuman-core/src/cron/completion_owner.rs
new file mode 100644
index 00000000000..b4417432aa8
--- /dev/null
+++ b/crates/openhuman-core/src/cron/completion_owner.rs
@@ -0,0 +1,49 @@
+//! Which agent a just-completed cron job belonged to.
+//!
+//! `CronJobCompleted` names only the job, and a one-shot job
+//! (`delete_after_run`) is already gone from the store by the time a
+//! subscriber sees the event, so its owner cannot be looked up there. The
+//! scheduler publishes the event inside the job's own task, which carries the
+//! agent's context, and notes the owner here first; the notification bridge
+//! takes it back out (`desktop::notifications`).
+//!
+//! In-process only — the bus delivers locally — and bounded: if no subscriber
+//! drains it (the bridge is not registered), it is cleared rather than grown.
+
+use std::collections::HashMap;
+use std::sync::{LazyLock, Mutex};
+
+/// Notes kept before the map is cleared outright.
+const CAPACITY: usize = 1024;
+
+/// Job id → the agent it ran as.
+static OWNERS: LazyLock<Mutex<HashMap<String, String>>> = LazyLock::new(Default::default);
+
+/// Notes that `job_id` just completed as the current context's agent, when
+/// it ran as one. Call just before publishing `CronJobCompleted`.
+pub fn note(job_id: &str) {
+    let Some(agent) = crate::core::runtime::CoreContext::current()
+        .and_then(|context| context.session_agent().map(str::to_string))
+    else {
+        return;
+    };
+    let mut owners = OWNERS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    if owners.len() >= CAPACITY {
+        owners.clear();
+    }
+    owners.insert(job_id.to_string(), agent);
+}
+
+/// The agent `job_id` completed as, if one was noted; removes the note.
+pub fn take(job_id: &str) -> Option<String> {
+    OWNERS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .remove(job_id)
+}
+
+#[cfg(test)]
+#[path = "completion_owner_tests.rs"]
+mod tests;
diff --git a/crates/openhuman-core/src/cron/completion_owner_tests.rs b/crates/openhuman-core/src/cron/completion_owner_tests.rs
new file mode 100644
index 00000000000..bc7d01b38af
--- /dev/null
+++ b/crates/openhuman-core/src/cron/completion_owner_tests.rs
@@ -0,0 +1,26 @@
+use super::*;
+use crate::core::runtime::{ContextOverlay, CoreContext, DomainSet};
+
+#[tokio::test]
+async fn a_job_completed_as_an_agent_is_noted_once() {
+    let agent = CoreContext::for_test(DomainSet::full(), None).derive_with(
+        ContextOverlay::new(
+            crate::config::Config::default(),
+            DomainSet::full(),
+            Default::default(),
+        )
+        .session_agent("completion-owner-agent"),
+    );
+    CoreContext::scope(agent, async { note("completion-owner-job") }).await;
+    assert_eq!(
+        take("completion-owner-job").as_deref(),
+        Some("completion-owner-agent")
+    );
+    assert_eq!(take("completion-owner-job"), None, "taken once");
+}
+
+#[test]
+fn a_job_completed_without_an_agent_is_not_noted() {
+    note("completion-owner-local");
+    assert_eq!(take("completion-owner-local"), None);
+}
diff --git a/crates/openhuman-core/src/cron/mod.rs b/crates/openhuman-core/src/cron/mod.rs
index a84a6f84e45..aff93a72f02 100644
--- a/crates/openhuman-core/src/cron/mod.rs
+++ b/crates/openhuman-core/src/cron/mod.rs
@@ -11,6 +11,7 @@ pub mod scheduler_gate;
 
 pub mod bus;
 pub mod channel_bridge;
+pub mod completion_owner;
 pub mod job_builder;
 pub mod ops;
 pub mod origin;
diff --git a/crates/openhuman-core/src/cron/scheduler.rs b/crates/openhuman-core/src/cron/scheduler.rs
index e0f6949fd96..5c64479da1e 100644
--- a/crates/openhuman-core/src/cron/scheduler.rs
+++ b/crates/openhuman-core/src/cron/scheduler.rs
@@ -99,9 +99,11 @@ pub async fn run(config: Config) -> Result<()> {
 }
 
 /// Runs the due jobs of every live embedded agent, each under that agent's
-/// own context: its config, provider route, policy and job database. An
-/// agent that is not live has no context to run under, so its jobs stay
-/// dormant until it is instantiated again.
+/// own context: its config, provider route, policy and job store. An agent
+/// that is not live has no context to run under — no host tools or prompt
+/// for its agent jobs — so its jobs stay dormant until it is instantiated
+/// again, including when a storage backend still records it
+/// (`crate::storage::agents`).
 pub(crate) async fn tick_live_agents(dispatcher: &mut JobDispatcher) {
     for (agent_id, ctx) in crate::core::runtime::AgentContextRegistry::live() {
         let agent = agent_id.clone();
@@ -114,7 +116,7 @@ pub(crate) async fn tick_live_agents(dispatcher: &mut JobDispatcher) {
                     return;
                 }
             };
-            if !crate::cron::store::db_path(&config).exists() {
+            if !agent_jobs_may_exist(&config) {
                 return;
             }
             let jobs = match due_jobs(&config, Utc::now()) {
@@ -138,6 +140,13 @@ pub(crate) async fn tick_live_agents(dispatcher: &mut JobDispatcher) {
     }
 }
 
+/// Whether an agent can have jobs to poll: with a storage backend its jobs
+/// live in the backend under its scope (`crate::storage`), so there is always
+/// something to ask; without one, only once its own `jobs.db` exists.
+pub(crate) fn agent_jobs_may_exist(config: &Config) -> bool {
+    crate::storage::installed().is_some() || crate::cron::store::db_path(config).exists()
+}
+
 /// Single poll cycle of the scheduler loop, extracted so tests can drive
 /// it without owning `tokio::time::interval`.
 ///
@@ -254,6 +263,9 @@ pub(super) async fn execute_and_persist_job(
     )
     .await;
 
+    // The job ran as its agent (when it had one); a one-shot job is already
+    // deleted, so subscribers learn the owner here, not from the store.
+    crate::cron::completion_owner::note(&job.id);
     BUS.publish(DomainEvent::CronJobCompleted {
         job_id: job.id.clone(),
         success,
diff --git a/crates/openhuman-core/src/cron/scheduler_tests.rs b/crates/openhuman-core/src/cron/scheduler_tests.rs
index 4aa6223ac01..012b0200b42 100644
--- a/crates/openhuman-core/src/cron/scheduler_tests.rs
+++ b/crates/openhuman-core/src/cron/scheduler_tests.rs
@@ -317,3 +317,19 @@ async fn a_pipeline_reports_its_last_stage_rather_than_pipefail() {
          a failure and re-status every existing job with a pipeline in it: {output}"
     );
 }
+
+#[tokio::test]
+async fn without_a_backend_an_agent_needs_its_own_job_database() {
+    // The lib test binary installs no storage backend into the process slot.
+    if crate::storage::installed().is_some() {
+        return;
+    }
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp).await;
+    let db = crate::cron::store::db_path(&config);
+    let _ = std::fs::remove_file(&db);
+    assert!(!agent_jobs_may_exist(&config));
+    std::fs::create_dir_all(db.parent().unwrap()).unwrap();
+    std::fs::write(&db, b"").unwrap();
+    assert!(agent_jobs_may_exist(&config));
+}
diff --git a/crates/openhuman-core/src/desktop/notifications/bus.rs b/crates/openhuman-core/src/desktop/notifications/bus.rs
index 242fbdb767b..c81d7c82c20 100644
--- a/crates/openhuman-core/src/desktop/notifications/bus.rs
+++ b/crates/openhuman-core/src/desktop/notifications/bus.rs
@@ -469,6 +469,30 @@ fn translate(event: &DomainEvent) -> Option<CoreNotificationEvent> {
     }
 }
 
+/// The agent whose storage scope a notification raised by `event` belongs
+/// in: the owner of the cron job a `CronJobCompleted` names, when a storage
+/// backend is installed and an agent's scope holds the job; `None` (`local`)
+/// otherwise.
+async fn event_owner(config: &crate::config::Config, event: &DomainEvent) -> Option<String> {
+    let DomainEvent::CronJobCompleted { job_id, .. } = event else {
+        return None;
+    };
+    // Noted by the scheduler as the job completed — the only record of a
+    // one-shot job, which is deleted before the event is published.
+    if let Some(agent) = crate::cron::completion_owner::take(job_id) {
+        return Some(agent);
+    }
+    // No backend: every record is `local`.
+    crate::storage::installed()?;
+    // Found in `local` (`Some(None)`) and found nowhere (`None`) both mean
+    // the notification is stored in `local`.
+    crate::storage::agents::find_owner("notification owner", || async {
+        crate::cron::get_job(config, job_id).is_ok()
+    })
+    .await
+    .flatten()
+}
+
 #[async_trait]
 impl EventHandler<DomainEvent> for NotificationBridgeSubscriber {
     fn name(&self) -> &str {
@@ -491,7 +515,15 @@ impl EventHandler<DomainEvent> for NotificationBridgeSubscriber {
                 // A workspace-bound event is stored in ITS OWN workspace, not
                 // whichever one this bridge was registered with (#5931).
                 let config = self.store_target(config, event);
-                match super::store::insert_core_notification(&config, &notification) {
+                // Stored with the record that raised it: a cron job's
+                // completion goes to the agent the job belongs to
+                // (`crate::storage`), `local` otherwise.
+                let owner = event_owner(&config, event).await;
+                let stored = crate::storage::agents::within_agent(owner.as_deref(), async {
+                    super::store::insert_core_notification(&config, &notification)
+                })
+                .await;
+                match stored {
                     Ok(true) => log::debug!(
                         "{LOG_PREFIX} persisted core notification id={}",
                         notification.id
diff --git a/crates/openhuman-core/src/flows/bus.rs b/crates/openhuman-core/src/flows/bus.rs
index db2a222e683..a196a4d0650 100644
--- a/crates/openhuman-core/src/flows/bus.rs
+++ b/crates/openhuman-core/src/flows/bus.rs
@@ -15,6 +15,7 @@
 mod tests;
 
 mod dedup_commit;
+mod owner;
 mod run_digest;
 mod trigger;
 
diff --git a/crates/openhuman-core/src/flows/bus/dedup_commit.rs b/crates/openhuman-core/src/flows/bus/dedup_commit.rs
index 8116d8a9049..9f9fc1aaee1 100644
--- a/crates/openhuman-core/src/flows/bus/dedup_commit.rs
+++ b/crates/openhuman-core/src/flows/bus/dedup_commit.rs
@@ -325,7 +325,13 @@ impl EventHandler<DomainEvent> for DedupCommitSubscriber {
             status,
         } = event
         {
-            self.handle_finished(flow_id, run_id, status).await;
+            // Settle the run as the agent its flow belongs to (`super::owner`).
+            let owner = super::owner::flow_owner(&self.config, flow_id).await;
+            crate::storage::agents::within_agent(
+                owner.as_deref(),
+                self.handle_finished(flow_id, run_id, status),
+            )
+            .await;
         }
     }
 }
diff --git a/crates/openhuman-core/src/flows/bus/owner.rs b/crates/openhuman-core/src/flows/bus/owner.rs
new file mode 100644
index 00000000000..fc99e206947
--- /dev/null
+++ b/crates/openhuman-core/src/flows/bus/owner.rs
@@ -0,0 +1,73 @@
+//! Which agent a flow belongs to, for the subscribers that act on a flow
+//! named only by id in an event (`FlowScheduleTick`, `FlowRunFinished`).
+//!
+//! A flow saved from inside an agent's context lives in that agent's storage
+//! scope (`crate::storage`). The subscribers run outside any agent, so they
+//! look the flow up here and handle the event as its owner
+//! (`crate::storage::agents::within_agent`); without a storage backend every
+//! flow is `local` and this is a no-op.
+
+use std::collections::HashMap;
+use std::sync::{LazyLock, Mutex};
+
+use crate::config::Config;
+
+/// Flows whose owner this process has resolved (`None` = `local`).
+static OWNERS: LazyLock<Mutex<HashMap<String, Option<String>>>> = LazyLock::new(Default::default);
+
+/// The agent `flow_id` belongs to: `None` for `local`, or when no scope has
+/// the flow (the handler then reports the flow as unknown, as before).
+pub(super) async fn flow_owner(config: &Config, flow_id: &str) -> Option<String> {
+    // No backend: every record is `local`.
+    crate::storage::installed()?;
+    resolve(config, flow_id).await
+}
+
+/// [`flow_owner`]'s cache and lookup, across whatever scopes exist.
+async fn resolve(config: &Config, flow_id: &str) -> Option<String> {
+    let cached = OWNERS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .get(flow_id)
+        .cloned();
+    if let Some(owner) = cached {
+        // Still there? A removed flow's id may be reused, or the flow moved:
+        // re-check the cached scope and resolve again when it lost the flow.
+        let still_there = crate::storage::agents::within_agent(owner.as_deref(), async {
+            matches!(crate::flows::store::get_flow(config, flow_id), Ok(Some(_)))
+        })
+        .await;
+        if still_there {
+            return owner;
+        }
+        forget(flow_id);
+    }
+    let found = crate::storage::agents::find_owner("flow owner", || async {
+        matches!(crate::flows::store::get_flow(config, flow_id), Ok(Some(_)))
+    })
+    .await;
+    let owner = found?;
+    tracing::debug!(
+        target: "flows",
+        %flow_id,
+        agent = owner.as_deref().unwrap_or("local"),
+        "[flows] resolved the flow's owner"
+    );
+    OWNERS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .insert(flow_id.to_string(), owner.clone());
+    owner
+}
+
+/// Forgets a cached owner (a removed flow's id may be reused).
+pub(super) fn forget(flow_id: &str) {
+    OWNERS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .remove(flow_id);
+}
+
+#[cfg(test)]
+#[path = "owner_tests.rs"]
+mod tests;
diff --git a/crates/openhuman-core/src/flows/bus/owner_tests.rs b/crates/openhuman-core/src/flows/bus/owner_tests.rs
new file mode 100644
index 00000000000..0cdc426de8c
--- /dev/null
+++ b/crates/openhuman-core/src/flows/bus/owner_tests.rs
@@ -0,0 +1,67 @@
+use super::*;
+
+#[tokio::test]
+async fn without_a_backend_every_flow_is_local() {
+    // The lib test binary never installs a backend into the process slot.
+    if crate::storage::installed().is_some() {
+        return;
+    }
+    let dir = tempfile::tempdir().unwrap();
+    let config = Config {
+        workspace_dir: dir.path().to_path_buf(),
+        ..Config::default()
+    };
+    assert_eq!(flow_owner(&config, "any-flow").await, None);
+}
+
+#[test]
+fn a_forgotten_owner_is_resolved_again() {
+    OWNERS
+        .lock()
+        .unwrap()
+        .insert("owner-test-flow".to_string(), Some("agent-1".to_string()));
+    forget("owner-test-flow");
+    assert!(!OWNERS.lock().unwrap().contains_key("owner-test-flow"));
+}
+
+fn local_flow(id: &str) -> crate::flows::Flow {
+    crate::flows::Flow {
+        id: id.to_string(),
+        name: id.to_string(),
+        enabled: true,
+        graph: tinyflows::model::WorkflowGraph::default(),
+        created_at: "2026-01-01T00:00:00Z".to_string(),
+        updated_at: "2026-01-01T00:00:00Z".to_string(),
+        last_run_at: None,
+        last_status: None,
+        require_approval: false,
+    }
+}
+
+#[tokio::test]
+async fn a_cached_owner_is_dropped_once_its_scope_loses_the_flow() {
+    if crate::storage::installed().is_some() {
+        return;
+    }
+    let dir = tempfile::tempdir().unwrap();
+    let config = Config {
+        workspace_dir: dir.path().to_path_buf(),
+        ..Config::default()
+    };
+    crate::flows::store::upsert_flow(&config, &local_flow("owner-test-revalidate")).unwrap();
+
+    // Found in `local`, and remembered there.
+    assert_eq!(resolve(&config, "owner-test-revalidate").await, None);
+    assert_eq!(
+        OWNERS.lock().unwrap().get("owner-test-revalidate"),
+        Some(&None)
+    );
+    // Still there: the cached answer stands.
+    assert_eq!(resolve(&config, "owner-test-revalidate").await, None);
+    assert!(OWNERS.lock().unwrap().contains_key("owner-test-revalidate"));
+
+    // Gone: the cache entry is dropped and nothing is found.
+    crate::flows::store::remove_flow(&config, "owner-test-revalidate").unwrap();
+    assert_eq!(resolve(&config, "owner-test-revalidate").await, None);
+    assert!(!OWNERS.lock().unwrap().contains_key("owner-test-revalidate"));
+}
diff --git a/crates/openhuman-core/src/flows/bus/run_digest.rs b/crates/openhuman-core/src/flows/bus/run_digest.rs
index ce92fdd0447..ff340fc83db 100644
--- a/crates/openhuman-core/src/flows/bus/run_digest.rs
+++ b/crates/openhuman-core/src/flows/bus/run_digest.rs
@@ -189,7 +189,13 @@ impl EventHandler<DomainEvent> for FlowRunDigestSubscriber {
             status,
         } = event
         {
-            self.handle_finished(flow_id, run_id, status).await;
+            // Settle the run as the agent its flow belongs to (`super::owner`).
+            let owner = super::owner::flow_owner(&self.config, flow_id).await;
+            crate::storage::agents::within_agent(
+                owner.as_deref(),
+                self.handle_finished(flow_id, run_id, status),
+            )
+            .await;
         }
     }
 }
diff --git a/crates/openhuman-core/src/flows/bus/trigger.rs b/crates/openhuman-core/src/flows/bus/trigger.rs
index 9fc7f26695b..80bacb58ef5 100644
--- a/crates/openhuman-core/src/flows/bus/trigger.rs
+++ b/crates/openhuman-core/src/flows/bus/trigger.rs
@@ -111,12 +111,26 @@ impl FlowTriggerSubscriber {
         })
     }
 
+    /// The configuration to handle an event under: the acting agent's own
+    /// when the handler runs as one (`super::owner`, `crate::storage::agents`)
+    /// — its provider, access policy and action directory — else the one this
+    /// subscriber was registered with.
+    pub(super) fn config_for_scope(&self) -> Arc<Config> {
+        let acting = crate::core::runtime::CoreContext::current()
+            .is_some_and(|context| context.session_agent().is_some());
+        match acting.then(crate::core::runtime::CoreContext::current_embedder_config) {
+            Some(Some(config)) => Arc::new(config),
+            _ => Arc::clone(&self.config),
+        }
+    }
+
     /// `DomainEvent::FlowScheduleTick` — a `flow`-type cron job fired. Loads
     /// the one named flow, checks it is still enabled with a `schedule`
     /// trigger (it may have been disabled/edited since the job was
     /// registered), and dispatches it with an empty trigger payload.
     async fn handle_schedule_tick(&self, flow_id: &str) {
-        let flow = match store::get_flow(&self.config, flow_id) {
+        let config = self.config_for_scope();
+        let flow = match store::get_flow(&config, flow_id) {
             Ok(Some(flow)) => flow,
             Ok(None) => {
                 tracing::debug!(target: "flows", %flow_id, "[flows] schedule tick for unknown/removed flow — ignoring");
@@ -149,7 +163,8 @@ impl FlowTriggerSubscriber {
     /// dispatches each match with the event payload as the run input
     /// (seeded into `run.trigger`, per the node-catalog contract).
     async fn handle_app_event(&self, toolkit: &str, trigger_slug: &str, payload: &Value) {
-        let (flows, skipped) = match store::list_enabled_flows(&self.config) {
+        let config = self.config_for_scope();
+        let (flows, skipped) = match store::list_enabled_flows(&config) {
             Ok(result) => result,
             Err(e) => {
                 tracing::warn!(target: "flows", %toolkit, %trigger_slug, error = %e, "[flows] failed to list enabled flows for app_event dispatch");
@@ -198,8 +213,9 @@ impl FlowTriggerSubscriber {
             return;
         };
 
-        let config = self.config.clone();
-        tokio::spawn(async move {
+        let config = self.config_for_scope();
+        // Scoped: the run keeps the owner's context the handler entered.
+        crate::core::runtime::spawn_scoped(async move {
             // Held for the lifetime of the run; released on drop (including
             // on panic) by `InFlightGuard`.
             let _guard = guard;
@@ -247,13 +263,28 @@ impl EventHandler<DomainEvent> for FlowTriggerSubscriber {
 
     async fn handle(&self, event: &DomainEvent) {
         match event {
-            DomainEvent::FlowScheduleTick { flow_id } => self.handle_schedule_tick(flow_id).await,
+            // The flow runs as the agent it belongs to (`super::owner`).
+            DomainEvent::FlowScheduleTick { flow_id } => {
+                let owner = super::owner::flow_owner(&self.config, flow_id).await;
+                crate::storage::agents::within_agent(
+                    owner.as_deref(),
+                    self.handle_schedule_tick(flow_id),
+                )
+                .await
+            }
+            // Every scope's flows may listen for this trigger: match them in
+            // each, so each run starts as its flow's owner.
             DomainEvent::ComposioTriggerReceived {
                 toolkit,
                 trigger,
                 payload,
                 ..
-            } => self.handle_app_event(toolkit, trigger, payload).await,
+            } => {
+                crate::storage::agents::for_each_live_scope("flows app_event", || {
+                    self.handle_app_event(toolkit, trigger, payload)
+                })
+                .await;
+            }
             DomainEvent::WebhookIncomingRequest { .. } => {
                 // Best-effort deviation (documented, not silently skipped —
                 // see `flows::ops::log_webhook_trigger_deferred` for the
diff --git a/crates/openhuman-core/src/flows/bus_tests.rs b/crates/openhuman-core/src/flows/bus_tests.rs
index b42463ce9ea..b723aee8111 100644
--- a/crates/openhuman-core/src/flows/bus_tests.rs
+++ b/crates/openhuman-core/src/flows/bus_tests.rs
@@ -114,3 +114,31 @@ fn dedup_state_namespace(flow_id: &str) -> String {
 mod dedup_commit_lock_tests;
 #[path = "bus_subscriber_tests.rs"]
 mod subscriber_tests;
+
+#[tokio::test]
+async fn a_trigger_handled_as_an_agent_uses_the_agent_config() {
+    use crate::core::runtime::{ContextOverlay, CoreContext, DomainSet};
+    let registered = crate::config::Config {
+        workspace_dir: std::path::PathBuf::from("/tmp/flows-trigger-registered"),
+        ..crate::config::Config::default()
+    };
+    let subscriber = FlowTriggerSubscriber::new(std::sync::Arc::new(registered));
+    assert_eq!(
+        subscriber.config_for_scope().workspace_dir,
+        std::path::PathBuf::from("/tmp/flows-trigger-registered"),
+        "outside an agent the registered config is used"
+    );
+    let agent_config = crate::config::Config {
+        workspace_dir: std::path::PathBuf::from("/tmp/flows-trigger-agent"),
+        ..crate::config::Config::default()
+    };
+    let agent = CoreContext::for_test(DomainSet::full(), None).derive_with(
+        ContextOverlay::new(agent_config, DomainSet::full(), Default::default())
+            .session_agent("flows-trigger-agent"),
+    );
+    let seen = CoreContext::scope(agent, async {
+        subscriber.config_for_scope().workspace_dir.clone()
+    })
+    .await;
+    assert_eq!(seen, std::path::PathBuf::from("/tmp/flows-trigger-agent"));
+}
diff --git a/crates/openhuman-core/src/flows/ops/run_management.rs b/crates/openhuman-core/src/flows/ops/run_management.rs
index fc75f018e4b..7d7171a0cea 100644
--- a/crates/openhuman-core/src/flows/ops/run_management.rs
+++ b/crates/openhuman-core/src/flows/ops/run_management.rs
@@ -270,28 +270,52 @@ 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;
+    match boot_sweep_plan(
+        crate::storage::installed_is_shared(),
+        crate::core::runtime::mode::is_saas(),
+    ) {
+        BootSweepPlan::Nothing => {
+            tracing::info!(target: "flows", "[flows] boot sweep: skipped, the storage backend is shared");
+            0
+        }
+        BootSweepPlan::LocalOnly => {
+            tracing::info!(target: "flows", "[flows] boot sweep: agent scopes skipped, the storage backend is shared");
+            sweep_orphaned_running_runs_in_scope(config).await
+        }
+        BootSweepPlan::EveryScope => {
+            crate::storage::agents::for_each_scope("flows boot sweep", || {
+                sweep_orphaned_running_runs_in_scope(config)
+            })
+            .await
+            .into_iter()
+            .map(|(_, swept)| swept)
+            .sum()
         }
-        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()
+}
+
+/// Which scopes the boot sweep reconciles.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum BootSweepPlan {
+    /// `local`, then every agent scope (`crate::storage::agents`).
+    EveryScope,
+    /// `local` only.
+    LocalOnly,
+    /// No scope.
+    Nothing,
+}
+
+/// The boot sweep's scopes. 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 — and in
+/// SaaS mode, where every scope is an agent's, nothing is swept.
+pub(crate) fn boot_sweep_plan(shared: bool, saas: bool) -> BootSweepPlan {
+    match (shared, saas) {
+        (false, _) => BootSweepPlan::EveryScope,
+        (true, false) => BootSweepPlan::LocalOnly,
+        (true, true) => BootSweepPlan::Nothing,
+    }
 }
 
 /// [`sweep_orphaned_running_runs_on_boot`] for the current storage scope.
diff --git a/crates/openhuman-core/src/flows/ops_tests.rs b/crates/openhuman-core/src/flows/ops_tests.rs
index fb5e5291d8e..36fbbb5ca77 100644
--- a/crates/openhuman-core/src/flows/ops_tests.rs
+++ b/crates/openhuman-core/src/flows/ops_tests.rs
@@ -672,3 +672,12 @@ mod tool_contract_and_wiring_warnings_tests;
 mod triggers_and_resume_tests;
 #[path = "ops_validate_warnings_and_connections_tests.rs"]
 mod validate_warnings_and_connections_tests;
+
+#[test]
+fn the_boot_sweep_leaves_shared_backends_alone() {
+    use super::run_management::{boot_sweep_plan, BootSweepPlan};
+    assert_eq!(boot_sweep_plan(false, false), BootSweepPlan::EveryScope);
+    assert_eq!(boot_sweep_plan(false, true), BootSweepPlan::EveryScope);
+    assert_eq!(boot_sweep_plan(true, false), BootSweepPlan::LocalOnly);
+    assert_eq!(boot_sweep_plan(true, true), BootSweepPlan::Nothing);
+}
diff --git a/crates/openhuman-core/src/integrations/task_sources/bus.rs b/crates/openhuman-core/src/integrations/task_sources/bus.rs
index d380df792d3..0d9c4806ac6 100644
--- a/crates/openhuman-core/src/integrations/task_sources/bus.rs
+++ b/crates/openhuman-core/src/integrations/task_sources/bus.rs
@@ -60,39 +60,56 @@ impl EventHandler<DomainEvent> for TaskSourcesConnectionSubscriber {
             return;
         }
 
-        let sources = match store::list_sources(&config) {
-            Ok(sources) => sources,
-            Err(e) => {
-                tracing::debug!(error = %e, "[task_sources:bus] list_sources failed, skipping");
-                return;
-            }
-        };
+        // Every scope may hold sources for this toolkit (`crate::storage`):
+        // fire each scope's under its own agent.
+        crate::storage::agents::for_each_live_scope("task_sources connection", || {
+            fire_for_connection(&config, provider, toolkit, connection_id)
+        })
+        .await;
+    }
+}
 
-        for source in sources
-            .into_iter()
-            .filter(|s| s.enabled && s.provider == provider)
-        {
-            // If the source pins a specific connection, only fire for it.
-            if let Some(pinned) = source.connection_id.as_deref() {
-                if pinned != connection_id {
-                    continue;
-                }
+/// Starts a one-shot fetch for every enabled source of `provider` in the
+/// current storage scope that the new connection serves.
+async fn fire_for_connection(
+    config: &crate::config::Config,
+    provider: ProviderSlug,
+    toolkit: &str,
+    connection_id: &str,
+) {
+    let sources = match store::list_sources(config) {
+        Ok(sources) => sources,
+        Err(e) => {
+            tracing::debug!(error = %e, "[task_sources:bus] list_sources failed, skipping");
+            return;
+        }
+    };
+
+    for source in sources
+        .into_iter()
+        .filter(|s| s.enabled && s.provider == provider)
+    {
+        // If the source pins a specific connection, only fire for it.
+        if let Some(pinned) = source.connection_id.as_deref() {
+            if pinned != connection_id {
+                continue;
             }
-            tracing::info!(
-                source_id = %source.id,
-                toolkit = %toolkit,
-                "[task_sources:bus] connection created → one-shot fetch"
-            );
-            // Spawn each fetch independently so the event handler does not
-            // block dispatch on N sequential network round-trips (same
-            // pattern as the periodic poll). Each fetch captures its own
-            // owned config + source.
-            let config = config.clone();
-            tokio::spawn(async move {
-                let _ = pipeline::run_source_once(&config, &source, FetchReason::ConnectionCreated)
-                    .await;
-            });
         }
+        tracing::info!(
+            source_id = %source.id,
+            toolkit = %toolkit,
+            "[task_sources:bus] connection created → one-shot fetch"
+        );
+        // Spawn each fetch independently so the event handler does not
+        // block dispatch on N sequential network round-trips (same
+        // pattern as the periodic poll). Each fetch captures its own
+        // owned config + source.
+        let config = config.clone();
+        // Scoped: the fetch keeps the scope's agent context.
+        crate::core::runtime::spawn_scoped(async move {
+            let _ =
+                pipeline::run_source_once(&config, &source, FetchReason::ConnectionCreated).await;
+        });
     }
 }
 
@@ -115,3 +132,7 @@ pub fn register_task_sources_subscriber() {
         }
     }
 }
+
+#[cfg(test)]
+#[path = "bus_tests.rs"]
+mod tests;
diff --git a/crates/openhuman-core/src/integrations/task_sources/bus_tests.rs b/crates/openhuman-core/src/integrations/task_sources/bus_tests.rs
new file mode 100644
index 00000000000..ece2eb15f26
--- /dev/null
+++ b/crates/openhuman-core/src/integrations/task_sources/bus_tests.rs
@@ -0,0 +1,61 @@
+use super::*;
+use crate::integrations::task_sources::types::{FilterSpec, SourceTarget};
+use serde_json::json;
+
+fn github_filter() -> FilterSpec {
+    FilterSpec::Github {
+        repo: Some("tinyhumansai/openhuman".into()),
+        labels: Vec::new(),
+        assignee_is_me: true,
+        state: None,
+        fetch_mode: Default::default(),
+        extra: json!({}),
+    }
+}
+
+/// A new connection starts a fetch for the scope's enabled sources of that
+/// toolkit — skipping ones pinned to another connection — and returns
+/// without waiting for the fetches.
+#[tokio::test]
+async fn a_new_connection_fires_the_matching_sources_of_the_scope() {
+    let tmp = tempfile::tempdir().unwrap();
+    let config = crate::config::Config {
+        workspace_dir: tmp.path().join("workspace"),
+        action_dir: tmp.path().join("workspace"),
+        config_path: tmp.path().join("config.toml"),
+        ..crate::config::Config::default()
+    };
+    std::fs::create_dir_all(&config.workspace_dir).unwrap();
+    for connection in [None, Some("conn-other".to_string())] {
+        store::add_source(
+            &config,
+            ProviderSlug::Github,
+            connection,
+            None,
+            github_filter(),
+            600,
+            SourceTarget::TodoOnly,
+            5,
+        )
+        .unwrap();
+    }
+    fire_for_connection(&config, ProviderSlug::Github, "github", "conn-1").await;
+    // A toolkit with no sources fires nothing.
+    fire_for_connection(&config, ProviderSlug::Notion, "notion", "conn-1").await;
+    assert_eq!(store::list_sources(&config).unwrap().len(), 2);
+}
+
+/// A scope whose sources cannot be listed is skipped, not an error.
+#[tokio::test]
+async fn an_unreadable_scope_is_skipped() {
+    let tmp = tempfile::tempdir().unwrap();
+    // A file where the workspace directory should be: the store cannot open.
+    let blocked = tmp.path().join("blocked");
+    std::fs::write(&blocked, b"").unwrap();
+    let config = crate::config::Config {
+        workspace_dir: blocked.clone(),
+        action_dir: blocked,
+        ..crate::config::Config::default()
+    };
+    fire_for_connection(&config, ProviderSlug::Github, "github", "conn-1").await;
+}
diff --git a/crates/openhuman-core/src/integrations/task_sources/periodic.rs b/crates/openhuman-core/src/integrations/task_sources/periodic.rs
index f5464fa82ef..d38cc0e530a 100644
--- a/crates/openhuman-core/src/integrations/task_sources/periodic.rs
+++ b/crates/openhuman-core/src/integrations/task_sources/periodic.rs
@@ -101,7 +101,7 @@ async fn run_loop() {
         // `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
+            crate::storage::agents::for_each_live_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)");
diff --git a/crates/openhuman-core/src/security/devices/bus.rs b/crates/openhuman-core/src/security/devices/bus.rs
index 076d426478d..d2dc9ef6086 100644
--- a/crates/openhuman-core/src/security/devices/bus.rs
+++ b/crates/openhuman-core/src/security/devices/bus.rs
@@ -92,17 +92,22 @@ impl EventHandler<DomainEvent> for DeviceTunnelSubscriber {
                     .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;
+                match super::owner::owner_of(channel_id, pending.as_ref()).await {
+                    Ok(owner) => {
+                        crate::storage::agents::within_agent(
+                            owner.as_deref(),
+                            handle_tunnel_frame(channel_id, payload_b64),
+                        )
+                        .await;
+                    }
+                    // Fail closed: an unknown owner must not become `local`.
+                    Err(failed) => log::warn!(
+                        "[devices/bus] dropping tunnel frame channel_id={channel_id}: owner \
+                         lookup failed in scope={} ({})",
+                        failed.agent.as_deref().unwrap_or("local"),
+                        failed.error
+                    ),
+                }
             }
             _ => {}
         }
diff --git a/crates/openhuman-core/src/security/devices/owner.rs b/crates/openhuman-core/src/security/devices/owner.rs
index bcb1ad59660..569401dca07 100644
--- a/crates/openhuman-core/src/security/devices/owner.rs
+++ b/crates/openhuman-core/src/security/devices/owner.rs
@@ -38,15 +38,24 @@ fn cached(channel_id: &str) -> Option<Option<String>> {
         .cloned()
 }
 
-/// The agent `channel_id` belongs to, resolved as described above. A channel
-/// no scope knows yet (a handshake still in flight) is `local`.
+/// A scope's device lookup failed, and no other scope has the device, so its
+/// owner is unknown. The frame is dropped rather than handled as `local`:
+/// guessing could run a paired device's RPCs as the wrong agent.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(super) struct OwnerLookupFailed {
+    /// The scope whose lookup failed (`None` = `local`) and why.
+    pub(super) agent: Option<String>,
+    pub(super) error: String,
+}
+
+/// The agent `channel_id` belongs to, resolved as described above: `None`
+/// for `local`, which is also where a channel no scope knows yet (a handshake
+/// still in flight) is handled.
 ///
 /// # 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`.
+/// [`OwnerLookupFailed`] when no scope has the device and at least one
+/// scope's lookup failed.
 pub(super) async fn owner_of(
     channel_id: &str,
     pending: Option<&PairingSession>,
@@ -59,40 +68,47 @@ pub(super) async fn owner_of(
     }
     // 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(),
-        )
+    let lookups = crate::storage::agents::for_each_scope("device owner", || async {
+        let config = crate::config::rpc::load_config_with_timeout()
+            .await
+            .map_err(|error| format!("load config: {error}"))?;
+        super::store::get_device(&config, channel_id)
+            .map(|device| device.is_some())
+            .map_err(|error| error.to_string())
     })
     .await;
-    if let Some(owner) = found
-        .iter()
-        .find_map(|(agent, has_device)| (*has_device == Some(true)).then(|| agent.clone()))
-    {
+    let owner = decide(lookups)?;
+    if let Some(owner) = &owner {
         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)
+    Ok(owner.flatten())
 }
 
-/// 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;
+/// The owner from each scope's lookup: the first scope that has the device
+/// (`Some(Some(agent))` / `Some(None)` for `local`); `None` when none has it
+/// and every lookup succeeded; an error when none has it and one failed.
+fn decide(
+    lookups: Vec<(Option<String>, Result<bool, String>)>,
+) -> Result<Option<Option<String>>, OwnerLookupFailed> {
+    let mut failure = None;
+    for (agent, lookup) in lookups {
+        match lookup {
+            Ok(true) => return Ok(Some(agent)),
+            Ok(false) => {}
+            Err(error) => {
+                failure.get_or_insert(OwnerLookupFailed { agent, error });
+            }
+        }
+    }
+    match failure {
+        Some(failure) => Err(failure),
+        None => Ok(None),
+    }
+}
 
 #[cfg(test)]
 #[path = "owner_tests.rs"]
diff --git a/crates/openhuman-core/src/security/devices/owner_tests.rs b/crates/openhuman-core/src/security/devices/owner_tests.rs
index 5c5df21dc61..945c3620298 100644
--- a/crates/openhuman-core/src/security/devices/owner_tests.rs
+++ b/crates/openhuman-core/src/security/devices/owner_tests.rs
@@ -83,3 +83,31 @@ async fn a_channel_no_scope_knows_is_local_and_not_remembered() {
     assert_eq!(owner, Ok(None));
     assert_eq!(cached("owner-test-missing"), None);
 }
+
+#[test]
+fn the_first_scope_holding_the_device_owns_it() {
+    let found = decide(vec![
+        (None, Ok(false)),
+        (Some("a".into()), Err("down".into())),
+        (Some("b".into()), Ok(true)),
+    ]);
+    assert_eq!(found, Ok(Some(Some("b".to_string()))));
+    assert_eq!(decide(vec![(None, Ok(true))]), Ok(Some(None)));
+}
+
+#[test]
+fn a_failed_lookup_without_a_match_fails_closed() {
+    let failed = decide(vec![
+        (None, Ok(false)),
+        (Some("a".into()), Err("down".into())),
+    ]);
+    assert_eq!(
+        failed,
+        Err(OwnerLookupFailed {
+            agent: Some("a".to_string()),
+            error: "down".to_string(),
+        })
+    );
+    assert_eq!(decide(vec![(None, Ok(false))]), Ok(None));
+    assert_eq!(decide(Vec::new()), Ok(None));
+}
diff --git a/crates/openhuman-core/src/storage/README.md b/crates/openhuman-core/src/storage/README.md
index d07737d2564..165916b3609 100644
--- a/crates/openhuman-core/src/storage/README.md
+++ b/crates/openhuman-core/src/storage/README.md
@@ -79,21 +79,31 @@ backend installed its records land in that agent's scope. Background work
 runs under the process default context and on its own would only see
 `local`. `storage::agents` closes the gap:
 
-- `registered(context)`: `CoreContext::derive_with` records every agent
-  context it builds (live, by agent id), and the agent id is written to the
-  backend's `local` scope (`storage_agents`) so a restarted process still
-  knows it.
+- Known agents: the live ones are `core::runtime::AgentContextRegistry`'s
+  (embed registers each agent it builds and deregisters it on drop).
+  `AgentContextRegistry::register` also records the agent id in the
+  backend's `local` scope (`storage_agents`, `agents::record`), so a
+  restarted process still knows it; `agents::contexts()` lists both.
 - `for_each_scope(label, step)`: runs `step` for `local`, then under each
   known agent's context — its live one, or the default context acting for it
   (`CoreContext::for_agent`). `for_each_agent` skips `local`.
 - `within_agent(agent, fut)` / `context_for(agent)`: re-enter an agent's
   scope when background work learned whose record it is handling.
+- `find_owner(label, probe)`: the scope (`local` first, then each agent)
+  where a record named only by id lives — for event subscribers, whose
+  events carry ids but no agent.
 
-Users: the cron scheduler (`cron::scheduler::tick_live_agents`), the task-source
-poller, the flows boot sweep and schedule-trigger reconcile, the run reaper,
-and the device tunnel (a paired device's frames run as the agent that paired
-it, `security::devices::owner`). Without a backend these run once, as
-before. In SaaS mode agent ids are not recorded and `local` is skipped;
+Users: the task-source poller, the flows boot sweep and schedule-trigger reconcile, the run reaper,
+the device tunnel (a paired device's frames run as the agent that paired
+it, `security::devices::owner`), and the event subscribers: a flow's
+schedule tick, run digest and dedup settlement run as the flow's owner
+(`flows::bus::owner`); Composio app-event triggers and new connections are
+matched in every scope; a cron job's completion notification is stored with
+the job's owner. Without a backend these run once, as
+before. The cron scheduler visits live agents only
+(`cron::scheduler::tick_live_agents`): an agent's jobs need its live
+context (host tools, prompt) to run, so a recorded agent's jobs wait until
+it is live again; with a backend it no longer needs the agent's `jobs.db`. In SaaS mode agent ids are not recorded and `local` is skipped;
 per-user background work there is `user_agents::background`.
 
 ## Boundaries
diff --git a/crates/openhuman-core/src/storage/agents.rs b/crates/openhuman-core/src/storage/agents.rs
index 30ec2e50fce..80cb1a5e4f8 100644
--- a/crates/openhuman-core/src/storage/agents.rs
+++ b/crates/openhuman-core/src/storage/agents.rs
@@ -9,64 +9,43 @@
 //! the `local` scope. [`for_each_scope`] closes that gap: it runs a step once
 //! for `local` and once under each known agent's context.
 //!
-//! An agent is known when a context is derived for it in this process
-//! ([`registered`], called by `CoreContext::derive_with`) — the live context
-//! is used, with the agent's own configuration — or when an earlier process
-//! did and recorded its id in the backend's `local` scope
-//! (`storage_agents`), so a restart still visits agents the host has not
-//! re-created yet; those are visited under the default context with the
-//! agent swapped in (`CoreContext::for_agent`).
+//! An agent is known when it is live in this process
+//! (`core::runtime::AgentContextRegistry`, which embed agents register on
+//! build) — its own context is used, with its configuration, policy and
+//! tools — or when an earlier process recorded its id in the backend's
+//! `local` scope (`storage_agents`, written by [`record`] when the agent
+//! registers), so a restart still visits agents the host has not re-created
+//! yet; those are visited under the default context acting for them
+//! (`CoreContext::for_agent`).
 //!
-//! Without a backend nothing here changes behavior: the SQLite stores do not
-//! split by agent, so [`for_each_scope`] runs the step once. In SaaS mode the
-//! process has no `local` scope and per-user background work is driven by
+//! Without a backend [`for_each_agent`] visits nothing: the SQLite stores
+//! these loops read do not split by agent. In SaaS mode the process has no
+//! `local` scope and per-user background work is driven by
 //! `user_agents::background`, so agent ids are not recorded and
-//! [`for_each_scope`] visits only live agent contexts.
+//! [`for_each_scope`] skips `local`.
 
 use std::collections::{BTreeMap, HashSet};
 use std::future::Future;
-use std::sync::{Arc, LazyLock, Mutex, Weak};
+use std::sync::{Arc, LazyLock, Mutex};
 
 use serde_json::json;
 use tinystoragedrivers::{CollectionSpec, Precondition, Query, Scope};
 
 use super::{block_on, installed, DocumentStoreExt, StorageBackend};
-use crate::core::runtime::CoreContext;
+use crate::core::runtime::{AgentContextRegistry, CoreContext};
 
 /// The `local`-scope collection recording which agents have their own scope.
 const AGENTS: &str = "storage_agents";
 
-/// Live agent contexts, by agent id.
-///
-/// Several contexts can name one agent (each `derive_with` makes one), so each
-/// agent keeps all of its live contexts, newest last.
-static LIVE: LazyLock<Mutex<BTreeMap<String, Vec<Weak<CoreContext>>>>> =
-    LazyLock::new(Default::default);
-
-/// Agent ids this process has already recorded in the backend.
-static RECORDED: LazyLock<Mutex<HashSet<String>>> = LazyLock::new(Default::default);
+/// `(backend, agent id)` pairs this process has already recorded, the
+/// backend identified by its address (see [`backend_key`]).
+static RECORDED: LazyLock<Mutex<HashSet<(usize, String)>>> = LazyLock::new(Default::default);
 
-/// Records `context` as its agent's live context (when it names one) and
-/// returns it unchanged. `CoreContext::derive_with` passes every derived
-/// context through here.
-pub fn registered(context: Arc<CoreContext>) -> Arc<CoreContext> {
-    if let Some(agent) = context.session_agent() {
-        let mut live = LIVE
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        // Agents come and go; forget the ones whose contexts are all gone so
-        // the registry stays as small as the set of live agents.
-        live.retain(|_, entries| {
-            entries.retain(|entry| entry.strong_count() > 0);
-            !entries.is_empty()
-        });
-        live.entry(agent.to_string())
-            .or_default()
-            .push(Arc::downgrade(&context));
-        drop(live);
-        record(agent);
-    }
-    context
+/// Identifies `backend` within this process. [`reset_recorded`] runs on every
+/// install and clear, so an address reused by a later backend never inherits
+/// an earlier one's records.
+fn backend_key(backend: &Arc<dyn StorageBackend>) -> usize {
+    Arc::as_ptr(backend).cast::<()>() as usize
 }
 
 /// Forgets which agents were recorded, so the next [`record`] writes them to
@@ -79,25 +58,20 @@ pub(super) fn reset_recorded() {
         .clear();
 }
 
-/// Records every live agent in the backend — for agents derived before the
-/// host installed its backend. Called by [`super::install`].
+/// Records every live agent in the backend — for agents registered before
+/// the host installed its backend. Called by [`super::install`].
 pub(super) fn record_live() {
-    let agents: Vec<String> = LIVE
-        .lock()
-        .unwrap_or_else(std::sync::PoisonError::into_inner)
-        .keys()
-        .cloned()
-        .collect();
-    for agent in agents {
+    for (agent, _) in AgentContextRegistry::live() {
         record(&agent);
     }
 }
 
 /// Writes `agent` to the backend's `storage_agents` collection, once per
-/// process. Best effort: a failure is logged and retried on the next
-/// registration, and only costs a restarted process its visits to that
-/// agent until the agent is derived again.
-fn record(agent: &str) {
+/// process. `AgentContextRegistry::register` calls it for every agent. Best
+/// effort: a failure is logged and retried on the next registration, and
+/// only costs a restarted process its visits to that agent until the agent
+/// registers again.
+pub fn record(agent: &str) {
     if crate::core::runtime::mode::is_saas() {
         return;
     }
@@ -109,10 +83,11 @@ fn record(agent: &str) {
 
 /// [`record`] against an explicit `backend`.
 fn record_in(backend: Arc<dyn StorageBackend>, agent: &str) {
+    let key = (backend_key(&backend), agent.to_string());
     if RECORDED
         .lock()
         .unwrap_or_else(std::sync::PoisonError::into_inner)
-        .contains(agent)
+        .contains(&key)
     {
         return;
     }
@@ -128,7 +103,7 @@ fn record_in(backend: Arc<dyn StorageBackend>, agent: &str) {
             RECORDED
                 .lock()
                 .unwrap_or_else(std::sync::PoisonError::into_inner)
-                .insert(agent.to_string());
+                .insert(key);
             tracing::debug!(%agent, "[storage::agents] recorded agent scope");
         }
         Err(error) => {
@@ -151,37 +126,28 @@ fn recorded(backend: Arc<dyn StorageBackend>) -> Vec<String> {
     })
 }
 
-/// The agents [`for_each_scope`] visits, each with the context to visit it
-/// under: its live context when one exists, else `fallback` acting for it.
-fn agent_contexts(fallback: Option<&Arc<CoreContext>>) -> Vec<(String, Arc<CoreContext>)> {
+/// Every agent background work should visit, each with the context to visit
+/// it under: the live ones (`AgentContextRegistry`), then — with a storage
+/// backend installed, outside SaaS mode — every agent recorded by this or an
+/// earlier process, under the current context acting for it.
+pub fn contexts() -> Vec<(String, Arc<CoreContext>)> {
     let backend = if crate::core::runtime::mode::is_saas() {
         None
     } else {
         installed()
     };
-    agent_contexts_in(backend, fallback)
+    contexts_in(backend, CoreContext::current().as_ref())
 }
 
-/// [`agent_contexts`] with the backend whose recorded agents are visited
-/// made explicit (`None` visits live contexts only).
-fn agent_contexts_in(
+/// [`contexts`] with the backend whose recorded agents are visited, and the
+/// context that acts for an agent with no live one, made explicit (`None`
+/// visits live contexts only).
+fn contexts_in(
     backend: Option<Arc<dyn StorageBackend>>,
     fallback: Option<&Arc<CoreContext>>,
 ) -> Vec<(String, Arc<CoreContext>)> {
-    let mut contexts: BTreeMap<String, Arc<CoreContext>> = BTreeMap::new();
-    {
-        let mut live = LIVE
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        live.retain(|agent, entries| {
-            entries.retain(|entry| entry.strong_count() > 0);
-            // The newest context that is still alive acts for the agent.
-            if let Some(context) = entries.iter().rev().find_map(Weak::upgrade) {
-                contexts.insert(agent.clone(), context);
-            }
-            !entries.is_empty()
-        });
-    }
+    let mut contexts: BTreeMap<String, Arc<CoreContext>> =
+        AgentContextRegistry::live().into_iter().collect();
     if let (Some(backend), Some(fallback)) = (backend, fallback) {
         for agent in recorded(backend) {
             contexts
@@ -193,14 +159,17 @@ fn agent_contexts_in(
 }
 
 /// The context to act for `agent` under: its live context when one exists,
-/// else the current context acting for it (`CoreContext::for_agent`).
+/// else — outside SaaS mode — the current context acting for it
+/// (`CoreContext::for_agent`).
 pub fn context_for(agent: &str) -> Option<Arc<CoreContext>> {
-    let live = LIVE
-        .lock()
-        .unwrap_or_else(std::sync::PoisonError::into_inner)
-        .get(agent)
-        .and_then(|entries| entries.iter().rev().find_map(Weak::upgrade));
-    live.or_else(|| CoreContext::current().map(|current| current.for_agent(agent)))
+    AgentContextRegistry::get(agent).or_else(|| {
+        // SaaS acts only through a user's own live context: a copy of the
+        // operator's would carry the wrong configuration.
+        if crate::core::runtime::mode::is_saas() {
+            return None;
+        }
+        CoreContext::current().map(|current| current.for_agent(agent))
+    })
 }
 
 /// Runs `fut` acting for `agent` when there is one — background work that
@@ -213,6 +182,21 @@ pub async fn within_agent<F: Future>(agent: Option<&str>, fut: F) -> F::Output {
     }
 }
 
+/// The scope a record lives in, for background work that holds only its id
+/// (an event naming a flow, a job, a device): the first scope — `local`
+/// first, then each known agent ([`for_each_scope`]) — where `probe` finds
+/// it. `Some(None)` is `local`, `Some(Some(agent))` an agent, `None` nowhere.
+pub async fn find_owner<F, Fut>(label: &str, probe: F) -> Option<Option<String>>
+where
+    F: Fn() -> Fut,
+    Fut: Future<Output = bool>,
+{
+    for_each_scope(label, probe)
+        .await
+        .into_iter()
+        .find_map(|(agent, found)| found.then_some(agent))
+}
+
 /// Runs `step` for every storage scope background work must cover: once
 /// under the current context (the `local` scope, outside SaaS mode), then
 /// once per known agent ([`for_each_agent`]). Each result is returned with
@@ -248,11 +232,46 @@ where
     if installed().is_none() {
         return Vec::new();
     }
-    let fallback = CoreContext::current();
+    visit(label, contexts(), step).await
+}
+
+/// [`for_each_scope`], visiting only live agents (`AgentContextRegistry`):
+/// for work that acts as the agent — runs its flows, fetches with its
+/// connections — and so needs the agent's own configuration and tools, which
+/// a recorded agent's stand-in context (`CoreContext::for_agent`) lacks.
+pub async fn for_each_live_scope<T, F, Fut>(label: &str, step: F) -> Vec<(Option<String>, T)>
+where
+    F: Fn() -> Fut,
+    Fut: Future<Output = T>,
+{
+    let mut results = Vec::new();
+    if !crate::core::runtime::mode::is_saas() {
+        results.push((None, step().await));
+    }
+    if installed().is_none() {
+        return results;
+    }
+    for (agent, value) in visit(label, AgentContextRegistry::live(), step).await {
+        results.push((Some(agent), value));
+    }
+    results
+}
+
+/// Runs `step` under each of `contexts`, building each step inside its
+/// agent's scope, so anything it reads while being set up is the agent's.
+async fn visit<T, F, Fut>(
+    label: &str,
+    contexts: Vec<(String, Arc<CoreContext>)>,
+    step: F,
+) -> Vec<(String, T)>
+where
+    F: Fn() -> Fut,
+    Fut: Future<Output = T>,
+{
     let mut results = Vec::new();
-    for (agent, context) in agent_contexts(fallback.as_ref()) {
+    for (agent, context) in contexts {
         tracing::trace!(%agent, label, "[storage::agents] visiting agent scope");
-        let value = CoreContext::scope(context, step()).await;
+        let value = CoreContext::scope(context, async { step().await }).await;
         results.push((agent, value));
     }
     results
diff --git a/crates/openhuman-core/src/storage/agents_tests.rs b/crates/openhuman-core/src/storage/agents_tests.rs
index 7875a3ec084..801eba9b58e 100644
--- a/crates/openhuman-core/src/storage/agents_tests.rs
+++ b/crates/openhuman-core/src/storage/agents_tests.rs
@@ -1,116 +1,78 @@
 use super::*;
 use crate::core::runtime::{ContextOverlay, DomainSet};
 
-fn agent_context(agent: &str) -> Arc<CoreContext> {
-    CoreContext::for_test(DomainSet::full(), None).derive_with(
+/// A context derived for `agent` and registered as that agent's live one,
+/// the way an embed agent is. Each test uses its own agent ids, so tests
+/// sharing the process-wide registry never see each other's agents.
+fn live_agent(agent: &str) -> Arc<CoreContext> {
+    let context = CoreContext::for_test(DomainSet::full(), None).derive_with(
         ContextOverlay::new(
             crate::config::Config::default(),
             DomainSet::full(),
             Default::default(),
         )
         .session_agent(agent),
-    )
+    );
+    AgentContextRegistry::register(agent, &context);
+    context
 }
 
-/// Whether `agent` stops being live. Another test's concurrent walk of the
-/// registry can hold a context for an instant, so allow it to let go.
-fn eventually_gone(agent: &str) -> bool {
-    (0..100).any(|_| {
-        let gone = !live_agents().contains(&agent.to_string());
-        if !gone {
-            std::thread::sleep(std::time::Duration::from_millis(10));
-        }
-        gone
-    })
+/// Serializes the tests that depend on, or clear, the process-wide record
+/// cache.
+static CACHE: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+fn memory_backend() -> Arc<dyn StorageBackend> {
+    Arc::new(crate::storage::MemoryStorage::new())
 }
 
-fn live_agents() -> Vec<String> {
-    agent_contexts(None).into_iter().map(|(id, _)| id).collect()
+fn visited(
+    backend: Option<Arc<dyn StorageBackend>>,
+    fallback: Option<&Arc<CoreContext>>,
+) -> Vec<String> {
+    contexts_in(backend, fallback)
+        .into_iter()
+        .map(|(id, _)| id)
+        .collect()
 }
 
 #[test]
-fn deriving_an_agent_context_registers_it_until_dropped() {
-    let context = agent_context("agents-test-live");
-    assert!(live_agents().contains(&"agents-test-live".to_string()));
-    let (_, found) = agent_contexts(None)
+fn a_registered_agent_is_visited_through_its_live_context() {
+    let context = live_agent("agents-test-live");
+    let (_, found) = contexts_in(None, None)
         .into_iter()
         .find(|(id, _)| id == "agents-test-live")
-        .unwrap();
+        .expect("a live agent is visited");
     assert!(Arc::ptr_eq(&found, &context), "the live context is used");
+    assert!(Arc::ptr_eq(
+        &context_for("agents-test-live").unwrap(),
+        &context
+    ));
+    assert!(AgentContextRegistry::deregister(
+        "agents-test-live",
+        &context
+    ));
     drop((found, context));
-    assert!(eventually_gone("agents-test-live"));
-}
-
-#[test]
-fn a_context_without_an_agent_is_not_registered() {
-    let before = live_agents().len();
-    let _plain = CoreContext::for_test(DomainSet::full(), None);
-    assert_eq!(live_agents().len(), before);
+    assert!(!visited(None, None).contains(&"agents-test-live".to_string()));
 }
 
 #[test]
 fn for_agent_swaps_only_the_agent() {
-    let parent = agent_context("agents-test-parent");
+    let parent = live_agent("agents-test-parent");
     let child = parent.for_agent("agents-test-other");
     assert_eq!(child.session_agent(), Some("agents-test-other"));
     assert_eq!(parent.session_agent(), Some("agents-test-parent"));
-}
-
-#[tokio::test]
-async fn without_a_backend_only_the_local_scope_runs() {
-    // The lib test binary never installs a backend into the process slot.
-    if installed().is_some() {
-        return;
-    }
-    let _agent = agent_context("agents-test-unvisited");
-    let runs = for_each_scope("test", || async {
-        CoreContext::current().and_then(|c| c.session_agent().map(str::to_string))
-    })
-    .await;
-    assert_eq!(runs.len(), 1);
-    assert_eq!(runs[0].0, None);
-    assert!(for_each_agent("test", || async {}).await.is_empty());
-}
-
-#[test]
-fn a_dropped_sibling_context_does_not_hide_a_live_one() {
-    let first = agent_context("agents-test-siblings");
-    let second = agent_context("agents-test-siblings");
-    drop(second);
-    let (_, found) = agent_contexts(None)
-        .into_iter()
-        .find(|(id, _)| id == "agents-test-siblings")
-        .expect("the first context is still live");
-    assert!(Arc::ptr_eq(&found, &first));
-    assert!(context_for("agents-test-siblings").is_some());
-    drop((found, first));
-    assert!(eventually_gone("agents-test-siblings"));
-}
-
-#[test]
-fn reset_recorded_forgets_what_was_recorded() {
-    RECORDED
-        .lock()
-        .unwrap()
-        .insert("agents-test-recorded".into());
-    reset_recorded();
-    assert!(!RECORDED.lock().unwrap().contains("agents-test-recorded"));
-}
-
-fn memory_backend() -> Arc<dyn StorageBackend> {
-    Arc::new(crate::storage::MemoryStorage::new())
+    AgentContextRegistry::deregister("agents-test-parent", &parent);
 }
 
 #[test]
 fn a_recorded_agent_is_visited_through_the_fallback_context() {
     let backend = memory_backend();
-    reset_recorded();
     record_in(Arc::clone(&backend), "agents-test-recorded-only");
     assert!(recorded(Arc::clone(&backend)).contains(&"agents-test-recorded-only".to_string()));
 
     let fallback = CoreContext::for_test(DomainSet::full(), None);
-    let visited = agent_contexts_in(Some(backend), Some(&fallback));
-    let (_, context) = visited
+    let found = contexts_in(Some(backend), Some(&fallback));
+    let (_, context) = found
         .iter()
         .find(|(id, _)| id == "agents-test-recorded-only")
         .expect("the recorded agent is visited");
@@ -118,41 +80,86 @@ fn a_recorded_agent_is_visited_through_the_fallback_context() {
 }
 
 #[test]
-fn recording_is_skipped_for_an_agent_already_recorded_to_that_backend() {
+fn without_a_fallback_recorded_agents_are_not_visited() {
+    let backend = memory_backend();
+    record_in(Arc::clone(&backend), "agents-test-no-fallback");
+    assert!(!visited(Some(backend), None).contains(&"agents-test-no-fallback".to_string()));
+}
+
+#[test]
+fn an_agent_is_recorded_once_per_backend() {
+    let _serial = CACHE
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     let first = memory_backend();
-    reset_recorded();
     record_in(Arc::clone(&first), "agents-test-once");
-    // A second backend is only populated once the cache is reset.
+    // Removing the record behind the cache's back shows the second call is
+    // skipped: the cache says this backend already has it.
+    let docs = Arc::clone(first.for_scope(&Scope::local()).unwrap().documents());
+    block_on(async move {
+        docs.delete(AGENTS, "agents-test-once", Precondition::None)
+            .await
+            .map(|_| ())
+    })
+    .unwrap();
+    record_in(Arc::clone(&first), "agents-test-once");
+    assert!(recorded(first).is_empty());
+    // Another backend is a separate record.
     let second = memory_backend();
     record_in(Arc::clone(&second), "agents-test-once");
-    assert!(recorded(Arc::clone(&second)).is_empty());
-    reset_recorded();
-    record_in(Arc::clone(&second), "agents-test-once");
     assert_eq!(recorded(second), vec!["agents-test-once".to_string()]);
 }
 
 #[test]
-fn without_a_fallback_only_live_contexts_are_visited() {
+fn reset_recorded_empties_the_cache() {
+    let _serial = CACHE
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     let backend = memory_backend();
+    record_in(Arc::clone(&backend), "agents-test-reset");
     reset_recorded();
-    record_in(Arc::clone(&backend), "agents-test-no-fallback");
-    assert!(!agent_contexts_in(Some(backend), None)
-        .iter()
-        .any(|(id, _)| id == "agents-test-no-fallback"));
+    assert!(!RECORDED
+        .lock()
+        .unwrap()
+        .contains(&(backend_key(&backend), "agents-test-reset".to_string())));
 }
 
-#[test]
-fn registering_forgets_agents_whose_contexts_are_all_gone() {
-    drop(agent_context("agents-test-gone"));
-    // Each registration prunes; a concurrent walk of the registry may hold the
-    // dropped context for an instant, so give it a few registrations.
-    let forgotten = (0..100).any(|_| {
-        let _other = agent_context("agents-test-other-live");
-        let gone = !LIVE.lock().unwrap().contains_key("agents-test-gone");
-        if !gone {
-            std::thread::sleep(std::time::Duration::from_millis(10));
-        }
-        gone
-    });
-    assert!(forgotten);
+#[tokio::test]
+async fn without_a_backend_only_the_local_scope_runs() {
+    // The lib test binary never installs a backend into the process slot.
+    if installed().is_some() {
+        return;
+    }
+    let agent = live_agent("agents-test-unvisited");
+    let runs = for_each_scope("test", || async {
+        CoreContext::current().and_then(|c| c.session_agent().map(str::to_string))
+    })
+    .await;
+    assert_eq!(runs.len(), 1);
+    assert_eq!(runs[0].0, None);
+    assert!(for_each_agent("test", || async {}).await.is_empty());
+    assert_eq!(
+        find_owner("test", || async { false }).await,
+        None,
+        "nothing anywhere"
+    );
+    assert_eq!(find_owner("test", || async { true }).await, Some(None));
+    let inside = within_agent(Some("agents-test-unvisited"), async {
+        CoreContext::current().and_then(|c| c.session_agent().map(str::to_string))
+    })
+    .await;
+    assert_eq!(inside.as_deref(), Some("agents-test-unvisited"));
+    assert_eq!(within_agent(None, async { 7 }).await, 7);
+    AgentContextRegistry::deregister("agents-test-unvisited", &agent);
+}
+
+#[tokio::test]
+async fn without_a_backend_the_live_pass_runs_only_local() {
+    if installed().is_some() {
+        return;
+    }
+    let agent = live_agent("agents-test-live-pass");
+    let runs = for_each_live_scope("test", || async { 1 }).await;
+    assert_eq!(runs, vec![(None, 1)]);
+    AgentContextRegistry::deregister("agents-test-live-pass", &agent);
 }
diff --git a/scripts/ci/saas-ambient-baseline.json b/scripts/ci/saas-ambient-baseline.json
index 3ef10d6d6a8..ff83e2fc0d0 100644
--- a/scripts/ci/saas-ambient-baseline.json
+++ b/scripts/ci/saas-ambient-baseline.json
@@ -468,13 +468,6 @@
     "text": "tokio::task::spawn_blocking(move || load_app_session_profile(&config_for_profile))",
     "occurrence": 1
   },
-  {
-    "rule": "bare-spawn",
-    "path": "crates/openhuman-core/src/flows/bus/trigger.rs",
-    "line": 202,
-    "text": "tokio::spawn(async move {",
-    "occurrence": 1
-  },
   {
     "rule": "bare-spawn",
     "path": "crates/openhuman-core/src/flows/ops/run.rs",
@@ -650,13 +643,6 @@
     "text": "tokio::spawn(async move {",
     "occurrence": 1
   },
-  {
-    "rule": "bare-spawn",
-    "path": "crates/openhuman-core/src/integrations/task_sources/bus.rs",
-    "line": 91,
-    "text": "tokio::spawn(async move {",
-    "occurrence": 1
-  },
   {
     "rule": "bare-spawn",
     "path": "crates/openhuman-core/src/integrations/test_support_backend.rs",
diff --git a/tests/storage_scope_e2e.rs b/tests/storage_scope_e2e.rs
index a2219a0e4fa..71c6ba29e5c 100644
--- a/tests/storage_scope_e2e.rs
+++ b/tests/storage_scope_e2e.rs
@@ -14,10 +14,10 @@ use std::sync::Arc;
 
 use openhuman_core::config::Config;
 use openhuman_core::core::runtime::{
-    ContextOverlay, CoreBuilder, CoreContext, DomainSet, ServiceSet,
+    AgentContextRegistry, ContextOverlay, CoreBuilder, CoreContext, DomainSet, ServiceSet,
 };
 use openhuman_core::cron::{self, Schedule};
-use openhuman_core::storage::agents::for_each_scope;
+use openhuman_core::storage::agents::{find_owner, for_each_scope, within_agent};
 use openhuman_core::HostKind;
 
 fn job_names(config: &Config) -> Vec<String> {
@@ -49,6 +49,9 @@ async fn background_work_visits_every_agent_scope() {
         ContextOverlay::new(config.clone(), DomainSet::none(), Default::default())
             .session_agent("agent-e2e"),
     );
+    // How a host's agent becomes known to background work (embed does this
+    // when it builds an agent).
+    AgentContextRegistry::register("agent-e2e", &agent);
     // The store calls block on storage's own bridge thread, so they are
     // made straight from the agent's task, where its context is in scope.
     CoreContext::scope(Arc::clone(&agent), async {
@@ -72,7 +75,19 @@ async fn background_work_visits_every_agent_scope() {
         "{live:?}"
     );
 
+    // An event naming only the job resolves to its owner, and handling it
+    // there sees the job.
+    let job_id = within_agent(Some("agent-e2e"), async {
+        cron::list_jobs(&config).unwrap()[0].id.clone()
+    })
+    .await;
+    let owner = find_owner("e2e", || async { cron::get_job(&config, &job_id).is_ok() }).await;
+    assert_eq!(owner, Some(Some("agent-e2e".to_string())));
+    let missing = find_owner("e2e", || async { cron::get_job(&config, "nope").is_ok() }).await;
+    assert_eq!(missing, None);
+
     // … and, once the agent is gone, through the id the backend recorded.
+    assert!(AgentContextRegistry::deregister("agent-e2e", &agent));
     drop(agent);
     let recorded = for_each_scope("e2e", || async { job_names(&config) }).await;
     assert!(