feat(storage): event subscribers act as the agent whose record the event names

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`.

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;
 

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,
         })
     }
 }

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;

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

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;

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,

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

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

crates/openhuman-core/src/flows/bus.rs

@@ -15,6 +15,7 @@
 mod tests;
 
 mod dedup_commit;
+mod owner;
 mod run_digest;
 mod trigger;
 

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

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;