From 0d95879a255885b24f130a5cf76ca3e384443165 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 18:13:25 +0300
Subject: [PATCH 01/17] feat(flows): run flow events under their owning agent
 scope

Flow trigger, dedup commit, and run digest subscribers now resolve the
agent that owns a flow and execute their work inside that agent's storage
scope, so background runs read and write the same data as the agent they
belong to. A new find_owner helper locates the scope holding a record by
probing each scope, and app event triggers are matched across every scope
so each run starts under its flow's owner.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/flows/bus.rs        |  1 +
 .../src/flows/bus/dedup_commit.rs             |  8 ++-
 crates/openhuman-core/src/flows/bus/owner.rs  | 64 +++++++++++++++++++
 .../src/flows/bus/run_digest.rs               |  8 ++-
 .../openhuman-core/src/flows/bus/trigger.rs   | 22 ++++++-
 crates/openhuman-core/src/storage/agents.rs   | 15 +++++
 6 files changed, 113 insertions(+), 5 deletions(-)
 create mode 100644 crates/openhuman-core/src/flows/bus/owner.rs

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..abc6c2d7bd4
--- /dev/null
+++ b/crates/openhuman-core/src/flows/bus/owner.rs
@@ -0,0 +1,64 @@
+//! 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> {
+    if crate::storage::installed().is_none() {
+        return None;
+    }
+    if let Some(owner) = OWNERS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .get(flow_id)
+        .cloned()
+    {
+        return owner;
+    }
+    let found = crate::storage::agents::find_owner("flow owner", || async {
+        matches!(crate::flows::store::get_flow(config, flow_id), Ok(Some(_)))
+    })
+    .await;
+    let Some(owner) = found else {
+        return None;
+    };
+    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).
+#[cfg_attr(not(test), allow(dead_code))]
+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/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..6c4cbaffa25 100644
--- a/crates/openhuman-core/src/flows/bus/trigger.rs
+++ b/crates/openhuman-core/src/flows/bus/trigger.rs
@@ -199,7 +199,8 @@ impl FlowTriggerSubscriber {
         };
 
         let config = self.config.clone();
-        tokio::spawn(async move {
+        // 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 +248,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_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/storage/agents.rs b/crates/openhuman-core/src/storage/agents.rs
index 1fbc6eaf452..0fc78814aaa 100644
--- a/crates/openhuman-core/src/storage/agents.rs
+++ b/crates/openhuman-core/src/storage/agents.rs
@@ -171,6 +171,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

From 2eacd3624e8e01f95eb4ca9b55320c48cf354ec4 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 18:14:26 +0300
Subject: [PATCH 02/17] feat(task_sources): fire connection fetches per storage
 scope

Connection-created events now iterate every storage scope and run each
scope's enabled sources under its own agent context, so sources belonging
to other scopes are no longer skipped. The spawned fetch uses the scoped
runtime so it retains that agent context.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../src/integrations/task_sources/bus.rs      | 22 ++++++++++++++++++-
 1 file changed, 21 insertions(+), 1 deletion(-)

diff --git a/crates/openhuman-core/src/integrations/task_sources/bus.rs b/crates/openhuman-core/src/integrations/task_sources/bus.rs
index d380df792d3..07f1225df2e 100644
--- a/crates/openhuman-core/src/integrations/task_sources/bus.rs
+++ b/crates/openhuman-core/src/integrations/task_sources/bus.rs
@@ -60,6 +60,25 @@ impl EventHandler<DomainEvent> for TaskSourcesConnectionSubscriber {
             return;
         }
 
+        // Every scope may hold sources for this toolkit (`crate::storage`):
+        // fire each scope's under its own agent.
+        crate::storage::agents::for_each_scope("task_sources connection", || {
+            fire_for_connection(&config, provider, toolkit, connection_id)
+        })
+        .await;
+    }
+}
+
+/// 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 config = config.clone();
         let sources = match store::list_sources(&config) {
             Ok(sources) => sources,
             Err(e) => {
@@ -88,7 +107,8 @@ impl EventHandler<DomainEvent> for TaskSourcesConnectionSubscriber {
             // pattern as the periodic poll). Each fetch captures its own
             // owned config + source.
             let config = config.clone();
-            tokio::spawn(async move {
+            // 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;
             });

From a5c78f61b2868f6f2f492d3cfd7cf123d7ab85e3 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 18:14:38 +0300
Subject: [PATCH 03/17] refactor(flows): collapse owner map initializer onto
 one line

Reformatted the static owner map declaration so its initializer fits on a single line, with no change in behaviour.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/flows/bus/owner.rs  |  3 +--
 .../src/flows/bus/owner_tests.rs              | 25 +++++++++++++++++++
 2 files changed, 26 insertions(+), 2 deletions(-)
 create mode 100644 crates/openhuman-core/src/flows/bus/owner_tests.rs

diff --git a/crates/openhuman-core/src/flows/bus/owner.rs b/crates/openhuman-core/src/flows/bus/owner.rs
index abc6c2d7bd4..702a3bc228e 100644
--- a/crates/openhuman-core/src/flows/bus/owner.rs
+++ b/crates/openhuman-core/src/flows/bus/owner.rs
@@ -13,8 +13,7 @@ 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);
+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).
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..91238d9b591
--- /dev/null
+++ b/crates/openhuman-core/src/flows/bus/owner_tests.rs
@@ -0,0 +1,25 @@
+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"));
+}

From 08e67218975280669c45aca4b19eaf30556eef00 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 18:15:36 +0300
Subject: [PATCH 04/17] refactor(task_sources): flatten redundant block in
 connection fetch

Removed an unnecessary nested block in the connection-created handler so the
source listing and per-source fetch loop sit at the function's top level. The
behaviour is unchanged; the extra indentation was a leftover from an earlier
structure.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../src/integrations/task_sources/bus.rs      | 63 +++++++++----------
 1 file changed, 30 insertions(+), 33 deletions(-)

diff --git a/crates/openhuman-core/src/integrations/task_sources/bus.rs b/crates/openhuman-core/src/integrations/task_sources/bus.rs
index 07f1225df2e..4dfc92e51dc 100644
--- a/crates/openhuman-core/src/integrations/task_sources/bus.rs
+++ b/crates/openhuman-core/src/integrations/task_sources/bus.rs
@@ -77,42 +77,39 @@ async fn fire_for_connection(
     toolkit: &str,
     connection_id: &str,
 ) {
-    {
-        let config = config.clone();
-        let sources = match store::list_sources(&config) {
-            Ok(sources) => sources,
-            Err(e) => {
-                tracing::debug!(error = %e, "[task_sources:bus] list_sources failed, skipping");
-                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;
+        }
+    };
 
-        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;
-                }
+    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();
-            // 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;
-            });
         }
+        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;
+        });
     }
 }
 

From e90d95732b7454991551e72434be87bfaad9b254 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 18:16:37 +0300
Subject: [PATCH 05/17] fix(notifications): store cron completion notifications
 under the job's owning agent

Cron job completion notifications were persisted in whichever storage
scope the bridge was registered with, so they could land outside the
agent that owns the job. The bridge now resolves the job's owner and
stores the notification within that agent's scope, falling back to
local when no owner is found.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../src/desktop/notifications/bus.rs          | 28 ++++++++++++++++++-
 .../src/integrations/task_sources/bus.rs      |  2 +-
 2 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/crates/openhuman-core/src/desktop/notifications/bus.rs b/crates/openhuman-core/src/desktop/notifications/bus.rs
index 242fbdb767b..d8ce148038f 100644
--- a/crates/openhuman-core/src/desktop/notifications/bus.rs
+++ b/crates/openhuman-core/src/desktop/notifications/bus.rs
@@ -469,6 +469,24 @@ 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;
+    };
+    if crate::storage::installed().is_none() {
+        return None;
+    }
+    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 +509,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/integrations/task_sources/bus.rs b/crates/openhuman-core/src/integrations/task_sources/bus.rs
index 4dfc92e51dc..7e7435b98cd 100644
--- a/crates/openhuman-core/src/integrations/task_sources/bus.rs
+++ b/crates/openhuman-core/src/integrations/task_sources/bus.rs
@@ -77,7 +77,7 @@ async fn fire_for_connection(
     toolkit: &str,
     connection_id: &str,
 ) {
-    let sources = match store::list_sources(&config) {
+    let sources = match store::list_sources(config) {
         Ok(sources) => sources,
         Err(e) => {
             tracing::debug!(error = %e, "[task_sources:bus] list_sources failed, skipping");

From 6fa9ba074bc0056547797d74c3e9b6fdf3fad3ca Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 18:22:15 +0300
Subject: [PATCH 06/17] test(storage): cover owner resolution for scoped jobs

Extend the storage scope e2e test to assert that an event naming only a job
resolves to its owning agent and that an unknown job resolves to nothing. The
ambient baseline drops the two bare-spawn entries that no longer exist.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 scripts/ci/saas-ambient-baseline.json | 14 --------------
 tests/storage_scope_e2e.rs            | 13 ++++++++++++-
 2 files changed, 12 insertions(+), 15 deletions(-)

diff --git a/scripts/ci/saas-ambient-baseline.json b/scripts/ci/saas-ambient-baseline.json
index f3669beb607..39b80247532 100644
--- a/scripts/ci/saas-ambient-baseline.json
+++ b/scripts/ci/saas-ambient-baseline.json
@@ -601,13 +601,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",
@@ -783,13 +776,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..20b064995eb 100644
--- a/tests/storage_scope_e2e.rs
+++ b/tests/storage_scope_e2e.rs
@@ -17,7 +17,7 @@ use openhuman_core::core::runtime::{
     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> {
@@ -72,6 +72,17 @@ 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.
     drop(agent);
     let recorded = for_each_scope("e2e", || async { job_names(&config) }).await;

From 04e37933ddd4ae8ff48fee92fbb9ad091f532156 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 18:27:50 +0300
Subject: [PATCH 07/17] refactor(core): use ? operator for early returns in
 owner lookups

Replace the explicit is_none and let-else early returns in the notification
and flow owner resolvers with the ? operator, which is equivalent here and
reads more directly.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/desktop/notifications/bus.rs | 5 ++---
 crates/openhuman-core/src/flows/bus/owner.rs           | 9 +++------
 2 files changed, 5 insertions(+), 9 deletions(-)

diff --git a/crates/openhuman-core/src/desktop/notifications/bus.rs b/crates/openhuman-core/src/desktop/notifications/bus.rs
index d8ce148038f..4fc100db804 100644
--- a/crates/openhuman-core/src/desktop/notifications/bus.rs
+++ b/crates/openhuman-core/src/desktop/notifications/bus.rs
@@ -477,9 +477,8 @@ async fn event_owner(config: &crate::config::Config, event: &DomainEvent) -> Opt
     let DomainEvent::CronJobCompleted { job_id, .. } = event else {
         return None;
     };
-    if crate::storage::installed().is_none() {
-        return None;
-    }
+    // No backend: every record is `local`.
+    crate::storage::installed()?;
     crate::storage::agents::find_owner("notification owner", || async {
         crate::cron::get_job(config, job_id).is_ok()
     })
diff --git a/crates/openhuman-core/src/flows/bus/owner.rs b/crates/openhuman-core/src/flows/bus/owner.rs
index 702a3bc228e..34dfe1f83e2 100644
--- a/crates/openhuman-core/src/flows/bus/owner.rs
+++ b/crates/openhuman-core/src/flows/bus/owner.rs
@@ -18,9 +18,8 @@ static OWNERS: LazyLock<Mutex<HashMap<String, Option<String>>>> = LazyLock::new(
 /// 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> {
-    if crate::storage::installed().is_none() {
-        return None;
-    }
+    // No backend: every record is `local`.
+    crate::storage::installed()?;
     if let Some(owner) = OWNERS
         .lock()
         .unwrap_or_else(std::sync::PoisonError::into_inner)
@@ -33,9 +32,7 @@ pub(super) async fn flow_owner(config: &Config, flow_id: &str) -> Option<String>
         matches!(crate::flows::store::get_flow(config, flow_id), Ok(Some(_)))
     })
     .await;
-    let Some(owner) = found else {
-        return None;
-    };
+    let owner = found?;
     tracing::debug!(
         target: "flows",
         %flow_id,

From 661cadf815a8e326b4992e81926234c99531df9e Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 18:30:16 +0300
Subject: [PATCH 08/17] docs(storage): document owner lookup and event
 subscriber scopes

Adds `find_owner` to the storage scope API list, which resolves the scope
holding a record known only by id so event subscribers can act on events
that carry ids but no agent. The users section now also covers the event
subscribers: flow schedule ticks, run digests and dedup settlement run as
the flow's owner, Composio app-event triggers and new connections are
matched in every scope, and cron completion notifications are stored with

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/storage/README.md | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/crates/openhuman-core/src/storage/README.md b/crates/openhuman-core/src/storage/README.md
index f659c2bcf35..3b9cb90136d 100644
--- a/crates/openhuman-core/src/storage/README.md
+++ b/crates/openhuman-core/src/storage/README.md
@@ -88,11 +88,18 @@ runs under the process default context and on its own would only see
   (`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_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
+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. In SaaS mode agent ids are not recorded and `local` is skipped;
 per-user background work there is `user_agents::background`.
 

From 099446a8dcf0d2feccb579b1a8a215425e06cb43 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 19:57:39 +0300
Subject: [PATCH 09/17] fix(storage): restrict agent context fallback in saas
 mode

Agent context lookup no longer falls back to a copy of the operator's
context when running in SaaS mode, since that copy would carry the wrong
configuration. A new for_each_live_scope helper visits only agents with a
live context, and the flow trigger, task source bus, and periodic loops
now use it so their work runs with each agent's own configuration and
tools.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../openhuman-core/src/flows/bus/trigger.rs   |  2 +-
 .../src/integrations/task_sources/bus.rs      |  2 +-
 .../src/integrations/task_sources/periodic.rs |  2 +-
 crates/openhuman-core/src/storage/agents.rs   | 53 +++++++++++++++++--
 4 files changed, 51 insertions(+), 8 deletions(-)

diff --git a/crates/openhuman-core/src/flows/bus/trigger.rs b/crates/openhuman-core/src/flows/bus/trigger.rs
index 6c4cbaffa25..595b7191b40 100644
--- a/crates/openhuman-core/src/flows/bus/trigger.rs
+++ b/crates/openhuman-core/src/flows/bus/trigger.rs
@@ -265,7 +265,7 @@ impl EventHandler<DomainEvent> for FlowTriggerSubscriber {
                 payload,
                 ..
             } => {
-                crate::storage::agents::for_each_scope("flows app_event", || {
+                crate::storage::agents::for_each_live_scope("flows app_event", || {
                     self.handle_app_event(toolkit, trigger, payload)
                 })
                 .await;
diff --git a/crates/openhuman-core/src/integrations/task_sources/bus.rs b/crates/openhuman-core/src/integrations/task_sources/bus.rs
index 7e7435b98cd..b396f56cddf 100644
--- a/crates/openhuman-core/src/integrations/task_sources/bus.rs
+++ b/crates/openhuman-core/src/integrations/task_sources/bus.rs
@@ -62,7 +62,7 @@ impl EventHandler<DomainEvent> for TaskSourcesConnectionSubscriber {
 
         // Every scope may hold sources for this toolkit (`crate::storage`):
         // fire each scope's under its own agent.
-        crate::storage::agents::for_each_scope("task_sources connection", || {
+        crate::storage::agents::for_each_live_scope("task_sources connection", || {
             fire_for_connection(&config, provider, toolkit, connection_id)
         })
         .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/storage/agents.rs b/crates/openhuman-core/src/storage/agents.rs
index b96b31ffa94..80cb1a5e4f8 100644
--- a/crates/openhuman-core/src/storage/agents.rs
+++ b/crates/openhuman-core/src/storage/agents.rs
@@ -159,10 +159,17 @@ fn 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>> {
-    AgentContextRegistry::get(agent)
-        .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
@@ -225,10 +232,46 @@ where
     if installed().is_none() {
         return Vec::new();
     }
+    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 contexts() {
+    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

From ec48a6936c8d91e0b09be7461ae0d7ffbd079147 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 19:59:10 +0300
Subject: [PATCH 10/17] fix(flows): revalidate cached flow owners and scope
 trigger config

Cached flow owners are now re-checked against the acting agent's scope before
being reused, so a reused flow id or a moved flow no longer resolves to a stale
owner. Trigger handlers also resolve their config from the acting agent's
context when one is present, ensuring runs use the right provider, access
policy and action directory.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/flows/bus/owner.rs  | 18 +++++++++++-----
 .../openhuman-core/src/flows/bus/trigger.rs   | 21 ++++++++++++++++---
 2 files changed, 31 insertions(+), 8 deletions(-)

diff --git a/crates/openhuman-core/src/flows/bus/owner.rs b/crates/openhuman-core/src/flows/bus/owner.rs
index 34dfe1f83e2..ea695c029a4 100644
--- a/crates/openhuman-core/src/flows/bus/owner.rs
+++ b/crates/openhuman-core/src/flows/bus/owner.rs
@@ -20,13 +20,22 @@ static OWNERS: LazyLock<Mutex<HashMap<String, Option<String>>>> = LazyLock::new(
 pub(super) async fn flow_owner(config: &Config, flow_id: &str) -> Option<String> {
     // No backend: every record is `local`.
     crate::storage::installed()?;
-    if let Some(owner) = OWNERS
+    let cached = OWNERS
         .lock()
         .unwrap_or_else(std::sync::PoisonError::into_inner)
         .get(flow_id)
-        .cloned()
-    {
-        return owner;
+        .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(_)))
@@ -47,7 +56,6 @@ pub(super) async fn flow_owner(config: &Config, flow_id: &str) -> Option<String>
 }
 
 /// Forgets a cached owner (a removed flow's id may be reused).
-#[cfg_attr(not(test), allow(dead_code))]
 pub(super) fn forget(flow_id: &str) {
     OWNERS
         .lock()
diff --git a/crates/openhuman-core/src/flows/bus/trigger.rs b/crates/openhuman-core/src/flows/bus/trigger.rs
index 595b7191b40..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,7 +213,7 @@ impl FlowTriggerSubscriber {
             return;
         };
 
-        let config = self.config.clone();
+        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

From 403445daa292a7a1ad7019da2db3141894a953ac Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 20:01:24 +0300
Subject: [PATCH 11/17] feat(cron): track job owner for completion
 notifications

Record the owning agent when a cron job finishes so completion
notifications can be attributed correctly. One-shot jobs are deleted
before the completion event is published, so the scheduler now notes the
owner and the notification bus consumes it, falling back to the existing
lookup when no note is present.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../src/cron/completion_owner.rs              | 49 +++++++++++++++++++
 .../src/cron/completion_owner_tests.rs        | 26 ++++++++++
 crates/openhuman-core/src/cron/mod.rs         |  1 +
 crates/openhuman-core/src/cron/scheduler.rs   |  3 ++
 .../src/desktop/notifications/bus.rs          |  7 +++
 5 files changed, 86 insertions(+)
 create mode 100644 crates/openhuman-core/src/cron/completion_owner.rs
 create mode 100644 crates/openhuman-core/src/cron/completion_owner_tests.rs

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 8ef1cba3006..5c64479da1e 100644
--- a/crates/openhuman-core/src/cron/scheduler.rs
+++ b/crates/openhuman-core/src/cron/scheduler.rs
@@ -263,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/desktop/notifications/bus.rs b/crates/openhuman-core/src/desktop/notifications/bus.rs
index 4fc100db804..c81d7c82c20 100644
--- a/crates/openhuman-core/src/desktop/notifications/bus.rs
+++ b/crates/openhuman-core/src/desktop/notifications/bus.rs
@@ -477,8 +477,15 @@ async fn event_owner(config: &crate::config::Config, event: &DomainEvent) -> Opt
     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()
     })

From a366d293ae062dabbb4c04562393a7584b9f538a Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 20:01:58 +0300
Subject: [PATCH 12/17] test(flows): cover agent-scoped trigger config and
 local live pass

Add a test asserting that a trigger handled inside an agent scope picks up
the agent's config rather than the registered one, and a test confirming
that without a backend the live pass only runs the local scope.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/flows/bus_tests.rs  | 28 +++++++++++++++++++
 .../src/storage/agents_tests.rs               | 11 ++++++++
 2 files changed, 39 insertions(+)

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/storage/agents_tests.rs b/crates/openhuman-core/src/storage/agents_tests.rs
index c6b5bcc7953..801eba9b58e 100644
--- a/crates/openhuman-core/src/storage/agents_tests.rs
+++ b/crates/openhuman-core/src/storage/agents_tests.rs
@@ -152,3 +152,14 @@ async fn without_a_backend_only_the_local_scope_runs() {
     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);
+}

From b485fceb877f8507b043b8a9b0c716f0f6ca36aa Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 20:13:47 +0300
Subject: [PATCH 13/17] refactor(flows): extract owner resolution into a helper

Split the cache and lookup logic out of flow_owner into a resolve helper so the
owner lookup can be reused across scopes.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/flows/bus/owner.rs | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/crates/openhuman-core/src/flows/bus/owner.rs b/crates/openhuman-core/src/flows/bus/owner.rs
index ea695c029a4..fc99e206947 100644
--- a/crates/openhuman-core/src/flows/bus/owner.rs
+++ b/crates/openhuman-core/src/flows/bus/owner.rs
@@ -20,6 +20,11 @@ static OWNERS: LazyLock<Mutex<HashMap<String, Option<String>>>> = LazyLock::new(
 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)

From e7fd961d81277634a77c768c259b8ad29cb06016 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 20:14:05 +0300
Subject: [PATCH 14/17] test(flows): cover owner cache invalidation when a flow
 is removed

Add a test asserting that a cached owner entry is dropped once its scope
no longer contains the flow, and that repeated lookups keep serving the
cached answer while the flow is still present.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../src/flows/bus/owner_tests.rs              | 42 +++++++++++++++++++
 1 file changed, 42 insertions(+)

diff --git a/crates/openhuman-core/src/flows/bus/owner_tests.rs b/crates/openhuman-core/src/flows/bus/owner_tests.rs
index 91238d9b591..0cdc426de8c 100644
--- a/crates/openhuman-core/src/flows/bus/owner_tests.rs
+++ b/crates/openhuman-core/src/flows/bus/owner_tests.rs
@@ -23,3 +23,45 @@ fn a_forgotten_owner_is_resolved_again() {
     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"));
+}

From cb9e4ea57dd4a9fb80501fa04a46a9829f924a82 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 20:14:50 +0300
Subject: [PATCH 15/17] chore: files changed
 crates/openhuman-core/src/integrations/task_sources/bus.rs,crates/openhuman-cor

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../src/integrations/task_sources/bus.rs      |  4 ++
 .../integrations/task_sources/bus_tests.rs    | 61 +++++++++++++++++++
 2 files changed, 65 insertions(+)
 create mode 100644 crates/openhuman-core/src/integrations/task_sources/bus_tests.rs

diff --git a/crates/openhuman-core/src/integrations/task_sources/bus.rs b/crates/openhuman-core/src/integrations/task_sources/bus.rs
index b396f56cddf..0d9c4806ac6 100644
--- a/crates/openhuman-core/src/integrations/task_sources/bus.rs
+++ b/crates/openhuman-core/src/integrations/task_sources/bus.rs
@@ -132,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;
+}

From 3c6232a0ab24744a9491e43132dc2a033af41b0c Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 20:16:53 +0300
Subject: [PATCH 16/17] fix(cron): visit agent jobs held in the storage backend

Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/cron/scheduler.rs | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/crates/openhuman-core/src/cron/scheduler.rs b/crates/openhuman-core/src/cron/scheduler.rs
index e0f6949fd96..b4af75b57af 100644
--- a/crates/openhuman-core/src/cron/scheduler.rs
+++ b/crates/openhuman-core/src/cron/scheduler.rs
@@ -114,7 +114,11 @@ pub(crate) async fn tick_live_agents(dispatcher: &mut JobDispatcher) {
                     return;
                 }
             };
-            if !crate::cron::store::db_path(&config).exists() {
+            // With a storage backend the agent's jobs live in its backend
+            // scope, not in a per-workspace SQLite file.
+            if crate::storage::installed().is_none()
+                && !crate::cron::store::db_path(&config).exists()
+            {
                 return;
             }
             let jobs = match due_jobs(&config, Utc::now()) {

From f1e69d5a90305f710d11a1fa999da2745a1077b5 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 9 Oct 2026 20:34:50 +0300
Subject: [PATCH 17/17] fix(security): report device scope as unsearched on
 read error

The owner lookup previously collapsed a failed device read into a
definite "not the owner" answer, which could let a scope be treated as
searched when it never was. It now propagates the read failure as an
unknown result so callers can distinguish an absent device from an
unreadable one.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/openhuman-core/src/security/devices/owner.rs | 10 ++++------
 crates/openhuman-core/src/storage/agents.rs         |  2 +-
 2 files changed, 5 insertions(+), 7 deletions(-)

diff --git a/crates/openhuman-core/src/security/devices/owner.rs b/crates/openhuman-core/src/security/devices/owner.rs
index bcb1ad59660..f2951d150d5 100644
--- a/crates/openhuman-core/src/security/devices/owner.rs
+++ b/crates/openhuman-core/src/security/devices/owner.rs
@@ -63,12 +63,10 @@ pub(super) async fn owner_of(
         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(),
-        )
+        // A read error is not "not here": report the scope as unsearched.
+        super::store::get_device(&config, channel_id)
+            .ok()
+            .map(|device| device.is_some())
     })
     .await;
     if let Some(owner) = found
diff --git a/crates/openhuman-core/src/storage/agents.rs b/crates/openhuman-core/src/storage/agents.rs
index 30ec2e50fce..baeee0edfc6 100644
--- a/crates/openhuman-core/src/storage/agents.rs
+++ b/crates/openhuman-core/src/storage/agents.rs
@@ -252,7 +252,7 @@ where
     let mut results = Vec::new();
     for (agent, context) in agent_contexts(fallback.as_ref()) {
         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