From 1205b86abc9a966b591aac6971c29232fdd4a7ba Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Thu, 23 Jul 2026 23:52:15 +0530 Subject: [PATCH 1/4] feat(flows): per-flow memory namespace, tools, and post-run digest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gives a running flow a private, sandboxed memory namespace so it can remember what it already did across runs — e.g. a scheduled newsletter-digest flow tracking which items it already sent, to avoid re-sending them on the next scheduled run, without re-deriving that state from the target service each time. Adds: - `flow_namespace(flow_id)` (`src/openhuman/flows/mod.rs`) — the single function that builds a flow's namespace string. - `flow_memory_recall` / `flow_memory_remember` agent tools (`src/openhuman/flows/memory_tools.rs`), registered under the `flows` tool-registry gate and the `FLOWS` domain-group name list. - `FlowRunDigestSubscriber` (`src/openhuman/flows/bus.rs`) — on a successful `FlowRunFinished`, writes a compact, bounded summary of the run into the flow's namespace, with a best-effort retention cap. Registered in `src/core/jsonrpc.rs` alongside `FlowTriggerSubscriber`. - A `flows_delete` cleanup hook that clears a deleted flow's namespace. Security invariant (non-negotiable, preserved throughout): there is no code path by which a flow can write to, or even name, a namespace other than its own. `flow_memory_remember` derives the namespace internally from `flow_id` via `flow_namespace` — it has no `namespace` parameter a caller could override — and every write is tainted `MemoryTaint::ExternalSync` (automation output, never treated as a user-authored fact). `flow_memory_recall`'s `scope: "flows"` is read-only cross-flow visibility, confined to `flow_*` namespaces, and can never see or write to the user's personal/global memory. Reuses the existing autonomy-tier + approval gate (`SecurityPolicy:: enforce_tool_operation`) for writes; no new permission path. Deviates from the originally specced `flow:` (colon) separator: `UnifiedMemory`'s namespace sanitizer collapses `:` to `_` on the store/recall/list/clear_namespace paths but NOT on `forget`, which would silently break the digest subscriber's retention sweep and `scope: "flows"` cross-flow listing. `flow_` (underscore) is already a fixed point of that sanitizer, so every `Memory` method agrees on the same literal namespace string. Documented in `flow_namespace`'s doc comment. Closes #5173 --- src/core/jsonrpc.rs | 19 + src/openhuman/flows/bus.rs | 417 ++++++++++++++++++- src/openhuman/flows/memory_tools.rs | 618 ++++++++++++++++++++++++++++ src/openhuman/flows/mod.rs | 61 +++ src/openhuman/flows/ops.rs | 19 +- src/openhuman/flows/ops_tests.rs | 48 +++ src/openhuman/tools/mod.rs | 2 + src/openhuman/tools/ops.rs | 19 +- src/openhuman/tools/ops_tests.rs | 4 + 9 files changed, 1204 insertions(+), 3 deletions(-) create mode 100644 src/openhuman/flows/memory_tools.rs diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 5891872289..aaa568f65c 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -2143,6 +2143,25 @@ fn register_domain_subscribers( "[event_bus] failed to register flows trigger subscriber — bus not initialized" ); } + // Post-run memory digest (issue #5173): on a successful + // `FlowRunFinished`, writes a compact summary into the flow's own + // private memory namespace so a later run can `flow_memory_recall` + // it (e.g. a scheduled digest flow deduping what it already sent). + // Registered in the same `group_first_time(DomainGroup::Flows)` + // block as the trigger subscriber above — that guard only returns + // `true` once per process, so a second, separate call here would + // never register. + if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( + crate::openhuman::flows::bus::FlowRunDigestSubscriber::new(Arc::new( + config.clone(), + )), + )) { + std::mem::forget(handle); + } else { + log::warn!( + "[event_bus] failed to register flows run-digest subscriber — bus not initialized" + ); + } } } else { log::debug!("[event_bus] flows trigger subscriber SKIPPED — Flows domain disabled"); diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index ec03b60a42..61a8616887 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -13,7 +13,8 @@ use crate::core::event_bus::{DomainEvent, EventHandler}; use crate::openhuman::config::Config; use crate::openhuman::flows::store; -use crate::openhuman::flows::Flow; +use crate::openhuman::flows::{flow_namespace, Flow, FlowRun}; +use crate::openhuman::memory::{Memory, MemoryCategory, MemoryTaint}; use async_trait::async_trait; use serde_json::Value; use std::collections::HashSet; @@ -257,6 +258,180 @@ impl EventHandler for FlowTriggerSubscriber { } } +/// Bounds a post-run memory digest to a compact, LLM-cheap size — a single +/// run's summary must never dominate a later `flow_memory_recall`. +const DIGEST_MAX_CHARS: usize = 1000; + +/// Cap on how many `run_digest:*` entries [`FlowRunDigestSubscriber`] keeps +/// per flow's memory namespace before pruning the oldest. +const DIGEST_RETENTION_CAP: usize = 50; + +/// Listens for `DomainEvent::FlowRunFinished` and, on a successful terminal +/// status, writes a compact digest of the run into the flow's own private +/// memory namespace ([`flow_namespace`]) — e.g. so a later run of the same +/// scheduled digest flow can `flow_memory_recall` what it already sent +/// without re-deriving that from the target service. +/// +/// Success-only: `"failed"` / `"cancelled"` / `"interrupted"` / any other +/// terminal status is ignored, since a digest of a run that didn't actually +/// complete its work would misleadingly look like a record of real output. +/// +/// Best-effort throughout: every failure here is logged via `tracing::warn!` +/// and swallowed, never propagated — by the time this subscriber observes +/// `FlowRunFinished`, the run has already settled its own `flow_runs` row, so +/// a memory-layer hiccup must never retroactively affect run status. +pub struct FlowRunDigestSubscriber { + config: Arc, +} + +impl FlowRunDigestSubscriber { + pub fn new(config: Arc) -> Self { + Self { config } + } + + async fn handle_finished(&self, flow_id: &str, run_id: &str, status: &str) { + if status != "completed" && status != "completed_with_warnings" { + tracing::trace!(target: "flows", %flow_id, %run_id, %status, "[flows] digest: ignoring non-success terminal status"); + return; + } + + let flow_name = match store::get_flow(&self.config, flow_id) { + Ok(Some(flow)) => flow.name, + Ok(None) => { + tracing::debug!(target: "flows", %flow_id, %run_id, "[flows] digest: flow no longer exists — skipping"); + return; + } + Err(e) => { + tracing::warn!(target: "flows", %flow_id, %run_id, error = %e, "[flows] digest: failed to load flow — skipping"); + return; + } + }; + + let run = match store::get_flow_run(&self.config, run_id) { + Ok(Some(run)) => run, + Ok(None) => { + tracing::warn!(target: "flows", %flow_id, %run_id, "[flows] digest: run row not found — skipping"); + return; + } + Err(e) => { + tracing::warn!(target: "flows", %flow_id, %run_id, error = %e, "[flows] digest: failed to load run — skipping"); + return; + } + }; + + let digest = render_run_digest(&flow_name, &run); + + let client = match crate::openhuman::memory::ops::helpers::active_memory_client().await { + Ok(client) => client, + Err(e) => { + tracing::warn!(target: "flows", %flow_id, %run_id, error = %e, "[flows] digest: memory client unavailable — skipping"); + return; + } + }; + let memory: Arc = client.memory_handle(); + let namespace = flow_namespace(flow_id); + let digest_key = format!("run_digest:{run_id}"); + + if let Err(e) = memory + .store_with_taint( + &namespace, + &digest_key, + &digest, + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + { + tracing::warn!(target: "flows", %flow_id, %run_id, %namespace, error = %e, "[flows] digest: failed to write run digest"); + return; + } + + self.enforce_retention_cap(&memory, &namespace).await; + } + + /// Best-effort prune: keeps at most [`DIGEST_RETENTION_CAP`] `run_digest:*` + /// entries per flow namespace, evicting the oldest (by `timestamp`) first. + async fn enforce_retention_cap(&self, memory: &Arc, namespace: &str) { + let entries = match memory.list(Some(namespace), None, None).await { + Ok(entries) => entries, + Err(e) => { + tracing::warn!(target: "flows", %namespace, error = %e, "[flows] digest: retention sweep failed to list namespace"); + return; + } + }; + let mut digests: Vec<_> = entries + .into_iter() + .filter(|entry| entry.key.starts_with("run_digest:")) + .collect(); + if digests.len() <= DIGEST_RETENTION_CAP { + return; + } + // Oldest first, so the excess taken below is the stalest entries. + digests.sort_by(|a, b| a.timestamp.cmp(&b.timestamp)); + let excess = digests.len() - DIGEST_RETENTION_CAP; + for entry in digests.into_iter().take(excess) { + if let Err(e) = memory.forget(namespace, &entry.key).await { + tracing::warn!(target: "flows", %namespace, key = %entry.key, error = %e, "[flows] digest: retention sweep failed to forget stale entry"); + } + } + } +} + +#[async_trait] +impl EventHandler for FlowRunDigestSubscriber { + fn name(&self) -> &str { + "flows::digest" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["cron"]) + } + + async fn handle(&self, event: &DomainEvent) { + if let DomainEvent::FlowRunFinished { + flow_id, + run_id, + status, + } = event + { + self.handle_finished(flow_id, run_id, status).await; + } + } +} + +/// Truncates `s` to at most `max` `char`s, appending `…` when truncated. +fn truncate_chars(s: &str, max: usize) -> String { + if s.chars().count() <= max { + return s.to_string(); + } + let truncated: String = s.chars().take(max.saturating_sub(1)).collect(); + format!("{truncated}…") +} + +/// Composes a compact, bounded summary of a finished run: flow name, +/// finished-at, status, node count, and per-node status + truncated output. +/// Bounded to [`DIGEST_MAX_CHARS`] total. +fn render_run_digest(flow_name: &str, run: &FlowRun) -> String { + use std::fmt::Write; + let mut out = String::new(); + let _ = writeln!(out, "Flow: {flow_name}"); + let _ = writeln!(out, "Status: {}", run.status); + if let Some(finished_at) = &run.finished_at { + let _ = writeln!(out, "Finished: {finished_at}"); + } + let _ = writeln!(out, "Nodes: {}", run.steps.len()); + for step in &run.steps { + if out.chars().count() >= DIGEST_MAX_CHARS { + break; + } + let status = step.status.as_deref().unwrap_or("?"); + let output = truncate_chars(&step.output.to_string(), 120); + let _ = writeln!(out, "- {} [{status}]: {output}", step.node_id); + } + truncate_chars(&out, DIGEST_MAX_CHARS) +} + #[cfg(test)] mod tests { use super::*; @@ -461,4 +636,244 @@ mod tests { let b = FlowTriggerSubscriber::new(config); assert_eq!(a.name(), b.name()); } + + // ── FlowRunDigestSubscriber ───────────────────────────────────── + + #[test] + fn digest_name_and_domains_are_stable() { + let tmp = tempfile::TempDir::new().unwrap(); + let sub = FlowRunDigestSubscriber::new(test_config(&tmp)); + assert_eq!(sub.name(), "flows::digest"); + assert_eq!(sub.domains(), Some(&["cron"][..])); + } + + #[tokio::test] + async fn digest_handle_does_not_panic_on_unrelated_events() { + let tmp = tempfile::TempDir::new().unwrap(); + let sub = FlowRunDigestSubscriber::new(test_config(&tmp)); + // Must not panic, and must not touch the memory layer at all, for + // any event other than `FlowRunFinished`. + sub.handle(&DomainEvent::CronJobTriggered { + job_id: "j1".into(), + job_name: "test".into(), + job_type: "shell".into(), + }) + .await; + } + + #[tokio::test] + async fn digest_ignores_failed_run() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + crate::openhuman::memory::global::init(config.workspace_dir.clone()) + .expect("init test memory client"); + + let flow = flow_with_trigger_config("f-failed", true, json!({})); + store::upsert_flow(&config, &flow).unwrap(); + store::insert_flow_run(&config, "run-failed", "f-failed", "thread-failed", "2026-01-01T00:00:00Z") + .unwrap(); + store::finish_flow_run( + &config, + "run-failed", + "failed", + "2026-01-01T00:05:00Z", + &[], + &[], + Some("boom"), + ) + .unwrap(); + + let sub = FlowRunDigestSubscriber::new(config); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-failed".into(), + run_id: "run-failed".into(), + status: "failed".into(), + }) + .await; + + let client = crate::openhuman::memory::global::client().expect("global client ready"); + let memory = client.memory_handle(); + let entry = memory + .get(&flow_namespace("f-failed"), "run_digest:run-failed") + .await + .unwrap(); + assert!( + entry.is_none(), + "a failed run must never produce a run_digest entry" + ); + } + + #[tokio::test] + async fn digest_ignores_cancelled_run() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + crate::openhuman::memory::global::init(config.workspace_dir.clone()) + .expect("init test memory client"); + + let flow = flow_with_trigger_config("f-cancelled", true, json!({})); + store::upsert_flow(&config, &flow).unwrap(); + store::insert_flow_run( + &config, + "run-cancelled", + "f-cancelled", + "thread-cancelled", + "2026-01-01T00:00:00Z", + ) + .unwrap(); + store::finish_flow_run( + &config, + "run-cancelled", + "cancelled", + "2026-01-01T00:05:00Z", + &[], + &[], + None, + ) + .unwrap(); + + let sub = FlowRunDigestSubscriber::new(config); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-cancelled".into(), + run_id: "run-cancelled".into(), + status: "cancelled".into(), + }) + .await; + + let client = crate::openhuman::memory::global::client().expect("global client ready"); + let memory = client.memory_handle(); + let entry = memory + .get(&flow_namespace("f-cancelled"), "run_digest:run-cancelled") + .await + .unwrap(); + assert!(entry.is_none()); + } + + #[tokio::test] + async fn digest_writes_run_digest_entry_for_completed_run() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + crate::openhuman::memory::global::init(config.workspace_dir.clone()) + .expect("init test memory client"); + + let flow = flow_with_trigger_config("f-ok", true, json!({})); + store::upsert_flow(&config, &flow).unwrap(); + store::insert_flow_run(&config, "run-ok", "f-ok", "thread-ok", "2026-01-01T00:00:00Z") + .unwrap(); + let step = crate::openhuman::flows::FlowRunStep { + node_id: "n1".to_string(), + output: json!({ "sent": 3 }), + port: None, + status: Some("success".to_string()), + duration_ms: Some(12), + diagnostics: Vec::new(), + }; + store::finish_flow_run( + &config, + "run-ok", + "completed", + "2026-01-01T00:05:00Z", + &[step], + &[], + None, + ) + .unwrap(); + + let sub = FlowRunDigestSubscriber::new(config); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-ok".into(), + run_id: "run-ok".into(), + status: "completed".into(), + }) + .await; + + let client = crate::openhuman::memory::global::client().expect("global client ready"); + let memory = client.memory_handle(); + let entry = memory + .get(&flow_namespace("f-ok"), "run_digest:run-ok") + .await + .unwrap() + .expect("completed run must produce a run_digest entry"); + assert_eq!(entry.taint, MemoryTaint::ExternalSync); + assert!(entry.content.contains("f-ok")); + assert!(entry.content.contains("completed")); + assert!(entry.content.contains("n1")); + assert!(entry.content.chars().count() <= DIGEST_MAX_CHARS); + } + + #[tokio::test] + async fn digest_treats_completed_with_warnings_as_success() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + crate::openhuman::memory::global::init(config.workspace_dir.clone()) + .expect("init test memory client"); + + let flow = flow_with_trigger_config("f-warn", true, json!({})); + store::upsert_flow(&config, &flow).unwrap(); + store::insert_flow_run(&config, "run-warn", "f-warn", "thread-warn", "2026-01-01T00:00:00Z") + .unwrap(); + store::finish_flow_run( + &config, + "run-warn", + "completed_with_warnings", + "2026-01-01T00:05:00Z", + &[], + &[], + None, + ) + .unwrap(); + + let sub = FlowRunDigestSubscriber::new(config); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-warn".into(), + run_id: "run-warn".into(), + status: "completed_with_warnings".into(), + }) + .await; + + let client = crate::openhuman::memory::global::client().expect("global client ready"); + let memory = client.memory_handle(); + let entry = memory + .get(&flow_namespace("f-warn"), "run_digest:run-warn") + .await + .unwrap(); + assert!(entry.is_some()); + } + + #[test] + fn truncate_chars_bounds_output_and_marks_truncation() { + let long = "x".repeat(50); + let truncated = truncate_chars(&long, 10); + assert_eq!(truncated.chars().count(), 10); + assert!(truncated.ends_with('…')); + + let short = "hello"; + assert_eq!(truncate_chars(short, 10), "hello"); + } + + #[test] + fn render_run_digest_is_bounded_and_includes_key_fields() { + let run = FlowRun { + id: "run-1".to_string(), + flow_id: "f1".to_string(), + thread_id: "thread-1".to_string(), + status: "completed".to_string(), + started_at: "2026-01-01T00:00:00Z".to_string(), + finished_at: Some("2026-01-01T00:05:00Z".to_string()), + steps: vec![crate::openhuman::flows::FlowRunStep { + node_id: "n1".to_string(), + output: json!({ "ok": true }), + port: None, + status: Some("success".to_string()), + duration_ms: Some(5), + diagnostics: Vec::new(), + }], + pending_approvals: Vec::new(), + error: None, + }; + let digest = render_run_digest("My Flow", &run); + assert!(digest.contains("My Flow")); + assert!(digest.contains("completed")); + assert!(digest.contains("n1")); + assert!(digest.chars().count() <= DIGEST_MAX_CHARS); + } } diff --git a/src/openhuman/flows/memory_tools.rs b/src/openhuman/flows/memory_tools.rs new file mode 100644 index 0000000000..4e3ce27681 --- /dev/null +++ b/src/openhuman/flows/memory_tools.rs @@ -0,0 +1,618 @@ +//! Agent tools giving a running flow a private, sandboxed memory namespace +//! (`flow_` — see [`super::flow_namespace`]). +//! +//! Motivating use case: a newsletter-digest flow that runs on a schedule +//! needs to remember which items it already sent so it doesn't re-send them +//! on the next run. Without a durable, flow-scoped place to note "already +//! sent: ", the agent node inside the flow has no way to dedupe +//! across runs other than re-deriving state from the target service itself +//! (which is often lossy or rate-limited). +//! +//! **Security invariant (non-negotiable):** there is no code path here by +//! which a flow can write to — or even name — a namespace other than its +//! own. [`FlowMemoryRememberTool`] derives the namespace internally via +//! [`super::flow_namespace`] from the caller-supplied `flow_id`; there is no +//! `namespace` parameter a caller could override. Every write is tainted +//! [`MemoryTaint::ExternalSync`] (automation output, not user-authored +//! fact), matching the same taint sync pipelines use for third-party +//! content, so the subconscious gate treats it exactly as conservatively. +//! [`FlowMemoryRecallTool`]'s `scope: "flows"` is read-only cross-flow +//! visibility — it can never be used to write outside a flow's own +//! namespace either. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::json; + +use crate::openhuman::flows::flow_namespace; +use crate::openhuman::memory::{Memory, MemoryCategory, MemoryEntry, MemoryTaint, RecallOpts}; +use crate::openhuman::security::policy::ToolOperation; +use crate::openhuman::security::SecurityPolicy; +use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; + +/// The persisted-namespace prefix matching every flow's memory namespace, as +/// [`Memory::namespace_summaries`] returns it. +/// +/// This is intentionally the *same* string as +/// [`super::FLOW_MEMORY_NAMESPACE_PREFIX`] (see that constant's doc comment +/// for why the separator is `_` rather than the originally specced `:`) — +/// kept as a separate, explicitly-named constant here so the "match against +/// what recall/list see" intent stays self-evident at each call site, +/// independent of whether the two ever need to diverge in the future. +const FLOW_MEMORY_NAMESPACE_LISTED_PREFIX: &str = super::FLOW_MEMORY_NAMESPACE_PREFIX; + +/// Read-only recall over a flow's own memory namespace, or (with +/// `scope: "flows"`) across every flow's namespace. +/// +/// `scope: "flows"` is intentionally still read-only and still confined to +/// `flow_*` namespaces — it can never see the user's personal/global memory, +/// only other flows' own automation output. +pub struct FlowMemoryRecallTool { + memory: Arc, +} + +impl FlowMemoryRecallTool { + pub fn new(memory: Arc) -> Self { + Self { memory } + } +} + +/// Renders recall hits the same way [`crate::openhuman::memory::tools::recall`] +/// does, with the flow's memory namespace context in each line so a +/// cross-flow `scope: "flows"` result is attributable. +fn render_entries(entries: &[MemoryEntry]) -> String { + if entries.is_empty() { + return "No memories found matching that query.".to_string(); + } + use std::fmt::Write; + let mut output = format!("Found {} memories:\n", entries.len()); + for entry in entries { + let score = entry + .score + .map_or_else(String::new, |s| format!(" [{s:.0}%]")); + let namespace = entry.namespace.as_deref().unwrap_or("?"); + let _ = writeln!( + output, + "- [{namespace}] [{}] {}: {}{score}", + entry.category, entry.key, entry.content + ); + } + output +} + +#[async_trait] +impl Tool for FlowMemoryRecallTool { + fn name(&self) -> &str { + "flow_memory_recall" + } + + fn description(&self) -> &str { + "Search a flow's own private memory namespace for relevant facts — e.g. so a scheduled \ + digest flow can check what it already sent before, to avoid duplicates. `scope: \"flow\"` \ + (the default) searches only the calling flow's own namespace. `scope: \"flows\"` searches \ + read-only across every flow's private namespace (useful when related flows should dedupe \ + against each other), merged and re-ranked by relevance. This tool never reads the user's \ + personal or global memory — only memory flows have written about their own runs." + } + + fn parameters_schema(&self) -> serde_json::Value { + json!({ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Keywords or phrase to search for" + }, + "flow_id": { + "type": "string", + "description": "The calling flow's id" + }, + "scope": { + "type": "string", + "enum": ["flow", "flows"], + "description": "\"flow\" (default) searches only this flow's own memory namespace; \"flows\" searches read-only across every flow's namespace." + }, + "limit": { + "type": "integer", + "description": "Max results to return (default: 5)" + } + }, + "required": ["query", "flow_id"] + }) + } + + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { + let query = args + .get("query") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("Missing 'query' parameter"))? + .trim(); + if query.is_empty() { + return Err(anyhow::anyhow!("query cannot be empty")); + } + let flow_id = args + .get("flow_id") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("Missing 'flow_id' parameter"))? + .trim(); + if flow_id.is_empty() { + return Err(anyhow::anyhow!("flow_id cannot be empty")); + } + let scope = args + .get("scope") + .and_then(|v| v.as_str()) + .unwrap_or("flow"); + + #[allow(clippy::cast_possible_truncation)] + let limit = args + .get("limit") + .and_then(serde_json::Value::as_u64) + .map_or(5, |v| v as usize); + + match scope { + "flow" => { + let namespace = flow_namespace(flow_id); + let opts = RecallOpts { + namespace: Some(namespace.as_str()), + ..RecallOpts::default() + }; + match self.memory.recall(query, limit, opts).await { + Ok(entries) => Ok(ToolResult::success(render_entries(&entries))), + Err(e) => Ok(ToolResult::error(format!("Flow memory recall failed: {e}"))), + } + } + "flows" => { + let summaries = match self.memory.namespace_summaries().await { + Ok(summaries) => summaries, + Err(e) => { + return Ok(ToolResult::error(format!( + "Failed to list flow memory namespaces: {e}" + ))) + } + }; + + let mut merged: Vec = Vec::new(); + for summary in summaries + .iter() + .filter(|s| s.namespace.starts_with(FLOW_MEMORY_NAMESPACE_LISTED_PREFIX)) + { + let opts = RecallOpts { + namespace: Some(summary.namespace.as_str()), + ..RecallOpts::default() + }; + match self.memory.recall(query, limit, opts).await { + Ok(entries) => merged.extend(entries), + Err(e) => { + log::warn!( + "[flows:memory] flow_memory_recall scope=flows failed for namespace={}: {e}", + summary.namespace + ); + } + } + } + merged.sort_by(|a, b| { + b.score + .unwrap_or(0.0) + .partial_cmp(&a.score.unwrap_or(0.0)) + .unwrap_or(std::cmp::Ordering::Equal) + }); + merged.truncate(limit); + Ok(ToolResult::success(render_entries(&merged))) + } + other => Ok(ToolResult::error(format!( + "Unknown scope '{other}': expected 'flow' or 'flows'" + ))), + } + } + + fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool { + true + } +} + +/// Write access to a flow's own private memory namespace — and *only* its +/// own. See the module doc for the security invariant this tool exists to +/// preserve. +pub struct FlowMemoryRememberTool { + memory: Arc, + security: Arc, +} + +impl FlowMemoryRememberTool { + pub fn new(memory: Arc, security: Arc) -> Self { + Self { memory, security } + } +} + +#[async_trait] +impl Tool for FlowMemoryRememberTool { + fn name(&self) -> &str { + "flow_memory_remember" + } + + fn description(&self) -> &str { + "Store a fact in THIS flow's own private memory namespace — e.g. so a scheduled digest \ + flow can remember which items it already sent, to avoid re-sending them on the next run. \ + The namespace is derived internally from `flow_id`; there is no way to target the user's \ + personal memory or another flow's namespace from this tool. Stored content is tainted as \ + externally-sourced automation output, never treated as a user-authored fact." + } + + fn parameters_schema(&self) -> serde_json::Value { + json!({ + "type": "object", + "properties": { + "flow_id": { + "type": "string", + "description": "The calling flow's id" + }, + "key": { + "type": "string", + "description": "Unique key for this memory within the flow's own namespace" + }, + "content": { + "type": "string", + "description": "The information to remember" + }, + "category": { + "type": "string", + "description": "Memory category: 'core' (permanent), 'daily' (session), 'conversation' (chat), or a custom category name. Defaults to 'core'." + } + }, + "required": ["flow_id", "key", "content"] + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::Write + } + + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { + let flow_id = args + .get("flow_id") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("Missing 'flow_id' parameter"))?; + let key = args + .get("key") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("Missing 'key' parameter"))?; + let content = args + .get("content") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("Missing 'content' parameter"))?; + + let category = match args.get("category").and_then(|v| v.as_str()) { + Some("core") | None => MemoryCategory::Core, + Some("daily") => MemoryCategory::Daily, + Some("conversation") => MemoryCategory::Conversation, + // Route custom categories through `FromStr` so a `custom:` + // wire value resolves back to `Custom("")` rather than + // double-prefixing — mirrors `memory_store::MemoryStoreTool`. + Some(other) => other + .parse() + .unwrap_or_else(|_| MemoryCategory::Custom(other.to_string())), + }; + + if let Err(error) = self + .security + .enforce_tool_operation(ToolOperation::Act, "flow_memory_remember") + { + return Ok(ToolResult::error(error)); + } + + let flow_id = flow_id.trim(); + if flow_id.is_empty() { + return Ok(ToolResult::error("flow_id cannot be empty".to_string())); + } + let key = key.trim(); + if key.is_empty() { + return Ok(ToolResult::error("key cannot be empty".to_string())); + } + + if crate::openhuman::memory_store::safety::has_likely_secret(content) { + log::warn!( + "[flows:memory:safety] flow_memory_remember rejected secret-like content flow_id_chars={} key_chars={} content_chars={}", + flow_id.chars().count(), + key.chars().count(), + content.chars().count() + ); + return Ok(ToolResult::error( + "Refusing to store content that looks like a secret. Remove credentials or tokens and try again.".to_string(), + )); + } + + // SECURITY: the namespace is derived internally from `flow_id` — + // this tool has no `namespace` parameter, so a flow can only ever + // write into its own `flow_` sandbox, never user/global memory + // or another flow's namespace. + let namespace = flow_namespace(flow_id); + let display_key = format!("{namespace}/{key}"); + match self + .memory + .store_with_taint( + &namespace, + key, + content, + category, + None, + MemoryTaint::ExternalSync, + ) + .await + { + Ok(()) => Ok(ToolResult::success(format!( + "Stored flow memory: {display_key}" + ))), + Err(e) => Ok(ToolResult::error(format!( + "Failed to store flow memory: {e}" + ))), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::embeddings::NoopEmbedding; + use crate::openhuman::memory_store::UnifiedMemory; + use crate::openhuman::security::AutonomyLevel; + use tempfile::TempDir; + + fn test_security() -> Arc { + Arc::new(SecurityPolicy::default()) + } + + fn test_mem() -> (TempDir, Arc) { + let tmp = TempDir::new().unwrap(); + let mem = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap(); + (tmp, Arc::new(mem)) + } + + // ── FlowMemoryRecallTool ──────────────────────────────────────── + + #[test] + fn recall_name_and_schema() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRecallTool::new(mem); + assert_eq!(tool.name(), "flow_memory_recall"); + let schema = tool.parameters_schema(); + assert!(schema["properties"]["query"].is_object()); + assert!(schema["properties"]["flow_id"].is_object()); + assert!(schema["properties"]["scope"].is_object()); + } + + #[tokio::test] + async fn recall_empty_returns_no_results() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRecallTool::new(mem); + let result = tool + .execute(json!({"query": "anything", "flow_id": "f1"})) + .await + .unwrap(); + assert!(!result.is_error); + assert!(result.output().contains("No memories found")); + } + + #[tokio::test] + async fn store_then_recall_matches() { + let (_tmp, mem) = test_mem(); + mem.store_with_taint( + &flow_namespace("f1"), + "sent_item_42", + "Sent newsletter item 42 to subscribers", + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + + let tool = FlowMemoryRecallTool::new(mem); + let result = tool + .execute(json!({"query": "newsletter item 42", "flow_id": "f1"})) + .await + .unwrap(); + assert!(!result.is_error); + assert!(result.output().contains("newsletter item 42") || result.output().contains("42")); + assert!(result.output().contains("Found 1")); + } + + #[tokio::test] + async fn scope_flow_isolates_to_own_namespace() { + let (_tmp, mem) = test_mem(); + mem.store_with_taint( + &flow_namespace("f1"), + "k", + "shared keyword hit", + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + mem.store_with_taint( + &flow_namespace("f2"), + "k", + "shared keyword hit", + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + + let tool = FlowMemoryRecallTool::new(mem); + let result = tool + .execute(json!({"query": "shared keyword", "flow_id": "f1", "scope": "flow"})) + .await + .unwrap(); + assert!(!result.is_error); + // Only f1's own entry, not f2's. + assert!(result.output().contains("Found 1")); + } + + #[tokio::test] + async fn scope_flows_crosses_namespaces() { + let (_tmp, mem) = test_mem(); + mem.store_with_taint( + &flow_namespace("f1"), + "k", + "shared keyword hit from f1", + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + mem.store_with_taint( + &flow_namespace("f2"), + "k", + "shared keyword hit from f2", + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + + let tool = FlowMemoryRecallTool::new(mem); + let result = tool + .execute(json!({"query": "shared keyword", "flow_id": "f1", "scope": "flows"})) + .await + .unwrap(); + assert!(!result.is_error); + // Both f1's and f2's namespaces are visible under scope="flows". + assert!(result.output().contains("Found 2")); + } + + #[tokio::test] + async fn recall_missing_query_errs() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRecallTool::new(mem); + let result = tool.execute(json!({"flow_id": "f1"})).await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn recall_missing_flow_id_errs() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRecallTool::new(mem); + let result = tool.execute(json!({"query": "anything"})).await; + assert!(result.is_err()); + } + + // ── FlowMemoryRememberTool ────────────────────────────────────── + + #[test] + fn remember_name_and_schema() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRememberTool::new(mem, test_security()); + assert_eq!(tool.name(), "flow_memory_remember"); + let schema = tool.parameters_schema(); + assert!(schema["properties"]["flow_id"].is_object()); + assert!(schema["properties"]["key"].is_object()); + assert!(schema["properties"]["content"].is_object()); + // No `namespace` parameter exists — the security invariant that a + // flow can never target another namespace. + assert!(schema["properties"]["namespace"].is_null()); + assert_eq!(tool.permission_level(), PermissionLevel::Write); + } + + #[tokio::test] + async fn remember_stores_with_external_sync_taint() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRememberTool::new(mem.clone(), test_security()); + let result = tool + .execute(json!({"flow_id": "f1", "key": "sent_item_42", "content": "Sent item 42"})) + .await + .unwrap(); + assert!(!result.is_error); + + let entry = mem + .get(&flow_namespace("f1"), "sent_item_42") + .await + .unwrap() + .expect("entry should be stored"); + assert_eq!(entry.content, "Sent item 42"); + assert_eq!(entry.taint, MemoryTaint::ExternalSync); + } + + #[tokio::test] + async fn remember_writes_only_to_own_flow_namespace() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRememberTool::new(mem.clone(), test_security()); + tool.execute(json!({"flow_id": "f1", "key": "k", "content": "f1 content"})) + .await + .unwrap(); + + // Never lands in another flow's namespace, the shared "flows" scope + // namespace, or global/user memory. + assert!(mem + .get(&flow_namespace("f2"), "k") + .await + .unwrap() + .is_none()); + assert!(mem.get("global", "k").await.unwrap().is_none()); + assert!(mem + .get("f1", "k") // raw flow_id, not the derived namespace + .await + .unwrap() + .is_none()); + } + + #[tokio::test] + async fn remember_blocked_in_readonly_autonomy() { + let (_tmp, mem) = test_mem(); + let readonly = Arc::new(SecurityPolicy { + autonomy: AutonomyLevel::ReadOnly, + ..SecurityPolicy::default() + }); + let tool = FlowMemoryRememberTool::new(mem.clone(), readonly); + let result = tool + .execute(json!({"flow_id": "f1", "key": "k", "content": "blocked"})) + .await + .unwrap(); + assert!(result.is_error); + assert!(result.output().contains("read-only mode")); + assert!(mem.get(&flow_namespace("f1"), "k").await.unwrap().is_none()); + } + + #[tokio::test] + async fn remember_rejects_secret_like_content() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRememberTool::new(mem.clone(), test_security()); + let result = tool + .execute(json!({ + "flow_id": "f1", + "key": "api", + "content": "api_key=sk-123456789012345678901234567890" + })) + .await + .unwrap(); + assert!(result.is_error); + assert!(result.output().contains("looks like a secret")); + assert!(mem + .get(&flow_namespace("f1"), "api") + .await + .unwrap() + .is_none()); + } + + #[tokio::test] + async fn remember_missing_flow_id_errs() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRememberTool::new(mem, test_security()); + let result = tool.execute(json!({"key": "k", "content": "c"})).await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn remember_missing_content_errs() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRememberTool::new(mem, test_security()); + let result = tool + .execute(json!({"flow_id": "f1", "key": "k"})) + .await; + assert!(result.is_err()); + } +} diff --git a/src/openhuman/flows/mod.rs b/src/openhuman/flows/mod.rs index 2eddc37cdc..ab1d1658b4 100644 --- a/src/openhuman/flows/mod.rs +++ b/src/openhuman/flows/mod.rs @@ -12,6 +12,7 @@ pub mod builder_tools; pub mod bus; pub mod discovery_tools; mod draft_store; +pub mod memory_tools; mod n8n_import; pub mod node_contracts; pub mod ops; @@ -43,3 +44,63 @@ pub use types::{ DraftOrigin, Flow, FlowConnection, FlowDraft, FlowImport, FlowRevision, FlowRun, FlowRunStep, FlowRunTrigger, FlowSuggestion, FlowValidation, FlowValidationError, SuggestionStatus, }; + +/// Prefix for a flow's private, sandboxed memory namespace (see +/// [`flow_namespace`]). +/// +/// **Deviates from the originally specced `"flow:"` (colon) separator — +/// deliberately.** The `Memory` trait's `UnifiedMemory` backend +/// (`src/openhuman/memory_store/`) is internally inconsistent about +/// namespace sanitization: `store_with_taint`/`recall`/`list`/ +/// `MemoryClient::clear_namespace` all route through +/// `UnifiedMemory::sanitize_namespace` +/// (`memory_store/namespace_store/init.rs`), which collapses any character +/// outside `[A-Za-z0-9_/-]` — including `:` — to `_` before touching SQLite. +/// But `Memory::forget` (`memory_store/memory_trait.rs`) queries +/// `WHERE namespace = ?1` against the **raw, unsanitized** argument. With a +/// `"flow:"` prefix, `forget("flow:", key)` would therefore silently +/// never match the row `store_with_taint` actually persisted as +/// `"flow_"` — the post-run digest subscriber's retention sweep +/// (`bus::FlowRunDigestSubscriber`) would then never evict old entries, and +/// `namespace_summaries()`-based cross-flow listing (`scope: "flows"` in +/// `memory_tools::FlowMemoryRecallTool`) would have to match the sanitized +/// form anyway since `namespace_summaries` reads the persisted (sanitized) +/// column back verbatim. Using `"flow_"` (already a fixed point of +/// `sanitize_namespace`, since flow ids are hyphenated UUIDs — no character +/// in either the prefix or a flow id ever needs sanitizing) makes every +/// `Memory` method agree with every other one on the exact namespace +/// string, with no silent mismatch anywhere. The namespace is still +/// shared-root and profile-independent exactly as specified — only the +/// separator character changed. +pub const FLOW_MEMORY_NAMESPACE_PREFIX: &str = "flow_"; + +/// Builds a flow's private, profile-independent memory namespace from a +/// `flow_id`. +/// +/// **Security invariant:** this is the *only* place in the codebase that may +/// construct this namespace string. Every caller — the `flow_memory_recall` +/// / `flow_memory_remember` agent tools (`memory_tools.rs`), the post-run +/// digest subscriber (`bus::FlowRunDigestSubscriber`), and the +/// `flows_delete` cleanup hook (`ops::flows_delete`) — goes through this +/// function with a `flow_id`, never with a caller-supplied raw namespace. +/// A flow can therefore never write to, or be told the name of, any memory +/// namespace but its own. +pub fn flow_namespace(flow_id: &str) -> String { + format!("{FLOW_MEMORY_NAMESPACE_PREFIX}{flow_id}") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn flow_namespace_uses_the_shared_root_prefix() { + assert_eq!(flow_namespace("abc-123"), "flow_abc-123"); + assert!(flow_namespace("abc-123").starts_with(FLOW_MEMORY_NAMESPACE_PREFIX)); + } + + #[test] + fn flow_namespace_is_distinct_per_flow() { + assert_ne!(flow_namespace("a"), flow_namespace("b")); + } +} diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 97fcff5d6a..31ae1dc256 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -23,7 +23,7 @@ use crate::openhuman::flows::store; use crate::openhuman::flows::types::{ FlowConnection, FlowRunStep, FlowRunTrigger, FlowSuggestion, SuggestionStatus, }; -use crate::openhuman::flows::{Flow, FlowRun}; +use crate::openhuman::flows::{flow_namespace, Flow, FlowRun}; use crate::rpc::RpcOutcome; /// Overall safety bound on a single `flows_run` / `flows_resume`. Individual @@ -3430,6 +3430,23 @@ pub async fn flows_delete(config: &Config, id: &str) -> Result store::remove_flow(config, id).map_err(|e| e.to_string())?; tracing::debug!(target: "flows", flow_id = %id, "[flows] flows_delete: removed"); + + // Best-effort: clear this flow's private memory namespace along with its + // row — a deleted flow must not leave stray `flow_memory_remember` + // entries or run digests behind. Never fails the delete itself: the flow + // row is already gone by this point regardless of what happens here. + let memory_namespace = flow_namespace(id); + match crate::openhuman::memory::ops::helpers::active_memory_client().await { + Ok(client) => { + if let Err(e) = client.clear_namespace(&memory_namespace).await { + tracing::warn!(target: "flows", flow_id = %id, namespace = %memory_namespace, error = %e, "[flows] flows_delete: failed to clear flow memory namespace"); + } + } + Err(e) => { + tracing::warn!(target: "flows", flow_id = %id, namespace = %memory_namespace, error = %e, "[flows] flows_delete: memory client unavailable — could not clear flow memory namespace"); + } + } + publish_flow_changed(id, "deleted", "system"); Ok(RpcOutcome::new( json!({ "id": id, "removed": true }), diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index 70b25bcdb8..4c6daab733 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -1218,6 +1218,54 @@ async fn flows_delete_unbinds_schedule_cron_job() { ); } +#[tokio::test] +async fn flows_delete_clears_flow_memory_namespace() { + use crate::openhuman::memory::{Memory, MemoryCategory, MemoryTaint}; + + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + crate::openhuman::memory::global::init(config.workspace_dir.clone()) + .expect("init test memory client"); + + let created = flows_create(&config, "with-memory".to_string(), trigger_only_graph(), false) + .await + .unwrap(); + let flow_id = created.value.id.clone(); + + let client = crate::openhuman::memory::global::client().expect("global client ready"); + let memory = client.memory_handle(); + memory + .store_with_taint( + &flow_namespace(&flow_id), + "sent_item_1", + "Sent item 1", + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + assert!( + memory + .get(&flow_namespace(&flow_id), "sent_item_1") + .await + .unwrap() + .is_some(), + "precondition: flow memory entry was stored" + ); + + flows_delete(&config, &flow_id).await.unwrap(); + + assert!( + memory + .get(&flow_namespace(&flow_id), "sent_item_1") + .await + .unwrap() + .is_none(), + "flows_delete must clear the flow's own memory namespace" + ); +} + #[tokio::test] async fn flows_update_rebinds_schedule_cron_job_when_trigger_schedule_changes() { let tmp = TempDir::new().unwrap(); diff --git a/src/openhuman/tools/mod.rs b/src/openhuman/tools/mod.rs index 5f2d91e4d1..74205f649b 100644 --- a/src/openhuman/tools/mod.rs +++ b/src/openhuman/tools/mod.rs @@ -31,6 +31,8 @@ pub use crate::openhuman::flows::builder_tools::*; #[cfg(feature = "flows")] pub use crate::openhuman::flows::discovery_tools::*; #[cfg(feature = "flows")] +pub use crate::openhuman::flows::memory_tools::*; +#[cfg(feature = "flows")] pub use crate::openhuman::flows::tools::*; pub use crate::openhuman::health::tools::*; pub use crate::openhuman::integrations::tools::*; diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 2c0b5dad64..dcd7dd6891 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -419,6 +419,17 @@ pub fn all_tools_with_runtime( // effect — writes only to the agent's own suggestions store. #[cfg(feature = "flows")] Box::new(SuggestWorkflowsTool::new(config.clone())), + // Per-flow sandboxed memory (issue #5173): lets a running flow + // (e.g. a scheduled newsletter-digest) remember what it already did + // — dedupe across runs — without ever touching the user's own + // memory. Namespace is derived internally from `flow_id`; there is + // no code path by which either tool can address a namespace other + // than the calling flow's own (`flow_memory_recall`'s `scope: + // "flows"` is read-only cross-flow visibility, not a write path). + #[cfg(feature = "flows")] + Box::new(FlowMemoryRecallTool::new(memory.clone())), + #[cfg(feature = "flows")] + Box::new(FlowMemoryRememberTool::new(memory.clone(), security.clone())), // Wallet tools — expose wallet operations to the agent tool-call pipeline // so the crypto sub-agent can prepare transfers, check status, etc. // Gated with the `web3` feature (the wallet domain is compiled out when @@ -1346,7 +1357,7 @@ fn tool_group(name: &str) -> crate::core::all::DomainGroup { // stays callable under a custom `DomainSet { platform: true, flows: false }`, // leaking the flows surface past the runtime gate (#4808 review; #4797 // maintainer review). Keep this in lockstep with the `#[cfg(feature = - // "flows")]` registrations in `all_tools_with_runtime` above — the same 26 + // "flows")]` registrations in `all_tools_with_runtime` above — the same 28 // names asserted by `default_tools_omits_flows_tools_when_feature_off`. const FLOWS: &[&str] = &[ "propose_workflow", @@ -1377,6 +1388,12 @@ fn tool_group(name: &str) -> crate::core::all::DomainGroup { // The `rhai_workflows` (.ragsh) tool is compile-gated with `flows` and // belongs to the same runtime domain — drop it when Flows is off too. "rhai_workflows", + // Per-flow sandboxed memory (issue #5173) — `flow_` prefixed, not + // `memory_`, so it does NOT fall under the `memory_` prefix check + // below and must be listed here explicitly like every other + // flow-owned tool. + "flow_memory_recall", + "flow_memory_remember", ]; // Voice family agent tools (audio_toolkit) — no `voice_`/`tts_`/`stt_` // prefix, so they must be listed explicitly or they fall through to diff --git a/src/openhuman/tools/ops_tests.rs b/src/openhuman/tools/ops_tests.rs index 599af03f8e..4e2528b0cd 100644 --- a/src/openhuman/tools/ops_tests.rs +++ b/src/openhuman/tools/ops_tests.rs @@ -2566,6 +2566,8 @@ fn tool_group_classifies_gate_and_harness_families() { "list_node_kinds", "get_node_kind_contract", "rhai_workflows", + "flow_memory_recall", + "flow_memory_remember", ] { assert_eq!( tool_group(flow_tool), @@ -2694,6 +2696,8 @@ fn default_tools_omits_flows_tools_when_feature_off() { "save_workflow", "suggest_workflows", "rhai_workflows", + "flow_memory_recall", + "flow_memory_remember", ] { assert!( !names.iter().any(|n| n == absent), From ec09c88ea588e73ae0992e6372b801fc11ed4d2a Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Fri, 24 Jul 2026 00:02:02 +0530 Subject: [PATCH 2/4] style(flows): rustfmt new flow-memory code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Formatting-only (rustfmt --edition 2021) on the files this PR adds/edits. The remaining Rust Quality diffs (caps.rs, provider.rs) are pre-existing on upstream/main — head e780277e7 already fails Rust Quality from #5151 et al. — and are untouched here. --- src/openhuman/flows/bus.rs | 30 +++++++++++++++++++++++------ src/openhuman/flows/memory_tools.rs | 15 +++------------ src/openhuman/flows/ops_tests.rs | 11 ++++++++--- src/openhuman/tools/ops.rs | 5 ++++- 4 files changed, 39 insertions(+), 22 deletions(-) diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index 61a8616887..d296f10f10 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -670,8 +670,14 @@ mod tests { let flow = flow_with_trigger_config("f-failed", true, json!({})); store::upsert_flow(&config, &flow).unwrap(); - store::insert_flow_run(&config, "run-failed", "f-failed", "thread-failed", "2026-01-01T00:00:00Z") - .unwrap(); + store::insert_flow_run( + &config, + "run-failed", + "f-failed", + "thread-failed", + "2026-01-01T00:00:00Z", + ) + .unwrap(); store::finish_flow_run( &config, "run-failed", @@ -757,8 +763,14 @@ mod tests { let flow = flow_with_trigger_config("f-ok", true, json!({})); store::upsert_flow(&config, &flow).unwrap(); - store::insert_flow_run(&config, "run-ok", "f-ok", "thread-ok", "2026-01-01T00:00:00Z") - .unwrap(); + store::insert_flow_run( + &config, + "run-ok", + "f-ok", + "thread-ok", + "2026-01-01T00:00:00Z", + ) + .unwrap(); let step = crate::openhuman::flows::FlowRunStep { node_id: "n1".to_string(), output: json!({ "sent": 3 }), @@ -809,8 +821,14 @@ mod tests { let flow = flow_with_trigger_config("f-warn", true, json!({})); store::upsert_flow(&config, &flow).unwrap(); - store::insert_flow_run(&config, "run-warn", "f-warn", "thread-warn", "2026-01-01T00:00:00Z") - .unwrap(); + store::insert_flow_run( + &config, + "run-warn", + "f-warn", + "thread-warn", + "2026-01-01T00:00:00Z", + ) + .unwrap(); store::finish_flow_run( &config, "run-warn", diff --git a/src/openhuman/flows/memory_tools.rs b/src/openhuman/flows/memory_tools.rs index 4e3ce27681..fba08bbba9 100644 --- a/src/openhuman/flows/memory_tools.rs +++ b/src/openhuman/flows/memory_tools.rs @@ -139,10 +139,7 @@ impl Tool for FlowMemoryRecallTool { if flow_id.is_empty() { return Err(anyhow::anyhow!("flow_id cannot be empty")); } - let scope = args - .get("scope") - .and_then(|v| v.as_str()) - .unwrap_or("flow"); + let scope = args.get("scope").and_then(|v| v.as_str()).unwrap_or("flow"); #[allow(clippy::cast_possible_truncation)] let limit = args @@ -547,11 +544,7 @@ mod tests { // Never lands in another flow's namespace, the shared "flows" scope // namespace, or global/user memory. - assert!(mem - .get(&flow_namespace("f2"), "k") - .await - .unwrap() - .is_none()); + assert!(mem.get(&flow_namespace("f2"), "k").await.unwrap().is_none()); assert!(mem.get("global", "k").await.unwrap().is_none()); assert!(mem .get("f1", "k") // raw flow_id, not the derived namespace @@ -610,9 +603,7 @@ mod tests { async fn remember_missing_content_errs() { let (_tmp, mem) = test_mem(); let tool = FlowMemoryRememberTool::new(mem, test_security()); - let result = tool - .execute(json!({"flow_id": "f1", "key": "k"})) - .await; + let result = tool.execute(json!({"flow_id": "f1", "key": "k"})).await; assert!(result.is_err()); } } diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index 4c6daab733..4570f504e1 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -1227,9 +1227,14 @@ async fn flows_delete_clears_flow_memory_namespace() { crate::openhuman::memory::global::init(config.workspace_dir.clone()) .expect("init test memory client"); - let created = flows_create(&config, "with-memory".to_string(), trigger_only_graph(), false) - .await - .unwrap(); + let created = flows_create( + &config, + "with-memory".to_string(), + trigger_only_graph(), + false, + ) + .await + .unwrap(); let flow_id = created.value.id.clone(); let client = crate::openhuman::memory::global::client().expect("global client ready"); diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index dcd7dd6891..0455971a5f 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -429,7 +429,10 @@ pub fn all_tools_with_runtime( #[cfg(feature = "flows")] Box::new(FlowMemoryRecallTool::new(memory.clone())), #[cfg(feature = "flows")] - Box::new(FlowMemoryRememberTool::new(memory.clone(), security.clone())), + Box::new(FlowMemoryRememberTool::new( + memory.clone(), + security.clone(), + )), // Wallet tools — expose wallet operations to the agent tool-call pipeline // so the crypto sub-agent can prepare transfers, check status, etc. // Gated with the `web3` feature (the wallet domain is compiled out when From 6cbd93f24d38eec11d5e703a19daa4a1273fccbe Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Fri, 24 Jul 2026 15:32:39 +0530 Subject: [PATCH 3/4] fix(flows): make the digest subscriber's memory injectable so its tests are isolated MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The FlowRunDigestSubscriber tests bound the process-global memory client via memory::global::init(tempdir), then read the digest back through memory::global::client(). But the global is a one-shot OnceLock: when an earlier test in the same binary has already initialised it, init(tempdir) silently no-ops (memory::global's own tests document this), so the subscriber wrote to — and the test read from — a contaminated/dropped workspace. digest_writes_run_digest_entry_for_completed_run then found no entry and panicked. Add a test-only memory-injection seam (memory_override + with_memory + resolve_memory); production is unchanged (None -> active_memory_client, the process-global). The four digest tests now inject a directly-constructed UnifiedMemory(NoopEmbedding) — the same shape flows::memory_tools' tests use — and read back through that same instance, so write/read are deterministic and isolated from the global singleton. --- src/openhuman/flows/bus.rs | 93 ++++++++++++++++++++++++++------------ 1 file changed, 65 insertions(+), 28 deletions(-) diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index d296f10f10..08c449676a 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -282,11 +282,51 @@ const DIGEST_RETENTION_CAP: usize = 50; /// a memory-layer hiccup must never retroactively affect run status. pub struct FlowRunDigestSubscriber { config: Arc, + /// Test-only memory override. In production this is `None` and the digest + /// resolves the process-global memory client via [`active_memory_client`]. + /// The process-global client is a one-shot `OnceLock`, so a unit test + /// cannot reliably rebind it to its own tempdir (an earlier test in the + /// same binary may already have initialised the singleton — see + /// `memory::global`'s own test notes). Injecting a directly-constructed + /// [`Memory`] here lets the digest tests write and read back through the + /// SAME instance deterministically, exactly as `flows::memory_tools`' + /// tests do with `UnifiedMemory::new`. + memory_override: Option>, } impl FlowRunDigestSubscriber { pub fn new(config: Arc) -> Self { - Self { config } + Self { + config, + memory_override: None, + } + } + + /// Test constructor: run the digest against an explicitly-provided memory + /// instance instead of the process-global client. See [`Self::memory_override`]. + #[cfg(test)] + fn with_memory(config: Arc, memory: Arc) -> Self { + Self { + config, + memory_override: Some(memory), + } + } + + /// Resolves the memory handle the digest writes to: the injected test + /// override when present, else the process-global client + /// ([`active_memory_client`]). Returns `None` (best-effort skip) when the + /// global client is unavailable. + async fn resolve_memory(&self) -> Option> { + if let Some(memory) = &self.memory_override { + return Some(memory.clone()); + } + match crate::openhuman::memory::ops::helpers::active_memory_client().await { + Ok(client) => Some(client.memory_handle()), + Err(e) => { + tracing::warn!(target: "flows", error = %e, "[flows] digest: memory client unavailable — skipping"); + None + } + } } async fn handle_finished(&self, flow_id: &str, run_id: &str, status: &str) { @@ -321,14 +361,9 @@ impl FlowRunDigestSubscriber { let digest = render_run_digest(&flow_name, &run); - let client = match crate::openhuman::memory::ops::helpers::active_memory_client().await { - Ok(client) => client, - Err(e) => { - tracing::warn!(target: "flows", %flow_id, %run_id, error = %e, "[flows] digest: memory client unavailable — skipping"); - return; - } + let Some(memory) = self.resolve_memory().await else { + return; }; - let memory: Arc = client.memory_handle(); let namespace = flow_namespace(flow_id); let digest_key = format!("run_digest:{run_id}"); @@ -435,10 +470,24 @@ fn render_run_digest(flow_name: &str, run: &FlowRun) -> String { #[cfg(test)] mod tests { use super::*; + use crate::openhuman::embeddings::NoopEmbedding; use crate::openhuman::flows::Flow; + use crate::openhuman::memory_store::UnifiedMemory; use serde_json::json; use tinyflows::model::{Node, NodeKind, WorkflowGraph}; + /// A directly-constructed, isolated [`Memory`] for the digest tests — NOT + /// the process-global `OnceLock` client. The global is one-shot, so an + /// earlier test in the same binary may already have bound it to a different + /// workspace, making `global::init(..)` here a silent no-op (see + /// `memory::global`'s own test notes). Injecting this instance into the + /// subscriber via [`FlowRunDigestSubscriber::with_memory`] makes writes and + /// read-backs go through the SAME store deterministically — the same shape + /// `flows::memory_tools`' tests use. + fn digest_test_memory(tmp: &tempfile::TempDir) -> Arc { + Arc::new(UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap()) + } + fn test_config(tmp: &tempfile::TempDir) -> Arc { let config = Config { workspace_dir: tmp.path().join("workspace"), @@ -665,8 +714,7 @@ mod tests { async fn digest_ignores_failed_run() { let tmp = tempfile::TempDir::new().unwrap(); let config = test_config(&tmp); - crate::openhuman::memory::global::init(config.workspace_dir.clone()) - .expect("init test memory client"); + let memory = digest_test_memory(&tmp); let flow = flow_with_trigger_config("f-failed", true, json!({})); store::upsert_flow(&config, &flow).unwrap(); @@ -689,7 +737,7 @@ mod tests { ) .unwrap(); - let sub = FlowRunDigestSubscriber::new(config); + let sub = FlowRunDigestSubscriber::with_memory(config, memory.clone()); sub.handle(&DomainEvent::FlowRunFinished { flow_id: "f-failed".into(), run_id: "run-failed".into(), @@ -697,8 +745,6 @@ mod tests { }) .await; - let client = crate::openhuman::memory::global::client().expect("global client ready"); - let memory = client.memory_handle(); let entry = memory .get(&flow_namespace("f-failed"), "run_digest:run-failed") .await @@ -713,8 +759,7 @@ mod tests { async fn digest_ignores_cancelled_run() { let tmp = tempfile::TempDir::new().unwrap(); let config = test_config(&tmp); - crate::openhuman::memory::global::init(config.workspace_dir.clone()) - .expect("init test memory client"); + let memory = digest_test_memory(&tmp); let flow = flow_with_trigger_config("f-cancelled", true, json!({})); store::upsert_flow(&config, &flow).unwrap(); @@ -737,7 +782,7 @@ mod tests { ) .unwrap(); - let sub = FlowRunDigestSubscriber::new(config); + let sub = FlowRunDigestSubscriber::with_memory(config, memory.clone()); sub.handle(&DomainEvent::FlowRunFinished { flow_id: "f-cancelled".into(), run_id: "run-cancelled".into(), @@ -745,8 +790,6 @@ mod tests { }) .await; - let client = crate::openhuman::memory::global::client().expect("global client ready"); - let memory = client.memory_handle(); let entry = memory .get(&flow_namespace("f-cancelled"), "run_digest:run-cancelled") .await @@ -758,8 +801,7 @@ mod tests { async fn digest_writes_run_digest_entry_for_completed_run() { let tmp = tempfile::TempDir::new().unwrap(); let config = test_config(&tmp); - crate::openhuman::memory::global::init(config.workspace_dir.clone()) - .expect("init test memory client"); + let memory = digest_test_memory(&tmp); let flow = flow_with_trigger_config("f-ok", true, json!({})); store::upsert_flow(&config, &flow).unwrap(); @@ -790,7 +832,7 @@ mod tests { ) .unwrap(); - let sub = FlowRunDigestSubscriber::new(config); + let sub = FlowRunDigestSubscriber::with_memory(config, memory.clone()); sub.handle(&DomainEvent::FlowRunFinished { flow_id: "f-ok".into(), run_id: "run-ok".into(), @@ -798,8 +840,6 @@ mod tests { }) .await; - let client = crate::openhuman::memory::global::client().expect("global client ready"); - let memory = client.memory_handle(); let entry = memory .get(&flow_namespace("f-ok"), "run_digest:run-ok") .await @@ -816,8 +856,7 @@ mod tests { async fn digest_treats_completed_with_warnings_as_success() { let tmp = tempfile::TempDir::new().unwrap(); let config = test_config(&tmp); - crate::openhuman::memory::global::init(config.workspace_dir.clone()) - .expect("init test memory client"); + let memory = digest_test_memory(&tmp); let flow = flow_with_trigger_config("f-warn", true, json!({})); store::upsert_flow(&config, &flow).unwrap(); @@ -840,7 +879,7 @@ mod tests { ) .unwrap(); - let sub = FlowRunDigestSubscriber::new(config); + let sub = FlowRunDigestSubscriber::with_memory(config, memory.clone()); sub.handle(&DomainEvent::FlowRunFinished { flow_id: "f-warn".into(), run_id: "run-warn".into(), @@ -848,8 +887,6 @@ mod tests { }) .await; - let client = crate::openhuman::memory::global::client().expect("global client ready"); - let memory = client.memory_handle(); let entry = memory .get(&flow_namespace("f-warn"), "run_digest:run-warn") .await From c5d36c6399550e801819daa7a3fa8224f14b2c30 Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Fri, 24 Jul 2026 16:52:33 +0530 Subject: [PATCH 4/4] =?UTF-8?q?fix(flows):=20address=20review=20=E2=80=94?= =?UTF-8?q?=20bind=20flow=20memory=20to=20trusted=20run=20origin,=20isolat?= =?UTF-8?q?e=20delete=20test,=20relocate=20namespace=20helper?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Codex P1 (SECURITY, cross-flow write): FlowMemoryRememberTool and FlowMemoryRecallTool's scope:"flow" case now resolve the governing flow id from the run's AgentTurnOrigin::TrustedAutomation { job_id, source: TrustedAutomationSource::Workflow { .. } } task-local when present, ignoring any model-supplied flow_id arg outright. A prompt-injected caller can no longer name a different flow's id to poison its namespace. Falls back to the flow_id arg (still required) only when invoked outside a flow run. Added a unit test scoping a mismatched flow_id arg inside a Workflow origin and asserting the write lands only in the trusted flow's namespace. - Greptile P1 (test isolation): flows_delete_clears_flow_memory_namespace relied on the memory::global OnceLock singleton, racy across parallel tests in the same binary. flows_delete now delegates to a private flows_delete_impl(config, id, memory_client_override) — mirroring FlowRunDigestSubscriber::with_memory — and the test injects a directly-constructed MemoryClient instead of the global, seeding and reading back through the same instance the delete call clears. - Codex P2 (module shape): moved FLOW_MEMORY_NAMESPACE_PREFIX, flow_namespace, and their unit tests out of flows::mod.rs into memory_tools.rs (the domain sibling that owns the tools consuming them), re-exporting from mod.rs so every existing flows::flow_namespace / flows::FLOW_MEMORY_NAMESPACE_PREFIX call site keeps resolving unchanged. - Greptile P2 (clarity): commented why FlowRunDigestSubscriber::domains() returns ["cron"] — FlowRunFinished is tagged "cron" by DomainEvent::domain(). --- src/openhuman/flows/bus.rs | 4 + src/openhuman/flows/memory_tools.rs | 228 ++++++++++++++++++++++++---- src/openhuman/flows/mod.rs | 67 +------- src/openhuman/flows/ops.rs | 25 ++- src/openhuman/flows/ops_tests.rs | 23 ++- 5 files changed, 254 insertions(+), 93 deletions(-) diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index 08c449676a..ffb4b402d1 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -420,6 +420,10 @@ impl EventHandler for FlowRunDigestSubscriber { } fn domains(&self) -> Option<&[&str]> { + // `FlowRunFinished` — the only event this subscriber handles — is + // itself tagged `"cron"` by `DomainEvent::domain()` (grouped there + // with the other flow-run/schedule events), not `"flows"`. This is + // matching that tag, not a typo. Some(&["cron"]) } diff --git a/src/openhuman/flows/memory_tools.rs b/src/openhuman/flows/memory_tools.rs index fba08bbba9..a275ed39ea 100644 --- a/src/openhuman/flows/memory_tools.rs +++ b/src/openhuman/flows/memory_tools.rs @@ -1,5 +1,5 @@ //! Agent tools giving a running flow a private, sandboxed memory namespace -//! (`flow_` — see [`super::flow_namespace`]). +//! (`flow_` — see [`flow_namespace`]). //! //! Motivating use case: a newsletter-digest flow that runs on a schedule //! needs to remember which items it already sent so it doesn't re-send them @@ -11,7 +11,7 @@ //! **Security invariant (non-negotiable):** there is no code path here by //! which a flow can write to — or even name — a namespace other than its //! own. [`FlowMemoryRememberTool`] derives the namespace internally via -//! [`super::flow_namespace`] from the caller-supplied `flow_id`; there is no +//! [`flow_namespace`] from the caller-supplied `flow_id`; there is no //! `namespace` parameter a caller could override. Every write is tainted //! [`MemoryTaint::ExternalSync`] (automation output, not user-authored //! fact), matching the same taint sync pipelines use for third-party @@ -25,22 +25,94 @@ use std::sync::Arc; use async_trait::async_trait; use serde_json::json; -use crate::openhuman::flows::flow_namespace; +use crate::openhuman::agent::turn_origin::{self, AgentTurnOrigin, TrustedAutomationSource}; use crate::openhuman::memory::{Memory, MemoryCategory, MemoryEntry, MemoryTaint, RecallOpts}; use crate::openhuman::security::policy::ToolOperation; use crate::openhuman::security::SecurityPolicy; use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; +/// Returns the flow id the *run itself* is scoped under, when the current +/// agent turn is executing inside a saved-flow run +/// (`AgentTurnOrigin::TrustedAutomation { job_id, source: Workflow { .. } }` — +/// see `flows::ops::workflow_origin`, scoped around every `flows_run` / +/// `flows_resume`). `job_id` on that variant IS the running flow's id. +/// +/// **Security invariant:** this is the ONLY trustworthy source of "which flow +/// is calling". A `flow_id` value handed in as an ordinary tool argument is +/// model-supplied and can be forged by a prompt-injected caller (or another +/// agent invoking the tool directly) to name a DIFFERENT flow's namespace. +/// When this returns `Some`, callers MUST use it — and ignore any +/// caller-supplied `flow_id` arg — for the `scope: "flow"` / write case. +/// Callers running outside a flow run (e.g. a chat agent with the tool +/// wired in some other context) get `None` here and fall back to requiring +/// the `flow_id` arg, exactly as before this invariant existed. +fn trusted_flow_id() -> Option { + match turn_origin::current() { + Some(AgentTurnOrigin::TrustedAutomation { + job_id, + source: TrustedAutomationSource::Workflow { .. }, + }) => Some(job_id), + _ => None, + } +} + +/// Prefix for a flow's private, sandboxed memory namespace (see +/// [`flow_namespace`]). +/// +/// **Deviates from the originally specced `"flow:"` (colon) separator — +/// deliberately.** The `Memory` trait's `UnifiedMemory` backend +/// (`src/openhuman/memory_store/`) is internally inconsistent about +/// namespace sanitization: `store_with_taint`/`recall`/`list`/ +/// `MemoryClient::clear_namespace` all route through +/// `UnifiedMemory::sanitize_namespace` +/// (`memory_store/namespace_store/init.rs`), which collapses any character +/// outside `[A-Za-z0-9_/-]` — including `:` — to `_` before touching SQLite. +/// But `Memory::forget` (`memory_store/memory_trait.rs`) queries +/// `WHERE namespace = ?1` against the **raw, unsanitized** argument. With a +/// `"flow:"` prefix, `forget("flow:", key)` would therefore silently +/// never match the row `store_with_taint` actually persisted as +/// `"flow_"` — the post-run digest subscriber's retention sweep +/// (`bus::FlowRunDigestSubscriber`) would then never evict old entries, and +/// `namespace_summaries()`-based cross-flow listing (`scope: "flows"` in +/// [`FlowMemoryRecallTool`]) would have to match the sanitized form anyway +/// since `namespace_summaries` reads the persisted (sanitized) column back +/// verbatim. Using `"flow_"` (already a fixed point of `sanitize_namespace`, +/// since flow ids are hyphenated UUIDs — no character in either the prefix +/// or a flow id ever needs sanitizing) makes every `Memory` method agree +/// with every other one on the exact namespace string, with no silent +/// mismatch anywhere. The namespace is still shared-root and +/// profile-independent exactly as specified — only the separator character +/// changed. +/// +/// Re-exported from `flows::mod` as `flows::FLOW_MEMORY_NAMESPACE_PREFIX` — +/// see that module for why this lives here rather than in `mod.rs` itself. +pub const FLOW_MEMORY_NAMESPACE_PREFIX: &str = "flow_"; + +/// Builds a flow's private, profile-independent memory namespace from a +/// `flow_id`. +/// +/// **Security invariant:** this is the *only* place in the codebase that may +/// construct this namespace string. Every caller — the `flow_memory_recall` +/// / `flow_memory_remember` agent tools below, the post-run digest +/// subscriber (`bus::FlowRunDigestSubscriber`), and the `flows_delete` +/// cleanup hook (`ops::flows_delete`) — goes through this function with a +/// `flow_id`, never with a caller-supplied raw namespace. A flow can +/// therefore never write to, or be told the name of, any memory namespace +/// but its own. +/// +/// Re-exported from `flows::mod` as `flows::flow_namespace`. +pub fn flow_namespace(flow_id: &str) -> String { + format!("{FLOW_MEMORY_NAMESPACE_PREFIX}{flow_id}") +} + /// The persisted-namespace prefix matching every flow's memory namespace, as /// [`Memory::namespace_summaries`] returns it. /// -/// This is intentionally the *same* string as -/// [`super::FLOW_MEMORY_NAMESPACE_PREFIX`] (see that constant's doc comment -/// for why the separator is `_` rather than the originally specced `:`) — -/// kept as a separate, explicitly-named constant here so the "match against +/// This is intentionally the *same* string as [`FLOW_MEMORY_NAMESPACE_PREFIX`] +/// — kept as a separate, explicitly-named constant here so the "match against /// what recall/list see" intent stays self-evident at each call site, /// independent of whether the two ever need to diverge in the future. -const FLOW_MEMORY_NAMESPACE_LISTED_PREFIX: &str = super::FLOW_MEMORY_NAMESPACE_PREFIX; +const FLOW_MEMORY_NAMESPACE_LISTED_PREFIX: &str = FLOW_MEMORY_NAMESPACE_PREFIX; /// Read-only recall over a flow's own memory namespace, or (with /// `scope: "flows"`) across every flow's namespace. @@ -106,7 +178,10 @@ impl Tool for FlowMemoryRecallTool { }, "flow_id": { "type": "string", - "description": "The calling flow's id" + "description": "The calling flow's id. Inside a running flow this is informational \ + only: the active flow's own id (from the run's trusted origin) is authoritative and \ + any value supplied here is ignored. Required only when this tool is invoked outside \ + a flow run (e.g. from a chat agent)." }, "scope": { "type": "string", @@ -131,14 +206,35 @@ impl Tool for FlowMemoryRecallTool { if query.is_empty() { return Err(anyhow::anyhow!("query cannot be empty")); } - let flow_id = args - .get("flow_id") - .and_then(|v| v.as_str()) - .ok_or_else(|| anyhow::anyhow!("Missing 'flow_id' parameter"))? - .trim(); - if flow_id.is_empty() { - return Err(anyhow::anyhow!("flow_id cannot be empty")); - } + + let flow_id_arg = args.get("flow_id").and_then(|v| v.as_str()).map(str::trim); + + // SECURITY: inside a running flow, the run's own trusted origin is + // the ONLY authoritative source for "which flow is calling" — never + // the model-supplied `flow_id` arg. Without this, a prompt-injected + // caller could pass a different flow's id and read across the + // sandbox boundary the module doc promises. See `trusted_flow_id`. + let trusted = trusted_flow_id(); + let flow_id: String = match &trusted { + Some(trusted_id) => { + tracing::debug!( + target: "flows", + flow_id = %trusted_id, + "[flows:memory] flow_memory_recall: flow id resolved from the trusted Workflow \ + run origin (any model-supplied flow_id arg is ignored)" + ); + trusted_id.clone() + } + None => { + let arg = + flow_id_arg.ok_or_else(|| anyhow::anyhow!("Missing 'flow_id' parameter"))?; + if arg.is_empty() { + return Err(anyhow::anyhow!("flow_id cannot be empty")); + } + arg.to_string() + } + }; + let flow_id = flow_id.as_str(); let scope = args.get("scope").and_then(|v| v.as_str()).unwrap_or("flow"); #[allow(clippy::cast_possible_truncation)] @@ -242,7 +338,10 @@ impl Tool for FlowMemoryRememberTool { "properties": { "flow_id": { "type": "string", - "description": "The calling flow's id" + "description": "The calling flow's id. Inside a running flow this is informational \ + only: the active flow's own id (from the run's trusted origin) is authoritative and \ + any value supplied here is ignored. Required only when this tool is invoked outside \ + a flow run (e.g. from a chat agent)." }, "key": { "type": "string", @@ -266,10 +365,7 @@ impl Tool for FlowMemoryRememberTool { } async fn execute(&self, args: serde_json::Value) -> anyhow::Result { - let flow_id = args - .get("flow_id") - .and_then(|v| v.as_str()) - .ok_or_else(|| anyhow::anyhow!("Missing 'flow_id' parameter"))?; + let flow_id_arg = args.get("flow_id").and_then(|v| v.as_str()); let key = args .get("key") .and_then(|v| v.as_str()) @@ -298,10 +394,34 @@ impl Tool for FlowMemoryRememberTool { return Ok(ToolResult::error(error)); } - let flow_id = flow_id.trim(); - if flow_id.is_empty() { - return Ok(ToolResult::error("flow_id cannot be empty".to_string())); - } + // SECURITY: resolve the namespace-governing flow id from the run's + // trusted origin when available — NEVER from the model-supplied + // `flow_id` arg. Without this, a prompt-injected caller (or another + // agent invoking this tool directly) could pass a DIFFERENT flow's + // id here and poison that flow's private namespace. See + // `trusted_flow_id`'s doc comment for the mechanism. + let trusted = trusted_flow_id(); + let flow_id: String = match &trusted { + Some(trusted_id) => { + tracing::debug!( + target: "flows", + flow_id = %trusted_id, + "[flows:memory] flow_memory_remember: flow id resolved from the trusted Workflow \ + run origin (any model-supplied flow_id arg is ignored)" + ); + trusted_id.clone() + } + None => { + let arg = + flow_id_arg.ok_or_else(|| anyhow::anyhow!("Missing 'flow_id' parameter"))?; + let trimmed = arg.trim(); + if trimmed.is_empty() { + return Ok(ToolResult::error("flow_id cannot be empty".to_string())); + } + trimmed.to_string() + } + }; + let flow_id = flow_id.as_str(); let key = key.trim(); if key.is_empty() { return Ok(ToolResult::error("key cannot be empty".to_string())); @@ -365,6 +485,20 @@ mod tests { (tmp, Arc::new(mem)) } + // ── flow_namespace / FLOW_MEMORY_NAMESPACE_PREFIX ─────────────── + // (relocated from `flows::mod` — see that module's re-export comment) + + #[test] + fn flow_namespace_uses_the_shared_root_prefix() { + assert_eq!(flow_namespace("abc-123"), "flow_abc-123"); + assert!(flow_namespace("abc-123").starts_with(FLOW_MEMORY_NAMESPACE_PREFIX)); + } + + #[test] + fn flow_namespace_is_distinct_per_flow() { + assert_ne!(flow_namespace("a"), flow_namespace("b")); + } + // ── FlowMemoryRecallTool ──────────────────────────────────────── #[test] @@ -553,6 +687,48 @@ mod tests { .is_none()); } + /// SECURITY (Fix 1): a running flow's own id — carried by the run's + /// `TrustedAutomation { Workflow }` origin — is authoritative. A + /// mismatched, model-supplied `flow_id` arg must be silently ignored, + /// never allowed to redirect the write into a different flow's + /// namespace. + #[tokio::test] + async fn remember_ignores_mismatched_flow_id_arg_inside_trusted_workflow_run() { + let (_tmp, mem) = test_mem(); + let tool = FlowMemoryRememberTool::new(mem.clone(), test_security()); + + let origin = AgentTurnOrigin::TrustedAutomation { + job_id: "f-real".to_string(), + source: TrustedAutomationSource::Workflow { + require_approval: false, + }, + }; + let result = turn_origin::with_origin( + origin, + tool.execute(json!({ + "flow_id": "f-other", + "key": "sent_item_1", + "content": "Sent item 1" + })), + ) + .await + .unwrap(); + assert!(!result.is_error); + + // Landed in the trusted run's own namespace... + assert!(mem + .get(&flow_namespace("f-real"), "sent_item_1") + .await + .unwrap() + .is_some()); + // ...and NOT the mismatched, model-supplied namespace. + assert!(mem + .get(&flow_namespace("f-other"), "sent_item_1") + .await + .unwrap() + .is_none()); + } + #[tokio::test] async fn remember_blocked_in_readonly_autonomy() { let (_tmp, mem) = test_mem(); diff --git a/src/openhuman/flows/mod.rs b/src/openhuman/flows/mod.rs index ab1d1658b4..e9df7f8fc8 100644 --- a/src/openhuman/flows/mod.rs +++ b/src/openhuman/flows/mod.rs @@ -44,63 +44,10 @@ pub use types::{ DraftOrigin, Flow, FlowConnection, FlowDraft, FlowImport, FlowRevision, FlowRun, FlowRunStep, FlowRunTrigger, FlowSuggestion, FlowValidation, FlowValidationError, SuggestionStatus, }; - -/// Prefix for a flow's private, sandboxed memory namespace (see -/// [`flow_namespace`]). -/// -/// **Deviates from the originally specced `"flow:"` (colon) separator — -/// deliberately.** The `Memory` trait's `UnifiedMemory` backend -/// (`src/openhuman/memory_store/`) is internally inconsistent about -/// namespace sanitization: `store_with_taint`/`recall`/`list`/ -/// `MemoryClient::clear_namespace` all route through -/// `UnifiedMemory::sanitize_namespace` -/// (`memory_store/namespace_store/init.rs`), which collapses any character -/// outside `[A-Za-z0-9_/-]` — including `:` — to `_` before touching SQLite. -/// But `Memory::forget` (`memory_store/memory_trait.rs`) queries -/// `WHERE namespace = ?1` against the **raw, unsanitized** argument. With a -/// `"flow:"` prefix, `forget("flow:", key)` would therefore silently -/// never match the row `store_with_taint` actually persisted as -/// `"flow_"` — the post-run digest subscriber's retention sweep -/// (`bus::FlowRunDigestSubscriber`) would then never evict old entries, and -/// `namespace_summaries()`-based cross-flow listing (`scope: "flows"` in -/// `memory_tools::FlowMemoryRecallTool`) would have to match the sanitized -/// form anyway since `namespace_summaries` reads the persisted (sanitized) -/// column back verbatim. Using `"flow_"` (already a fixed point of -/// `sanitize_namespace`, since flow ids are hyphenated UUIDs — no character -/// in either the prefix or a flow id ever needs sanitizing) makes every -/// `Memory` method agree with every other one on the exact namespace -/// string, with no silent mismatch anywhere. The namespace is still -/// shared-root and profile-independent exactly as specified — only the -/// separator character changed. -pub const FLOW_MEMORY_NAMESPACE_PREFIX: &str = "flow_"; - -/// Builds a flow's private, profile-independent memory namespace from a -/// `flow_id`. -/// -/// **Security invariant:** this is the *only* place in the codebase that may -/// construct this namespace string. Every caller — the `flow_memory_recall` -/// / `flow_memory_remember` agent tools (`memory_tools.rs`), the post-run -/// digest subscriber (`bus::FlowRunDigestSubscriber`), and the -/// `flows_delete` cleanup hook (`ops::flows_delete`) — goes through this -/// function with a `flow_id`, never with a caller-supplied raw namespace. -/// A flow can therefore never write to, or be told the name of, any memory -/// namespace but its own. -pub fn flow_namespace(flow_id: &str) -> String { - format!("{FLOW_MEMORY_NAMESPACE_PREFIX}{flow_id}") -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn flow_namespace_uses_the_shared_root_prefix() { - assert_eq!(flow_namespace("abc-123"), "flow_abc-123"); - assert!(flow_namespace("abc-123").starts_with(FLOW_MEMORY_NAMESPACE_PREFIX)); - } - - #[test] - fn flow_namespace_is_distinct_per_flow() { - assert_ne!(flow_namespace("a"), flow_namespace("b")); - } -} +// `FLOW_MEMORY_NAMESPACE_PREFIX` / `flow_namespace` live in `memory_tools` +// (the domain logic sibling that owns the agent tools consuming them) and are +// re-exported here so every existing `flows::flow_namespace` / +// `flows::FLOW_MEMORY_NAMESPACE_PREFIX` call site (`bus.rs`, `ops.rs`, this +// module's own doc comments) keeps resolving unchanged — `mod.rs` stays +// export-focused only, per this repo's canonical module shape. +pub use memory_tools::{flow_namespace, FLOW_MEMORY_NAMESPACE_PREFIX}; diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 6834f66473..0d0abfc884 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -24,6 +24,7 @@ use crate::openhuman::flows::types::{ FlowConnection, FlowRunStep, FlowRunTrigger, FlowSuggestion, SuggestionStatus, }; use crate::openhuman::flows::{flow_namespace, Flow, FlowRun}; +use crate::openhuman::memory_store::MemoryClientRef; use crate::rpc::RpcOutcome; /// Overall safety bound on a single `flows_run` / `flows_resume`. Individual @@ -3420,6 +3421,24 @@ pub async fn flows_rollback( /// itself — `store::remove_flow` below still errors clearly if `id` doesn't /// exist. pub async fn flows_delete(config: &Config, id: &str) -> Result, String> { + flows_delete_impl(config, id, None).await +} + +/// Backs [`flows_delete`]. `memory_client_override`, when `Some`, is used in +/// place of the process-global memory client for the namespace-clear step +/// below — mirrors `bus::FlowRunDigestSubscriber`'s `with_memory` seam. +/// +/// The process-global client (`memory::global`) is a single shared `OnceLock` +/// that any test in the binary may rebind to its own tempdir workspace, so a +/// test asserting this clear step deterministically must not depend on it — +/// injecting a directly-constructed [`MemoryClientRef`] lets the test seed +/// and read back through the SAME instance `flows_delete` itself writes to, +/// with no race against the global. +async fn flows_delete_impl( + config: &Config, + id: &str, + memory_client_override: Option, +) -> Result, String> { match store::get_flow(config, id) { Ok(Some(flow)) => unbind_trigger(config, &flow), Ok(None) => {} @@ -3436,7 +3455,11 @@ pub async fn flows_delete(config: &Config, id: &str) -> Result // entries or run digests behind. Never fails the delete itself: the flow // row is already gone by this point regardless of what happens here. let memory_namespace = flow_namespace(id); - match crate::openhuman::memory::ops::helpers::active_memory_client().await { + let client_result = match memory_client_override { + Some(client) => Ok(client), + None => crate::openhuman::memory::ops::helpers::active_memory_client().await, + }; + match client_result { Ok(client) => { if let Err(e) = client.clear_namespace(&memory_namespace).await { tracing::warn!(target: "flows", flow_id = %id, namespace = %memory_namespace, error = %e, "[flows] flows_delete: failed to clear flow memory namespace"); diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index de6047da03..d1afe24402 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -1221,11 +1221,21 @@ async fn flows_delete_unbinds_schedule_cron_job() { #[tokio::test] async fn flows_delete_clears_flow_memory_namespace() { use crate::openhuman::memory::{Memory, MemoryCategory, MemoryTaint}; + use crate::openhuman::memory_store::MemoryClient; let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - crate::openhuman::memory::global::init(config.workspace_dir.clone()) - .expect("init test memory client"); + + // A directly-constructed `MemoryClient`, injected via `flows_delete_impl` + // below, instead of `memory::global` — that singleton is a single + // process-wide `OnceLock` any other test in this binary may rebind to + // its own tempdir workspace, which would make this test's pass/fail + // depend on run order / thread interleaving rather than its own setup. + // See `flows_delete_impl`'s doc comment (mirrors + // `bus::FlowRunDigestSubscriber::with_memory`'s injection seam). + let memory_client: MemoryClientRef = + Arc::new(MemoryClient::from_workspace_dir(config.workspace_dir.clone()).unwrap()); + let memory = memory_client.memory_handle(); let created = flows_create( &config, @@ -1237,8 +1247,6 @@ async fn flows_delete_clears_flow_memory_namespace() { .unwrap(); let flow_id = created.value.id.clone(); - let client = crate::openhuman::memory::global::client().expect("global client ready"); - let memory = client.memory_handle(); memory .store_with_taint( &flow_namespace(&flow_id), @@ -1256,10 +1264,13 @@ async fn flows_delete_clears_flow_memory_namespace() { .await .unwrap() .is_some(), - "precondition: flow memory entry was stored" + "precondition: flow memory entry was stored (through the SAME client flows_delete_impl \ + is about to clear)" ); - flows_delete(&config, &flow_id).await.unwrap(); + flows_delete_impl(&config, &flow_id, Some(memory_client)) + .await + .unwrap(); assert!( memory