mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(flows): per-flow memory namespace + post-run digest (flow:<id>) (#5176)
This commit is contained in:
Cyrus Gray
@@ -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");
|
||||
|
||||
+475
-1
@@ -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<Config>,
|
||||
/// 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<Arc<dyn Memory>>,
|
||||
}
|
||||
|
||||
impl FlowRunDigestSubscriber {
|
||||
pub fn new(config: Arc<Config>) -> 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<Config>, memory: Arc<dyn Memory>) -> 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<Arc<dyn Memory>> {
|
||||
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<dyn Memory>, 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<dyn Memory> {
|
||||
Arc::new(UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap())
|
||||
}
|
||||
|
||||
fn test_config(tmp: &tempfile::TempDir) -> Arc<Config> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,785 @@
|
||||
//! Agent tools giving a running flow a private, sandboxed memory namespace
|
||||
//! (`flow_<flow_id>` — 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: <item id>", 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<String> {
|
||||
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:<id>", key)` would therefore silently
|
||||
/// never match the row `store_with_taint` actually persisted as
|
||||
/// `"flow_<id>"` — 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<dyn Memory>,
|
||||
}
|
||||
|
||||
impl FlowMemoryRecallTool {
|
||||
pub fn new(memory: Arc<dyn Memory>) -> 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<ToolResult> {
|
||||
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<MemoryEntry> = 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<dyn Memory>,
|
||||
security: Arc<SecurityPolicy>,
|
||||
}
|
||||
|
||||
impl FlowMemoryRememberTool {
|
||||
pub fn new(memory: Arc<dyn Memory>, security: Arc<SecurityPolicy>) -> 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<ToolResult> {
|
||||
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:<name>`
|
||||
// wire value resolves back to `Custom("<name>")` 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_<id>` 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<SecurityPolicy> {
|
||||
Arc::new(SecurityPolicy::default())
|
||||
}
|
||||
|
||||
fn test_mem() -> (TempDir, Arc<dyn Memory>) {
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -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};
|
||||
|
||||
@@ -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<RpcOutcome<Value>, 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<MemoryClientRef>,
|
||||
) -> Result<RpcOutcome<Value>, 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<RpcOutcome<Value>
|
||||
|
||||
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 }),
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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::*;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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),
|
||||
|
||||
Reference in New Issue
Block a user