diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 321619b5f..8aaf349f2 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 ec03b60a4..ffb4b402d 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,13 +258,240 @@ 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, + /// 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, + 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) { + 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 Some(memory) = self.resolve_memory().await else { + return; + }; + 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]> { + // `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"]) + } + + 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::*; + 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"), @@ -461,4 +689,250 @@ 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); + let memory = digest_test_memory(&tmp); + + 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::with_memory(config, memory.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-failed".into(), + run_id: "run-failed".into(), + status: "failed".into(), + }) + .await; + + 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); + let memory = digest_test_memory(&tmp); + + 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::with_memory(config, memory.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-cancelled".into(), + run_id: "run-cancelled".into(), + status: "cancelled".into(), + }) + .await; + + 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); + let memory = digest_test_memory(&tmp); + + 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::with_memory(config, memory.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-ok".into(), + run_id: "run-ok".into(), + status: "completed".into(), + }) + .await; + + 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); + let memory = digest_test_memory(&tmp); + + 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::with_memory(config, memory.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-warn".into(), + run_id: "run-warn".into(), + status: "completed_with_warnings".into(), + }) + .await; + + 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 000000000..a275ed39e --- /dev/null +++ b/src/openhuman/flows/memory_tools.rs @@ -0,0 +1,785 @@ +//! Agent tools giving a running flow a private, sandboxed memory 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 +//! 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 +//! [`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::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 [`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 = 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. 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", + "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_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)] + 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. 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", + "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_arg = args.get("flow_id").and_then(|v| v.as_str()); + 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)); + } + + // 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())); + } + + 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)) + } + + // ── 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] + 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()); + } + + /// 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(); + 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 2eddc37cd..e9df7f8fc 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,10 @@ pub use types::{ DraftOrigin, Flow, FlowConnection, FlowDraft, FlowImport, FlowRevision, FlowRun, FlowRunStep, FlowRunTrigger, FlowSuggestion, FlowValidation, FlowValidationError, SuggestionStatus, }; +// `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 93fac053a..d2b3f2dcc 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -23,7 +23,8 @@ 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::openhuman::memory_store::MemoryClientRef; use crate::rpc::RpcOutcome; /// Overall safety bound on a single `flows_run` / `flows_resume`. Individual @@ -3492,6 +3493,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) => {} @@ -3502,6 +3521,27 @@ 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); + 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"); + } + } + 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 5779ddb5f..75dadf00a 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -1218,6 +1218,70 @@ 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); + + // 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, + "with-memory".to_string(), + trigger_only_graph(), + false, + ) + .await + .unwrap(); + let flow_id = created.value.id.clone(); + + 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 (through the SAME client flows_delete_impl \ + is about to clear)" + ); + + flows_delete_impl(&config, &flow_id, Some(memory_client)) + .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 b25777df1..40275ecf5 100644 --- a/src/openhuman/tools/mod.rs +++ b/src/openhuman/tools/mod.rs @@ -30,6 +30,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 dfed55971..aba7262b6 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -396,6 +396,20 @@ 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 @@ -1299,7 +1313,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", @@ -1330,6 +1344,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 048caea95..a5dbd1a16 100644 --- a/src/openhuman/tools/ops_tests.rs +++ b/src/openhuman/tools/ops_tests.rs @@ -2437,6 +2437,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), @@ -2565,6 +2567,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),