From 48565b1c80890ed087932b37f3f6eb495187cc53 Mon Sep 17 00:00:00 2001 From: cyrus Date: Mon, 27 Jul 2026 23:35:59 +0530 Subject: [PATCH 1/4] feat(nodes): add memory node kind (recall/flavour/people + flow-scoped remember) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the 13th node kind, `memory`, as a delivery surface onto a host-injected `MemoryProvider` capability (recall/search/flavour/people reads, remember/forget writes). Enforces the hard security invariant at validate time: a remember/forget operation may never target scope "user" — writes are restricted to flow-scoped memory, so a workflow can never plant or erase durable facts about the user. Includes a shaped MockMemory (wired into mock_capabilities by default) so dry-run keeps working, and node/validate/ catalog test coverage. --- README.md | 5 +- src/caps/mock.rs | 128 +++++++- src/caps/mod.rs | 106 +++++- src/catalog.rs | 129 +++++++- src/model/node_kind.rs | 7 + src/nodes/integration/memory.rs | 555 ++++++++++++++++++++++++++++++++ src/nodes/integration/mod.rs | 6 +- src/nodes/mod.rs | 10 +- src/validate.rs | 343 ++++++++++++++++++++ 9 files changed, 1274 insertions(+), 15 deletions(-) create mode 100644 src/nodes/integration/memory.rs diff --git a/README.md b/README.md index 054d1e9..99ae120 100644 --- a/README.md +++ b/README.md @@ -35,8 +35,8 @@ Rust 2024 · MSRV 1.85 · `#![forbid(unsafe_code)]` · GPL-3.0-or-later. - Full node catalog implemented and tested — control-flow (`condition`, `switch`, `merge`, `split_out`, `transform`) and capability-backed (`agent`, - `tool_call`, `http_request`, `code`, `output_parser`, `sub_workflow`), plus the - `trigger` entry node. + `tool_call`, `http_request`, `code`, `output_parser`, `sub_workflow`, + `memory`), plus the `trigger` entry node. **Reliability** @@ -189,6 +189,7 @@ done | `code` | Runs sandboxed user code (JavaScript or Python). | | `output_parser` | Parses / validates an upstream agent's output into a structured shape. | | `sub_workflow` | Runs another workflow as a nested sub-graph and returns its output. | +| `memory` | Reads/writes host-managed memory (recall/search/flavour/people/remember/forget). | | `condition` | Two-way IF; emits on the `true` or `false` port. | | `switch` | Multi-way branch keyed by an expression result. | | `merge` | Fan-in barrier that combines multiple inputs; waits for all wired predecessors. | diff --git a/src/caps/mock.rs b/src/caps/mock.rs index 34abec7..db30618 100644 --- a/src/caps/mock.rs +++ b/src/caps/mock.rs @@ -10,8 +10,8 @@ use async_trait::async_trait; use serde_json::{Value, json}; use crate::caps::{ - AgentRunner, Capabilities, CodeLanguage, CodeRunner, HttpClient, LlmProvider, StateStore, - ToolInvoker, WorkflowResolver, + AgentRunner, Capabilities, CodeLanguage, CodeRunner, HttpClient, LlmProvider, MemoryProvider, + StateStore, ToolInvoker, WorkflowResolver, }; use crate::error::{EngineError, Result}; use crate::model::WorkflowGraph; @@ -77,6 +77,61 @@ impl CodeRunner for MockCode { } } +/// A [`MemoryProvider`] returning small, deterministic, **shaped** canned data. +/// +/// Unlike a naive echo mock, each method returns a plausible result shape (a +/// `results` array for recall/search, a `traits` object for flavour, a `people` +/// array for people lookups) rather than `null`/an empty object — the same +/// "shaped mock" precedent OpenHuman's `SchemaAwareMockAgentRunner` / +/// `SchemaAwareMockLlm` set, because a naive mock made dry-runs meaningless +/// (downstream `condition`/`transform` nodes had nothing to bind to). Wired +/// into [`mock_capabilities`] by default (not behind an opt-in helper, unlike +/// [`MockAgentRunner`]) precisely so a workflow containing a `memory` node +/// dry-runs out of the box. +#[derive(Debug, Default, Clone)] +pub struct MockMemory; + +#[async_trait] +impl MemoryProvider for MockMemory { + async fn recall(&self, scope: &str, query: &str, opts: Value) -> Result { + Ok(json!({ + "scope": scope, + "query": query, + "opts": opts, + "results": [ + { "id": "mem_1", "text": format!("mock memory matching '{query}'"), "score": 0.92 }, + { "id": "mem_2", "text": "a second mock memory", "score": 0.81 }, + ], + })) + } + + async fn flavour(&self, slug: &str) -> Result { + Ok(json!({ + "slug": slug, + "summary": format!("mock flavour profile for '{slug}'"), + "traits": { "tone": "warm", "formality": "casual" }, + })) + } + + async fn people(&self, query: Option<&str>) -> Result { + Ok(json!({ + "query": query, + "people": [ + { "id": "person_1", "name": "Mock Person A" }, + { "id": "person_2", "name": "Mock Person B" }, + ], + })) + } + + async fn remember(&self, _scope: &str, _key: &str, _value: Value) -> Result<()> { + Ok(()) + } + + async fn forget(&self, _scope: &str, _key: &str) -> Result<()> { + Ok(()) + } +} + /// A [`StateStore`] backed by an in-memory map guarded by a mutex. #[derive(Debug, Default)] pub struct MockStateStore { @@ -129,7 +184,11 @@ impl WorkflowResolver for MockWorkflowResolver { /// Builds a [`Capabilities`] bundle wired entirely to the mock implementations. /// /// The bundled [`MockWorkflowResolver`] is empty; use -/// [`mock_capabilities_with_resolver`] to supply one that resolves ids. +/// [`mock_capabilities_with_resolver`] to supply one that resolves ids. Unlike +/// [`Capabilities::agent`] (which defaults `None`), [`Capabilities::memory`] is +/// wired to [`MockMemory`] by default — a `memory` node must dry-run +/// successfully out of the box; use `Capabilities { memory: None, ..caps }` to +/// exercise the "host wired no memory store" error path instead. #[must_use] pub fn mock_capabilities() -> Capabilities { mock_capabilities_with_resolver(MockWorkflowResolver::default()) @@ -149,6 +208,8 @@ pub fn mock_capabilities_with_resolver(resolver: impl WorkflowResolver + 'static // No agent registry by default: `agent` nodes use `MockLlm`. Use // [`mock_capabilities_with_agent`] to exercise the `agent_ref` path. agent: None, + // Wired by default (unlike `agent`) — see the doc comment above. + memory: Some(Arc::new(MockMemory)), } } @@ -162,6 +223,17 @@ pub fn mock_capabilities_with_agent(agent: impl AgentRunner + 'static) -> Capabi } } +/// Like [`mock_capabilities`], but with a caller-supplied [`MemoryProvider`] in +/// place of the default [`MockMemory`] — for tests that need custom recall / +/// flavour / people / remember / forget behavior. +#[must_use] +pub fn mock_capabilities_with_memory(memory: impl MemoryProvider + 'static) -> Capabilities { + Capabilities { + memory: Some(Arc::new(memory)), + ..mock_capabilities() + } +} + #[cfg(test)] mod tests { use super::*; @@ -239,6 +311,47 @@ mod tests { assert_eq!(py["result"], json!([1, 2, 3])); } + #[tokio::test] + async fn mock_memory_recall_returns_shaped_results() { + let memory = MockMemory; + let out = memory + .recall("flow", "budget", json!({ "operation": "recall" })) + .await + .unwrap(); + assert_eq!(out["scope"], "flow"); + assert_eq!(out["query"], "budget"); + let results = out["results"].as_array().expect("results array"); + assert!(!results.is_empty(), "mock recall should return shaped results"); + assert!(results[0].get("text").is_some()); + } + + #[tokio::test] + async fn mock_memory_flavour_returns_shaped_object() { + let memory = MockMemory; + let out = memory.flavour("email-tone").await.unwrap(); + assert_eq!(out["slug"], "email-tone"); + assert!(out["traits"].is_object()); + } + + #[tokio::test] + async fn mock_memory_people_returns_shaped_list() { + let memory = MockMemory; + let out = memory.people(Some("cyrus")).await.unwrap(); + assert_eq!(out["query"], "cyrus"); + let people = out["people"].as_array().expect("people array"); + assert!(!people.is_empty(), "mock people should return a shaped list"); + + let out_no_query = memory.people(None).await.unwrap(); + assert!(out_no_query["query"].is_null()); + } + + #[tokio::test] + async fn mock_memory_remember_and_forget_are_ok() { + let memory = MockMemory; + memory.remember("flow", "k", json!({ "v": 1 })).await.unwrap(); + memory.forget("flow", "k").await.unwrap(); + } + #[tokio::test] async fn mock_state_store_round_trips_and_misses() { let store = MockStateStore::default(); @@ -285,5 +398,14 @@ mod tests { .unwrap()["result"], "x" ); + assert!( + caps.memory + .as_ref() + .expect("memory wired by default") + .flavour("f") + .await + .unwrap()["slug"] + == "f" + ); } } diff --git a/src/caps/mod.rs b/src/caps/mod.rs index 0a0d7b0..6bc74e1 100644 --- a/src/caps/mod.rs +++ b/src/caps/mod.rs @@ -57,6 +57,94 @@ pub trait AgentRunner: Send + Sync { -> Result; } +/// Host-injected memory access for `memory` nodes. +/// +/// A `memory` node is a **delivery surface** onto whatever durable memory store +/// the host already maintains — it exposes no capability of its own. Six config +/// `operation`s map onto five trait methods: +/// +/// - `recall` and `search` both call [`recall`](Self::recall); `search` passes +/// `opts.operation == "search"` (alongside the same `query`) so a host that +/// distinguishes semantic recall from full-text search can branch on it, but +/// a host that treats them identically may ignore the distinction entirely. +/// This keeps the trait small (one read-with-a-query method rather than two +/// near-duplicates) while still letting `opts` carry the distinction through +/// to the host. +/// - `flavour` and `people` are read-only lookups with no free-text `query` +/// requirement, each with its own method. +/// - `remember` and `forget` are the two write operations. +/// +/// `scope` is an opaque, host-defined string (the model layer never interprets +/// it) but the wire contract in practice is `"user"` (the caller's durable, +/// cross-flow memory — read-only from a workflow), `"flow"` (this flow's own +/// memory — the only scope `remember`/`forget` may target), and `"flows"` +/// (cross-flow read access — read-only). [`crate::validate`] enforces the hard +/// invariant that a `remember`/`forget` operation may never carry +/// `scope: "user"`, structurally, before a run ever starts — that boundary is +/// NOT re-checked here, so a host implementing this trait directly (bypassing +/// the node + validator) must not treat that check as already done. +/// +/// This capability is **optional**, matching the [`AgentRunner`] precedent: +/// hosts without a memory store leave [`Capabilities::memory`] `None`, and a +/// `memory` node then fails at run time with a capability error rather than +/// silently no-opping (there is no meaningful fallback for a read/write memory +/// call the way `agent` falls back to a bare completion). +#[async_trait] +pub trait MemoryProvider: Send + Sync { + /// Recalls memory matching `query` within `scope`. + /// + /// Backs both the `recall` and `search` node operations; `opts` carries + /// `"operation"` (`"recall"` or `"search"`) plus any of the node's optional + /// `limit` / `min_score` config, each present only when the author set it. + /// The return shape is host-defined — typically an object with a `results` + /// (or similar) array — and is passed through to the node's output + /// envelope unchanged. + /// + /// # Errors + /// Returns an [`EngineError::Capability`](crate::error::EngineError::Capability) + /// when the scope is unknown to the host or the lookup fails. + async fn recall(&self, scope: &str, query: &str, opts: Value) -> Result; + + /// Looks up a named "flavour" (a host-defined ask/persona/style profile, + /// e.g. `"email-tone"`) by its slug. + /// + /// # Errors + /// Returns an [`EngineError::Capability`](crate::error::EngineError::Capability) + /// when `slug` is unknown to the host. + async fn flavour(&self, slug: &str) -> Result; + + /// Looks up people the host's memory knows about, optionally narrowed by a + /// free-text `query`; `None` returns the host's default listing. + /// + /// # Errors + /// Returns an [`EngineError::Capability`](crate::error::EngineError::Capability) + /// when the lookup fails. + async fn people(&self, query: Option<&str>) -> Result; + + /// Persists `value` under `key` within `scope`. + /// + /// Callers (the `memory` node, via the validator) must never pass + /// `scope: "user"` here — see the trait-level docs. `value` is caller + /// (workflow-author / upstream node) controlled; hosts should apply the + /// same taint/provenance handling they use for any externally-influenced + /// write. + /// + /// # Errors + /// Returns an [`EngineError::Capability`](crate::error::EngineError::Capability) + /// when the scope is unknown to the host or the write fails. + async fn remember(&self, scope: &str, key: &str, value: Value) -> Result<()>; + + /// Deletes the memory stored under `key` within `scope`. + /// + /// Same `scope: "user"` restriction as [`remember`](Self::remember). + /// Deleting an already-absent `key` is not an error. + /// + /// # Errors + /// Returns an [`EngineError::Capability`](crate::error::EngineError::Capability) + /// when the scope is unknown to the host or the delete fails. + async fn forget(&self, scope: &str, key: &str) -> Result<()>; +} + /// Invokes a named integration tool (e.g. a curated Composio action). #[async_trait] pub trait ToolInvoker: Send + Sync { @@ -126,9 +214,11 @@ pub trait StateStore: Send + Sync { /// The bundle of capabilities handed to the engine for a run. /// /// Construct one per run from the host's concrete implementations. It carries -/// all five host-injected capabilities: the [`LlmProvider`], [`ToolInvoker`], -/// [`HttpClient`], [`CodeRunner`], and [`StateStore`]. Nodes reach each one -/// through `ctx.caps` during execution. +/// every host-injected capability: the always-present [`LlmProvider`], +/// [`ToolInvoker`], [`HttpClient`], [`CodeRunner`], [`StateStore`], and +/// [`WorkflowResolver`], plus the optional [`AgentRunner`] and +/// [`MemoryProvider`]. Nodes reach each one through `ctx.caps` during +/// execution. #[derive(Clone)] pub struct Capabilities { /// LLM provider for agent / output-parser nodes. @@ -156,6 +246,12 @@ pub struct Capabilities { /// [`LlmProvider`] completion. `None` on hosts without an agent registry, in /// which case `agent` nodes always use [`LlmProvider`]. pub agent: Option>, + /// Optional host-managed memory access for `memory` nodes. `None` on hosts + /// without a memory store, in which case a `memory` node fails at run time + /// with a capability error (there is no meaningful no-op fallback for a + /// read/write memory call). See [`MemoryProvider`] for the `scope` + /// contract and the `remember`/`forget` write restriction. + pub memory: Option>, } #[cfg(test)] @@ -190,6 +286,10 @@ mod tests { assert!(Arc::ptr_eq(&caps.code, &clone.code)); assert!(Arc::ptr_eq(&caps.state, &clone.state)); assert!(Arc::ptr_eq(&caps.resolver, &clone.resolver)); + assert!(Arc::ptr_eq( + caps.memory.as_ref().expect("mock wires memory"), + clone.memory.as_ref().expect("mock wires memory") + )); } #[tokio::test] diff --git a/src/catalog.rs b/src/catalog.rs index 39eb459..f90f85e 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -22,7 +22,7 @@ use serde_json::{Value, json}; /// The node kinds, in the canonical order used wherever the DSL is enumerated /// (matches [`NodeKind`](crate::model::NodeKind)'s serde discriminators). -pub const NODE_KINDS: [&str; 12] = [ +pub const NODE_KINDS: [&str; 13] = [ "trigger", "agent", "tool_call", @@ -35,6 +35,7 @@ pub const NODE_KINDS: [&str; 12] = [ "transform", "output_parser", "sub_workflow", + "memory", ]; /// One config field a node of a given kind reads at run time. @@ -162,7 +163,7 @@ pub fn all_contracts() -> Vec { .collect() } -/// The contract for one node kind, or `None` if `kind` is not one of the 12. +/// The contract for one node kind, or `None` if `kind` is not one of the 13. pub fn contract_for(kind: &str) -> Option { let c = match kind { "trigger" => NodeKindContract { @@ -467,6 +468,81 @@ pub fn contract_for(kind: &str) -> Option { .to_string(), ], }, + "memory" => NodeKindContract { + kind: "memory".to_string(), + summary: "Reads or writes host-managed memory via the MemoryProvider capability." + .to_string(), + description: "config.operation selects recall / search (query lookups), flavour \ + (a named ask/persona/style profile by slug), people (a people lookup), or \ + remember / forget (writes). What memory actually contains, how recall ranks \ + results, and what a flavour/people entry looks like are host concerns — the \ + engine only shapes the call and envelopes the response." + .to_string(), + config_fields: vec![ + ConfigField::required( + "operation", + "enum", + "Which memory action this node performs.", + ) + .with_enum(&["recall", "search", "flavour", "people", "remember", "forget"]), + ConfigField::optional( + "scope", + "enum", + "Required for recall / remember / forget. Host-defined: \"user\" (the \ + caller's durable, cross-flow memory — READ-ONLY from a workflow), \"flow\" \ + (this flow's own memory — the only scope remember/forget may target), or \ + \"flows\" (cross-flow read access — read-only).", + ) + .with_enum(&["user", "flow", "flows"]), + ConfigField::optional( + "query", + "\"=expr\"", + "Required for recall / search (optional for people). The lookup query.", + ), + ConfigField::optional( + "flavour", + "string", + "Required for the flavour operation: the ask/persona/style slug to look up, \ + e.g. \"email-tone\".", + ), + ConfigField::optional( + "key", + "\"=expr\"", + "Required for remember / forget: the memory key to write or delete.", + ), + ConfigField::optional( + "value", + "\"=expr\"", + "Required for remember: the value to persist under key.", + ), + ConfigField::optional( + "limit", + "number", + "Optional cap on the number of results for recall / search.", + ), + ConfigField::optional( + "min_score", + "number", + "Optional relevance-score floor for recall / search results.", + ), + ], + ports: PortSpec::linear(), + example: json!({ + "id": "check_seen", "kind": "memory", "name": "Already published?", + "config": { "operation": "recall", "scope": "flow", "query": "=item.title" } + }), + notes: vec![ + "HARD SECURITY RULE: a remember/forget node with scope \"user\" is a hard reject \ + at validate time — writes may only target scope \"flow\". This is enforced \ + structurally so an author (or an LLM authoring a graph) cannot plant or erase \ + durable facts about the user via workflow content." + .to_string(), + "Default execution is per_item (like tool_call), so a split_out fan-out runs one \ + memory call per item — set execution: \"once\" to run a single call against the \ + first item instead." + .to_string(), + ], + }, _ => return None, }; Some(c) @@ -499,7 +575,7 @@ mod tests { } } } - assert_eq!(all_contracts().len(), 12); + assert_eq!(all_contracts().len(), 13); } #[test] @@ -525,6 +601,53 @@ mod tests { assert!(contract_for("").is_none()); } + #[test] + fn node_kinds_has_13_entries_including_memory() { + assert_eq!(NODE_KINDS.len(), 13); + assert!(NODE_KINDS.contains(&"memory")); + // "memory" is the 13th (last) entry — added at the end per the + // sequenced-last design rationale. + assert_eq!(NODE_KINDS[12], "memory"); + } + + #[test] + fn memory_contract_documents_the_six_operations_and_scope_enum() { + let c = contract_for("memory").expect("memory contract exists"); + let operation_field = c + .config_fields + .iter() + .find(|f| f.name == "operation") + .expect("memory contract declares `operation`"); + assert!(operation_field.required); + assert_eq!( + operation_field.enum_values, + Some( + vec!["recall", "search", "flavour", "people", "remember", "forget"] + .into_iter() + .map(str::to_string) + .collect::>() + ) + ); + let scope_field = c + .config_fields + .iter() + .find(|f| f.name == "scope") + .expect("memory contract declares `scope`"); + assert_eq!( + scope_field.enum_values, + Some( + vec!["user", "flow", "flows"] + .into_iter() + .map(str::to_string) + .collect::>() + ) + ); + assert!( + c.notes.iter().any(|n| n.contains("HARD SECURITY RULE")), + "memory contract must document the user-scope write rejection" + ); + } + #[test] fn with_note_appends_a_host_caveat() { let c = contract_for("tool_call").unwrap().with_note("host says hi"); diff --git a/src/model/node_kind.rs b/src/model/node_kind.rs index aca26ce..138e943 100644 --- a/src/model/node_kind.rs +++ b/src/model/node_kind.rs @@ -44,6 +44,12 @@ pub enum NodeKind { OutputParser, /// Runs another workflow as a nested sub-graph. SubWorkflow, + /// Reads or writes host-managed memory via the injected + /// [`MemoryProvider`](crate::caps::MemoryProvider) capability: `recall` / + /// `search` / `flavour` / `people` for reads, `remember` / `forget` for + /// writes. Additive kind — see [`crate::validate`] for the hard invariant + /// that a `remember`/`forget` operation may never target `scope: "user"`. + Memory, } /// How a [`NodeKind::Trigger`] node is fired. @@ -99,6 +105,7 @@ mod tests { assert_wire(&NodeKind::Transform, "transform"); assert_wire(&NodeKind::OutputParser, "output_parser"); assert_wire(&NodeKind::SubWorkflow, "sub_workflow"); + assert_wire(&NodeKind::Memory, "memory"); } #[test] diff --git a/src/nodes/integration/memory.rs b/src/nodes/integration/memory.rs new file mode 100644 index 0000000..dd923b6 --- /dev/null +++ b/src/nodes/integration/memory.rs @@ -0,0 +1,555 @@ +//! The `memory` node: recall/search/flavour/people/remember/forget against the +//! host's injected [`crate::caps::MemoryProvider`]. +//! +//! A **delivery surface**, not a capability of its own — it exposes whatever +//! durable memory store the host already maintains to the declarative graph, so +//! a node kind that "cannot reason" (a `condition` branch, a `transform`) can +//! still gate or bind on recalled memory without an agent turn in the loop. See +//! the crate's design notes for the full rationale. +//! +//! Config (`config.operation` selects the shape read below): +//! +//! | Field | Used by | Required | +//! |---|---|---| +//! | `operation` | all | always: `recall`\|`search`\|`flavour`\|`people`\|`remember`\|`forget` | +//! | `scope` | recall, remember, forget | yes (host-defined: `"user"`\|`"flow"`\|`"flows"`) | +//! | `query` | recall, search | yes (`=`-bindable) | +//! | `flavour` | flavour | yes (a slug string) | +//! | `key` | remember, forget | yes (`=`-bindable) | +//! | `value` | remember | yes (`=`-bindable) | +//! | `limit`, `min_score` | recall, search | no | +//! +//! [`crate::validate`] enforces the hard security invariant — `remember`/ +//! `forget` may never carry `scope: "user"` — and the required-field checks +//! above, structurally, before a run starts. This executor still defends +//! against a missing/malformed config (e.g. when driven directly in a test, +//! bypassing the validator) with the same [`crate::error::EngineError::Capability`] +//! pattern every other integration node uses for a bad config. + +use async_trait::async_trait; +use serde_json::Value; + +use crate::data::Item; +use crate::error::{EngineError, Result}; +use crate::nodes::integration::envelope; +use crate::nodes::{ExecutionMode, NodeContext, NodeExecutor, NodeOutput, execution_mode}; + +/// Stable `tracing` grep prefix for every log line this node emits. +const LOG_PREFIX: &str = "[memory-node]"; + +/// Reads/writes host-managed memory via +/// [`MemoryProvider`](crate::caps::MemoryProvider). +/// +/// **Execution** (`config.execution`, default `per_item`): matches `tool_call` / +/// `http_request` — in `per_item` mode the node maps over its input, calling +/// the provider once per item with config re-resolved against that item (the +/// `split_out` → `memory[recall]` dedupe-check pattern depends on this: each +/// candidate's own `=item.title` must reach its own recall call). `once` +/// invokes a single time against the first item. With no input, either mode +/// invokes once. +/// +/// Output is wrapped in the stable `{ json, text, raw }` +/// [envelope](crate::nodes::integration::envelope), matching every other +/// capability node. +#[derive(Debug, Default, Clone)] +pub struct MemoryNode; + +/// Resolves `operation`/`scope`/`query`/etc. from an already-resolved `cfg` and +/// calls the matching [`MemoryProvider`](crate::caps::MemoryProvider) method, +/// returning the provider's (unenveloped) result. +async fn call_provider(ctx: &NodeContext<'_>, cfg: &Value) -> Result { + let provider = ctx.caps.memory.as_ref().ok_or_else(|| { + EngineError::Capability( + "memory node: host has not wired a MemoryProvider capability".to_string(), + ) + })?; + + let operation = cfg.get("operation").and_then(Value::as_str).ok_or_else(|| { + EngineError::Capability("memory node: missing `operation` in config".to_string()) + })?; + let scope = cfg.get("scope").and_then(Value::as_str).unwrap_or(""); + + tracing::debug!( + node = %ctx.node.id, + operation, + scope, + "{LOG_PREFIX} executing" + ); + + let result = match operation { + "recall" | "search" => { + let query = cfg.get("query").and_then(Value::as_str).ok_or_else(|| { + EngineError::Capability(format!( + "memory node: `{operation}` operation requires `query`" + )) + })?; + tracing::debug!( + node = %ctx.node.id, + query, + "{LOG_PREFIX} resolved query, calling provider.recall" + ); + let mut opts = serde_json::Map::new(); + opts.insert("operation".to_string(), Value::String(operation.to_string())); + if let Some(limit) = cfg.get("limit") { + opts.insert("limit".to_string(), limit.clone()); + } + if let Some(min_score) = cfg.get("min_score") { + opts.insert("min_score".to_string(), min_score.clone()); + } + provider.recall(scope, query, Value::Object(opts)).await? + } + "flavour" => { + let slug = cfg.get("flavour").and_then(Value::as_str).ok_or_else(|| { + EngineError::Capability( + "memory node: `flavour` operation requires `flavour` (slug)".to_string(), + ) + })?; + tracing::debug!( + node = %ctx.node.id, + slug, + "{LOG_PREFIX} calling provider.flavour" + ); + provider.flavour(slug).await? + } + "people" => { + let query = cfg.get("query").and_then(Value::as_str); + tracing::debug!( + node = %ctx.node.id, + query = query.unwrap_or(""), + "{LOG_PREFIX} calling provider.people" + ); + provider.people(query).await? + } + "remember" => { + let key = cfg.get("key").and_then(Value::as_str).ok_or_else(|| { + EngineError::Capability("memory node: `remember` operation requires `key`".to_string()) + })?; + let value = cfg.get("value").cloned().ok_or_else(|| { + EngineError::Capability( + "memory node: `remember` operation requires `value`".to_string(), + ) + })?; + tracing::debug!( + node = %ctx.node.id, + key, + "{LOG_PREFIX} calling provider.remember" + ); + provider.remember(scope, key, value).await?; + serde_json::json!({ "ok": true, "operation": "remember", "key": key }) + } + "forget" => { + let key = cfg.get("key").and_then(Value::as_str).ok_or_else(|| { + EngineError::Capability("memory node: `forget` operation requires `key`".to_string()) + })?; + tracing::debug!( + node = %ctx.node.id, + key, + "{LOG_PREFIX} calling provider.forget" + ); + provider.forget(scope, key).await?; + serde_json::json!({ "ok": true, "operation": "forget", "key": key }) + } + other => { + return Err(EngineError::Capability(format!( + "memory node: unknown operation {other:?}" + ))); + } + }; + + tracing::debug!( + node = %ctx.node.id, + operation, + result_size = result_size(&result), + "{LOG_PREFIX} provider call returned" + ); + Ok(result) +} + +/// A coarse "how big is this" hint for the debug log — the length of an array +/// result (e.g. `recall`'s `results`), the object's field count, or `1` for a +/// scalar/`null`. Not part of the node's output contract, purely diagnostic. +fn result_size(value: &Value) -> usize { + match value { + Value::Array(items) => items.len(), + Value::Object(map) => { + for field in map.values() { + if let Some(arr) = field.as_array() { + return arr.len(); + } + } + map.len() + } + _ => 1, + } +} + +#[async_trait] +impl NodeExecutor for MemoryNode { + async fn execute(&self, ctx: NodeContext<'_>) -> Result { + let per_item = execution_mode(&ctx.node.config, ExecutionMode::PerItem) + == ExecutionMode::PerItem + && !ctx.input.is_empty(); + + tracing::debug!( + node = %ctx.node.id, + per_item, + input_len = ctx.input.len(), + "{LOG_PREFIX} entering execute" + ); + + if per_item { + let mut items = Vec::with_capacity(ctx.input.len()); + let mut diagnostics = Vec::new(); + for (index, input_item) in ctx.input.iter().enumerate() { + let (cfg, diags) = + crate::nodes::resolve_config_traced_for_item(&ctx, input_item.json.clone()); + let result = call_provider(&ctx, &cfg).await?; + items.push(Item::new(envelope::wrap(result)).paired_with(index)); + diagnostics.extend(diags); + } + tracing::debug!( + node = %ctx.node.id, + emitted = items.len(), + "{LOG_PREFIX} exiting execute (per_item)" + ); + Ok(NodeOutput::main(items).with_diagnostics(diagnostics)) + } else { + let (cfg, diagnostics) = crate::nodes::resolve_config_traced(&ctx); + let result = call_provider(&ctx, &cfg).await?; + tracing::debug!( + node = %ctx.node.id, + "{LOG_PREFIX} exiting execute (once)" + ); + Ok(NodeOutput::main(vec![Item::new(envelope::wrap(result))]) + .with_diagnostics(diagnostics)) + } + } +} + +#[cfg(test)] +mod tests { + use crate::caps::mock::{mock_capabilities, mock_capabilities_with_memory}; + use crate::compiler::compile; + use crate::engine::run; + use crate::model::{Edge, Node, NodeKind, WorkflowGraph}; + use serde_json::{Value, json}; + + fn wf(config: Value) -> WorkflowGraph { + WorkflowGraph { + nodes: vec![ + Node { + id: "t".into(), + kind: NodeKind::Trigger, + type_version: 1, + name: "t".into(), + config: Value::Null, + ports: vec![], + position: None, + }, + Node { + id: "n".into(), + kind: NodeKind::Memory, + type_version: 1, + name: "n".into(), + config, + ports: vec![], + position: None, + }, + ], + edges: vec![Edge { + from_node: "t".into(), + from_port: "main".into(), + to_node: "n".into(), + to_port: "main".into(), + }], + ..Default::default() + } + } + + #[tokio::test] + async fn recall_executes_against_mock_and_emits_results_on_output() { + let graph = wf(json!({ "operation": "recall", "scope": "flow", "query": "budget" })); + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, Value::Null, &mock_capabilities()) + .await + .expect("run"); + let results = &out.output["nodes"]["n"]["items"][0]["json"]["json"]["results"]; + assert!( + results.as_array().is_some_and(|r| !r.is_empty()), + "expected shaped recall results, got {results:?}" + ); + } + + #[tokio::test] + async fn flavour_operation_shapes_output() { + let graph = wf(json!({ "operation": "flavour", "flavour": "email-tone" })); + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, Value::Null, &mock_capabilities()) + .await + .expect("run"); + assert_eq!( + out.output["nodes"]["n"]["items"][0]["json"]["json"]["slug"], + "email-tone" + ); + assert!( + out.output["nodes"]["n"]["items"][0]["json"]["json"]["traits"].is_object() + ); + } + + #[tokio::test] + async fn people_operation_shapes_output() { + let graph = wf(json!({ "operation": "people", "query": "cyrus" })); + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, Value::Null, &mock_capabilities()) + .await + .expect("run"); + let people = &out.output["nodes"]["n"]["items"][0]["json"]["json"]["people"]; + assert!(people.as_array().is_some_and(|p| !p.is_empty())); + } + + #[tokio::test] + async fn remember_flow_scope_writes_via_provider_and_passes_items_through() { + let graph = wf(json!({ + "operation": "remember", "scope": "flow", "key": "k1", "value": { "v": 1 } + })); + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, Value::Null, &mock_capabilities()) + .await + .expect("run"); + assert_eq!( + out.output["nodes"]["n"]["items"][0]["json"]["json"]["ok"], + true + ); + assert_eq!( + out.output["nodes"]["n"]["items"][0]["json"]["json"]["key"], + "k1" + ); + } + + #[tokio::test] + async fn forget_flow_scope_executes() { + let graph = wf(json!({ "operation": "forget", "scope": "flow", "key": "k1" })); + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, Value::Null, &mock_capabilities()) + .await + .expect("run"); + assert_eq!( + out.output["nodes"]["n"]["items"][0]["json"]["json"]["operation"], + "forget" + ); + } + + use super::MemoryNode; + use crate::data::Item; + use crate::error::EngineError; + use crate::nodes::{NodeContext, NodeExecutor}; + + fn memory_node(config: Value) -> Node { + Node { + id: "n".into(), + kind: NodeKind::Memory, + type_version: 1, + name: "n".into(), + config, + ports: vec![], + position: None, + } + } + + #[tokio::test] + async fn resolves_query_expression_per_item() { + // `query="=item.id"` must resolve per-item before the provider call — + // the `split_out` → `memory[recall]` dedupe pattern depends on this. + let node = memory_node(json!({ + "operation": "recall", "scope": "flow", "query": "=item.id" + })); + let input = vec![ + Item::new(json!({ "id": "a" })), + Item::new(json!({ "id": "b" })), + ]; + let caps = mock_capabilities(); + let run_meta = Value::Null; + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + }; + let out = MemoryNode.execute(ctx).await.expect("execute"); + assert_eq!(out.items.len(), 2, "per_item default maps over input"); + assert_eq!(out.items[0].json["json"]["query"], "a"); + assert_eq!(out.items[1].json["json"]["query"], "b"); + assert_eq!(out.items[1].paired_item, Some(1)); + } + + #[tokio::test] + async fn search_operation_passes_operation_through_opts() { + // `search` reuses `recall` but tags `opts.operation` so a host that + // distinguishes semantic recall from full-text search can branch on it. + struct OptsEchoingMemory; + #[async_trait::async_trait] + impl crate::caps::MemoryProvider for OptsEchoingMemory { + async fn recall( + &self, + _scope: &str, + _query: &str, + opts: Value, + ) -> crate::error::Result { + Ok(json!({ "opts": opts })) + } + async fn flavour(&self, _slug: &str) -> crate::error::Result { + Ok(Value::Null) + } + async fn people(&self, _query: Option<&str>) -> crate::error::Result { + Ok(Value::Null) + } + async fn remember( + &self, + _scope: &str, + _key: &str, + _value: Value, + ) -> crate::error::Result<()> { + Ok(()) + } + async fn forget(&self, _scope: &str, _key: &str) -> crate::error::Result<()> { + Ok(()) + } + } + + let node = memory_node(json!({ + "operation": "search", "scope": "flow", "query": "x", "limit": 5 + })); + let input: Vec = vec![]; + let caps = mock_capabilities_with_memory(OptsEchoingMemory); + let run_meta = Value::Null; + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + }; + let out = MemoryNode.execute(ctx).await.expect("execute"); + assert_eq!(out.items[0].json["json"]["opts"]["operation"], "search"); + assert_eq!(out.items[0].json["json"]["opts"]["limit"], 5); + } + + #[tokio::test] + async fn missing_query_on_recall_is_a_capability_error() { + let node = memory_node(json!({ "operation": "recall", "scope": "flow" })); + let input = vec![]; + let caps = mock_capabilities(); + let run_meta = Value::Null; + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + }; + let err = MemoryNode + .execute(ctx) + .await + .expect_err("missing query must error"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("query")), + "expected a capability error mentioning `query`, got: {err:?}" + ); + } + + #[tokio::test] + async fn missing_key_on_remember_is_a_capability_error() { + let node = memory_node(json!({ + "operation": "remember", "scope": "flow", "value": 1 + })); + let input = vec![]; + let caps = mock_capabilities(); + let run_meta = Value::Null; + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + }; + let err = MemoryNode + .execute(ctx) + .await + .expect_err("missing key must error"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("key")), + "expected a capability error mentioning `key`, got: {err:?}" + ); + } + + #[tokio::test] + async fn missing_operation_is_a_capability_error() { + let node = memory_node(json!({ "scope": "flow" })); + let input = vec![]; + let caps = mock_capabilities(); + let run_meta = Value::Null; + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + }; + let err = MemoryNode + .execute(ctx) + .await + .expect_err("missing operation must error"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("operation")), + "expected a capability error mentioning `operation`, got: {err:?}" + ); + } + + #[tokio::test] + async fn no_memory_provider_wired_is_a_capability_error() { + let node = memory_node(json!({ "operation": "people" })); + let input = vec![]; + let mut caps = mock_capabilities(); + caps.memory = None; + let run_meta = Value::Null; + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + }; + let err = MemoryNode + .execute(ctx) + .await + .expect_err("no MemoryProvider must error"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("MemoryProvider")), + "expected a capability error mentioning MemoryProvider, got: {err:?}" + ); + } + + #[tokio::test] + async fn execution_once_collapses_the_batch_to_a_single_call() { + let node = memory_node(json!({ + "operation": "recall", "scope": "flow", "query": "=item.id", "execution": "once" + })); + let input = vec![ + Item::new(json!({ "id": "a" })), + Item::new(json!({ "id": "b" })), + ]; + let caps = mock_capabilities(); + let run_meta = Value::Null; + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + }; + let out = MemoryNode.execute(ctx).await.expect("execute"); + assert_eq!(out.items.len(), 1, "once mode emits a single item"); + assert_eq!(out.items[0].json["json"]["query"], "a"); + } +} diff --git a/src/nodes/integration/mod.rs b/src/nodes/integration/mod.rs index 38ce963..a2a2e28 100644 --- a/src/nodes/integration/mod.rs +++ b/src/nodes/integration/mod.rs @@ -1,6 +1,6 @@ //! Capability-backed node executors: `agent`, `tool_call`, `http_request`, -//! `code`, `output_parser`, and `sub_workflow`. These reach the outside world -//! through the host capabilities in [`crate::caps`]. +//! `code`, `output_parser`, `sub_workflow`, and `memory`. These reach the +//! outside world through the host capabilities in [`crate::caps`]. //! //! One module per node kind so parallel work can edit them without conflicts. @@ -8,6 +8,7 @@ pub mod agent; pub mod code; pub(crate) mod envelope; pub mod http_request; +pub mod memory; pub mod output_parser; pub(crate) mod schema; pub mod sub_workflow; @@ -16,6 +17,7 @@ pub mod tool_call; pub use agent::AgentNode; pub use code::CodeNode; pub use http_request::HttpRequestNode; +pub use memory::MemoryNode; pub use output_parser::OutputParserNode; pub use sub_workflow::SubWorkflowNode; pub use tool_call::ToolCallNode; diff --git a/src/nodes/mod.rs b/src/nodes/mod.rs index e086d25..1aa58f5 100644 --- a/src/nodes/mod.rs +++ b/src/nodes/mod.rs @@ -269,6 +269,7 @@ pub(crate) fn executor_for(kind: &NodeKind) -> Box { NodeKind::Code => Box::new(integration::CodeNode), NodeKind::OutputParser => Box::new(integration::OutputParserNode), NodeKind::SubWorkflow => Box::new(integration::SubWorkflowNode), + NodeKind::Memory => Box::new(integration::MemoryNode), NodeKind::Condition => Box::new(control_flow::ConditionNode), NodeKind::Switch => Box::new(control_flow::SwitchNode), NodeKind::Merge => Box::new(control_flow::MergeNode), @@ -288,8 +289,8 @@ mod tests { /// Every [`NodeKind`] variant, so the coverage below stays exhaustive. fn all_kinds() -> Vec { use NodeKind::{ - Agent, Code, Condition, HttpRequest, Merge, OutputParser, SplitOut, SubWorkflow, - Switch, ToolCall, Transform, Trigger, + Agent, Code, Condition, HttpRequest, Memory, Merge, OutputParser, SplitOut, + SubWorkflow, Switch, ToolCall, Transform, Trigger, }; vec![ Trigger, @@ -304,6 +305,7 @@ mod tests { Transform, OutputParser, SubWorkflow, + Memory, ] } @@ -314,6 +316,10 @@ mod tests { NodeKind::SubWorkflow => json!({ "workflow": { "nodes": [{ "id": "ct", "kind": "trigger", "name": "ct" }], "edges": [] } }), + // `people` needs no `scope`/`query`, so it runs against the + // default mock capabilities (which wire a `MemoryProvider`) with + // the minimal config every other kind gets via `Value::Null`. + NodeKind::Memory => json!({ "operation": "people" }), _ => Value::Null, } } diff --git a/src/validate.rs b/src/validate.rs index 463ff76..48d0a05 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -137,6 +137,128 @@ pub fn validate_all(graph: &WorkflowGraph) -> Vec { } } + // `memory` node config checks, including THE hard security invariant: a + // `remember`/`forget` operation may never target `scope: "user"` — the + // caller's durable, cross-flow memory. Rejecting this structurally, at the + // door, means a workflow (or an LLM authoring one) can never plant or erase + // durable facts about the user by way of a `remember`/`forget` node; the + // only scope those two operations may write through is `"flow"`. + for node in &graph.nodes { + if node.kind != NodeKind::Memory { + continue; + } + + let operation = node.config.get("operation").and_then(Value::as_str); + let Some(operation) = operation else { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: "memory node requires `operation` (recall|search|flavour|people|\ + remember|forget)" + .to_string(), + }); + continue; + }; + if !matches!( + operation, + "recall" | "search" | "flavour" | "people" | "remember" | "forget" + ) { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!( + "memory node has unknown operation {operation:?} (expected one of \ + recall|search|flavour|people|remember|forget)" + ), + }); + continue; + } + + let scope = node.config.get("scope").and_then(Value::as_str); + + // THE hard invariant (see the block comment above): reject before any + // other config check, so it can never be masked by a different error. + if matches!(operation, "remember" | "forget") && scope == Some("user") { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!( + "memory node operation {operation:?} may not target scope \"user\" — \ + remember/forget may only write scope \"flow\"; the user's cross-flow \ + memory is read-only from a workflow" + ), + }); + } + + if let Some(scope) = scope { + if !matches!(scope, "user" | "flow" | "flows") { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!( + "memory node has unknown scope {scope:?} (expected \ + user|flow|flows)" + ), + }); + } + } + + // `scope` is required for recall/remember/forget (not search/flavour/ + // people — see the catalog contract for the exact per-operation table). + if matches!(operation, "recall" | "remember" | "forget") && scope.is_none() { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!("memory node operation {operation:?} requires `scope`"), + }); + } + + if matches!(operation, "recall" | "search") { + let has_query = node + .config + .get("query") + .and_then(Value::as_str) + .is_some_and(|s| !s.is_empty()); + if !has_query { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!("memory node operation {operation:?} requires `query`"), + }); + } + } + + if operation == "flavour" { + let has_flavour = node + .config + .get("flavour") + .and_then(Value::as_str) + .is_some_and(|s| !s.is_empty()); + if !has_flavour { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: "memory node operation \"flavour\" requires `flavour` (slug)" + .to_string(), + }); + } + } + + if matches!(operation, "remember" | "forget") { + let has_key = node + .config + .get("key") + .and_then(Value::as_str) + .is_some_and(|s| !s.is_empty()); + if !has_key { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!("memory node operation {operation:?} requires `key`"), + }); + } + } + + if operation == "remember" && node.config.get("value").is_none() { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: "memory node operation \"remember\" requires `value`".to_string(), + }); + } + } + // A `condition` node's outgoing edges must emit on `from_port` "true" or // "false" — routing is keyed EXCLUSIVELY on `from_port` (see // `engine::outgoing_by_port` / `handler_routing`), so any other value @@ -363,6 +485,227 @@ mod tests { )); } + fn memory_node(id: &str, config: serde_json::Value) -> Node { + let mut n = node(id, NodeKind::Memory); + n.config = config; + n + } + + fn graph_with_memory_node(config: serde_json::Value) -> WorkflowGraph { + WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), memory_node("mem", config)], + edges: vec![Edge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "mem".to_string(), + to_port: "main".to_string(), + }], + ..Default::default() + } + } + + // --- the hard invariant: remember/forget may never target scope "user" --- + + #[test] + fn memory_rejects_remember_user_scope() { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "remember", "scope": "user", "key": "k", "value": 1 + })); + let err = validate(&graph).expect_err("remember·user must be rejected"); + match err { + ValidationError::InvalidNodeConfig { node, reason } => { + assert_eq!(node, "mem"); + assert!(reason.contains("\"user\""), "reason: {reason}"); + assert!(reason.contains("remember"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig, got {other:?}"), + } + } + + #[test] + fn memory_rejects_forget_user_scope() { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "forget", "scope": "user", "key": "k" + })); + let err = validate(&graph).expect_err("forget·user must be rejected"); + match err { + ValidationError::InvalidNodeConfig { node, reason } => { + assert_eq!(node, "mem"); + assert!(reason.contains("\"user\""), "reason: {reason}"); + assert!(reason.contains("forget"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig, got {other:?}"), + } + } + + #[test] + fn memory_accepts_remember_flow_scope() { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "remember", "scope": "flow", "key": "k", "value": { "v": 1 } + })); + assert_eq!(validate(&graph), Ok(())); + } + + #[test] + fn memory_accepts_forget_flow_scope() { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "forget", "scope": "flow", "key": "k" + })); + assert_eq!(validate(&graph), Ok(())); + } + + #[test] + fn memory_rejects_unknown_scope_value() { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "recall", "scope": "everyone", "query": "x" + })); + assert!(matches!( + validate(&graph), + Err(ValidationError::InvalidNodeConfig { .. }) + )); + } + + // --- required-field checks per operation --- + + #[test] + fn memory_recall_accepts_user_and_flows_scope() { + // Only remember/forget are scope-restricted; reads may target any + // declared scope, including the read-only ones. + for scope in ["user", "flow", "flows"] { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "recall", "scope": scope, "query": "x" + })); + assert_eq!(validate(&graph), Ok(()), "scope {scope} should be valid for recall"); + } + } + + #[test] + fn memory_requires_operation() { + let graph = graph_with_memory_node(serde_json::json!({ "scope": "flow" })); + match validate(&graph) { + Err(ValidationError::InvalidNodeConfig { node, reason }) => { + assert_eq!(node, "mem"); + assert!(reason.contains("operation"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig, got {other:?}"), + } + } + + #[test] + fn memory_rejects_unknown_operation() { + let graph = graph_with_memory_node(serde_json::json!({ "operation": "levitate" })); + assert!(matches!( + validate(&graph), + Err(ValidationError::InvalidNodeConfig { .. }) + )); + } + + #[test] + fn memory_recall_requires_scope() { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "recall", "query": "x" + })); + match validate(&graph) { + Err(ValidationError::InvalidNodeConfig { node, reason }) => { + assert_eq!(node, "mem"); + assert!(reason.contains("scope"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig, got {other:?}"), + } + } + + #[test] + fn memory_recall_requires_query() { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "recall", "scope": "flow" + })); + match validate(&graph) { + Err(ValidationError::InvalidNodeConfig { node, reason }) => { + assert_eq!(node, "mem"); + assert!(reason.contains("query"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig, got {other:?}"), + } + } + + #[test] + fn memory_search_requires_query_but_not_scope() { + let missing_query = graph_with_memory_node(serde_json::json!({ "operation": "search" })); + assert!(matches!( + validate(&missing_query), + Err(ValidationError::InvalidNodeConfig { .. }) + )); + + let no_scope_ok = graph_with_memory_node(serde_json::json!({ + "operation": "search", "query": "x" + })); + assert_eq!(validate(&no_scope_ok), Ok(())); + } + + #[test] + fn memory_flavour_requires_flavour_slug() { + let graph = graph_with_memory_node(serde_json::json!({ "operation": "flavour" })); + match validate(&graph) { + Err(ValidationError::InvalidNodeConfig { node, reason }) => { + assert_eq!(node, "mem"); + assert!(reason.contains("flavour"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig, got {other:?}"), + } + let ok = graph_with_memory_node(serde_json::json!({ + "operation": "flavour", "flavour": "email-tone" + })); + assert_eq!(validate(&ok), Ok(())); + } + + #[test] + fn memory_people_requires_nothing() { + // `people` has no required `scope`/`query` — an empty config is valid. + let graph = graph_with_memory_node(serde_json::json!({ "operation": "people" })); + assert_eq!(validate(&graph), Ok(())); + } + + #[test] + fn memory_remember_requires_key_and_value() { + let missing_both = graph_with_memory_node(serde_json::json!({ + "operation": "remember", "scope": "flow" + })); + match validate(&missing_both) { + Err(ValidationError::InvalidNodeConfig { node, reason }) => { + assert_eq!(node, "mem"); + assert!(reason.contains("key"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig (key), got {other:?}"), + } + + let missing_value = graph_with_memory_node(serde_json::json!({ + "operation": "remember", "scope": "flow", "key": "k" + })); + match validate(&missing_value) { + Err(ValidationError::InvalidNodeConfig { node, reason }) => { + assert_eq!(node, "mem"); + assert!(reason.contains("value"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig (value), got {other:?}"), + } + } + + #[test] + fn memory_forget_requires_key_but_not_value() { + let missing_key = graph_with_memory_node(serde_json::json!({ + "operation": "forget", "scope": "flow" + })); + assert!(matches!( + validate(&missing_key), + Err(ValidationError::InvalidNodeConfig { .. }) + )); + + let ok = graph_with_memory_node(serde_json::json!({ + "operation": "forget", "scope": "flow", "key": "k" + })); + assert_eq!(validate(&ok), Ok(())); + } + fn tool_node(id: &str, config: serde_json::Value) -> Node { let mut n = node(id, NodeKind::ToolCall); n.config = config; From af16eb058e338ca27d99b6bccea2a6728ad04316 Mon Sep 17 00:00:00 2001 From: cyrus Date: Tue, 28 Jul 2026 20:21:07 +0530 Subject: [PATCH 2/4] style: rustfmt the memory node files (stable rustfmt) --- src/caps/mock.rs | 15 ++++++++++++--- src/catalog.rs | 14 +++++++++----- src/nodes/integration/memory.rs | 26 +++++++++++++++++--------- src/validate.rs | 6 +++++- 4 files changed, 43 insertions(+), 18 deletions(-) diff --git a/src/caps/mock.rs b/src/caps/mock.rs index db30618..298bc7b 100644 --- a/src/caps/mock.rs +++ b/src/caps/mock.rs @@ -321,7 +321,10 @@ mod tests { assert_eq!(out["scope"], "flow"); assert_eq!(out["query"], "budget"); let results = out["results"].as_array().expect("results array"); - assert!(!results.is_empty(), "mock recall should return shaped results"); + assert!( + !results.is_empty(), + "mock recall should return shaped results" + ); assert!(results[0].get("text").is_some()); } @@ -339,7 +342,10 @@ mod tests { let out = memory.people(Some("cyrus")).await.unwrap(); assert_eq!(out["query"], "cyrus"); let people = out["people"].as_array().expect("people array"); - assert!(!people.is_empty(), "mock people should return a shaped list"); + assert!( + !people.is_empty(), + "mock people should return a shaped list" + ); let out_no_query = memory.people(None).await.unwrap(); assert!(out_no_query["query"].is_null()); @@ -348,7 +354,10 @@ mod tests { #[tokio::test] async fn mock_memory_remember_and_forget_are_ok() { let memory = MockMemory; - memory.remember("flow", "k", json!({ "v": 1 })).await.unwrap(); + memory + .remember("flow", "k", json!({ "v": 1 })) + .await + .unwrap(); memory.forget("flow", "k").await.unwrap(); } diff --git a/src/catalog.rs b/src/catalog.rs index f90f85e..7e7aa5d 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -484,7 +484,9 @@ pub fn contract_for(kind: &str) -> Option { "enum", "Which memory action this node performs.", ) - .with_enum(&["recall", "search", "flavour", "people", "remember", "forget"]), + .with_enum(&[ + "recall", "search", "flavour", "people", "remember", "forget", + ]), ConfigField::optional( "scope", "enum", @@ -622,10 +624,12 @@ mod tests { assert_eq!( operation_field.enum_values, Some( - vec!["recall", "search", "flavour", "people", "remember", "forget"] - .into_iter() - .map(str::to_string) - .collect::>() + vec![ + "recall", "search", "flavour", "people", "remember", "forget" + ] + .into_iter() + .map(str::to_string) + .collect::>() ) ); let scope_field = c diff --git a/src/nodes/integration/memory.rs b/src/nodes/integration/memory.rs index dd923b6..288857d 100644 --- a/src/nodes/integration/memory.rs +++ b/src/nodes/integration/memory.rs @@ -64,9 +64,12 @@ async fn call_provider(ctx: &NodeContext<'_>, cfg: &Value) -> Result { ) })?; - let operation = cfg.get("operation").and_then(Value::as_str).ok_or_else(|| { - EngineError::Capability("memory node: missing `operation` in config".to_string()) - })?; + let operation = cfg + .get("operation") + .and_then(Value::as_str) + .ok_or_else(|| { + EngineError::Capability("memory node: missing `operation` in config".to_string()) + })?; let scope = cfg.get("scope").and_then(Value::as_str).unwrap_or(""); tracing::debug!( @@ -89,7 +92,10 @@ async fn call_provider(ctx: &NodeContext<'_>, cfg: &Value) -> Result { "{LOG_PREFIX} resolved query, calling provider.recall" ); let mut opts = serde_json::Map::new(); - opts.insert("operation".to_string(), Value::String(operation.to_string())); + opts.insert( + "operation".to_string(), + Value::String(operation.to_string()), + ); if let Some(limit) = cfg.get("limit") { opts.insert("limit".to_string(), limit.clone()); } @@ -122,7 +128,9 @@ async fn call_provider(ctx: &NodeContext<'_>, cfg: &Value) -> Result { } "remember" => { let key = cfg.get("key").and_then(Value::as_str).ok_or_else(|| { - EngineError::Capability("memory node: `remember` operation requires `key`".to_string()) + EngineError::Capability( + "memory node: `remember` operation requires `key`".to_string(), + ) })?; let value = cfg.get("value").cloned().ok_or_else(|| { EngineError::Capability( @@ -139,7 +147,9 @@ async fn call_provider(ctx: &NodeContext<'_>, cfg: &Value) -> Result { } "forget" => { let key = cfg.get("key").and_then(Value::as_str).ok_or_else(|| { - EngineError::Capability("memory node: `forget` operation requires `key`".to_string()) + EngineError::Capability( + "memory node: `forget` operation requires `key`".to_string(), + ) })?; tracing::debug!( node = %ctx.node.id, @@ -291,9 +301,7 @@ mod tests { out.output["nodes"]["n"]["items"][0]["json"]["json"]["slug"], "email-tone" ); - assert!( - out.output["nodes"]["n"]["items"][0]["json"]["json"]["traits"].is_object() - ); + assert!(out.output["nodes"]["n"]["items"][0]["json"]["json"]["traits"].is_object()); } #[tokio::test] diff --git a/src/validate.rs b/src/validate.rs index 48d0a05..4d8f502 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -575,7 +575,11 @@ mod tests { let graph = graph_with_memory_node(serde_json::json!({ "operation": "recall", "scope": scope, "query": "x" })); - assert_eq!(validate(&graph), Ok(()), "scope {scope} should be valid for recall"); + assert_eq!( + validate(&graph), + Ok(()), + "scope {scope} should be valid for recall" + ); } } From 7b3c72e3f8a3977b13785e7edacbf48ebc36ac47 Mon Sep 17 00:00:00 2001 From: cyrus Date: Tue, 28 Jul 2026 20:27:28 +0530 Subject: [PATCH 3/4] fix: add memory field to standalone_capabilities after merging main (#15 companion) --- src/main.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/main.rs b/src/main.rs index dc17e21..1d27f6e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -320,6 +320,7 @@ fn standalone_capabilities() -> tinyflows::caps::Capabilities { state: Arc::new(MemoryState::default()), resolver: Arc::new(NoResolver), agent: None, + memory: None, } } From b1f62f8916659405b4c7f6be6780c94074aa8f08 Mon Sep 17 00:00:00 2001 From: cyrus Date: Tue, 28 Jul 2026 20:38:51 +0530 Subject: [PATCH 4/4] fix(memory): reject flow-write to read-only 'flows' scope + guard direct-drive writes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - validate: remember/forget now reject scope "flows" (read-only), not just "user" — closes the write-restriction bypass (Greptile P1). Documents that the literal-enum scope check is what makes the invariant unbypassable (CodeRabbit). - executor: remember/forget hard-error unless scope=="flow", so a direct drive (bypassing validate) can never silently write an empty/read-only scope (CodeRabbit). Adds validate + executor regression tests. --- src/nodes/integration/memory.rs | 59 +++++++++++++++++++++++++++++++++ src/validate.rs | 53 ++++++++++++++++++++++++++--- 2 files changed, 108 insertions(+), 4 deletions(-) diff --git a/src/nodes/integration/memory.rs b/src/nodes/integration/memory.rs index 288857d..06511f8 100644 --- a/src/nodes/integration/memory.rs +++ b/src/nodes/integration/memory.rs @@ -127,6 +127,16 @@ async fn call_provider(ctx: &NodeContext<'_>, cfg: &Value) -> Result { provider.people(query).await? } "remember" => { + // Defense-in-depth (the validator already rejects non-"flow" writes, + // but this executor may be driven directly): writes go ONLY to + // scope "flow". An absent scope (defaulted to "") or a read-only + // scope ("user"/"flows") is a hard error, never a silent write to + // the wrong place. + if scope != "flow" { + return Err(EngineError::Capability(format!( + "memory node: `remember` may only write scope \"flow\", got {scope:?}" + ))); + } let key = cfg.get("key").and_then(Value::as_str).ok_or_else(|| { EngineError::Capability( "memory node: `remember` operation requires `key`".to_string(), @@ -146,6 +156,12 @@ async fn call_provider(ctx: &NodeContext<'_>, cfg: &Value) -> Result { serde_json::json!({ "ok": true, "operation": "remember", "key": key }) } "forget" => { + // Same flow-only write invariant as `remember` (see above). + if scope != "flow" { + return Err(EngineError::Capability(format!( + "memory node: `forget` may only write scope \"flow\", got {scope:?}" + ))); + } let key = cfg.get("key").and_then(Value::as_str).ok_or_else(|| { EngineError::Capability( "memory node: `forget` operation requires `key`".to_string(), @@ -491,6 +507,49 @@ mod tests { ); } + #[tokio::test] + async fn write_to_non_flow_scope_is_a_capability_error_even_when_driven_directly() { + // The validator rejects non-"flow" writes, but the executor is the last + // line of defense when driven directly (bypassing validate). A + // remember/forget against "flows" (read-only) or an absent scope must + // hard-error, never silently write to the wrong place. + for (op, cfg) in [ + ( + "remember", + json!({ "operation": "remember", "scope": "flows", "key": "k", "value": 1 }), + ), + ( + "forget", + json!({ "operation": "forget", "scope": "flows", "key": "k" }), + ), + // absent scope defaults to "" — also not "flow" — and must error. + ( + "remember", + json!({ "operation": "remember", "key": "k", "value": 1 }), + ), + ] { + let node = memory_node(cfg); + let input = vec![]; + let caps = mock_capabilities(); + let run_meta = Value::Null; + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + }; + let err = MemoryNode + .execute(ctx) + .await + .expect_err("non-flow write must error"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("flow") && m.contains(op)), + "expected a capability error mentioning `flow` and `{op}`, got: {err:?}" + ); + } + } + #[tokio::test] async fn missing_operation_is_a_capability_error() { let node = memory_node(json!({ "scope": "flow" })); diff --git a/src/validate.rs b/src/validate.rs index 4d8f502..2c0dab2 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -176,13 +176,24 @@ pub fn validate_all(graph: &WorkflowGraph) -> Vec { // THE hard invariant (see the block comment above): reject before any // other config check, so it can never be masked by a different error. - if matches!(operation, "remember" | "forget") && scope == Some("user") { + // remember/forget may write ONLY scope "flow". BOTH read-only scopes are + // rejected here — "user" (the user's durable memory) and "flows" + // (cross-flow read). This gate is unbypassable precisely because `scope` + // is validated as a literal enum (below): an "=expr" binding resolves at + // runtime and is never one of user|flow|flows, so it fails the enum + // check and can never smuggle a write past this into + // provider.remember/forget. If a future change makes `scope` bindable, + // this invariant reopens — keep the enum check. + if matches!(operation, "remember" | "forget") + && matches!(scope, Some("user") | Some("flows")) + { + let bad = scope.unwrap_or_default(); errors.push(ValidationError::InvalidNodeConfig { node: node.id.clone(), reason: format!( - "memory node operation {operation:?} may not target scope \"user\" — \ - remember/forget may only write scope \"flow\"; the user's cross-flow \ - memory is read-only from a workflow" + "memory node operation {operation:?} may not target scope {bad:?} — \ + remember/forget may only write scope \"flow\"; scopes \"user\" and \ + \"flows\" are read-only from a workflow" ), }); } @@ -538,6 +549,40 @@ mod tests { } } + #[test] + fn memory_rejects_remember_flows_scope() { + // "flows" is a read-only cross-flow scope — a write to it must be + // rejected at validate time, not just backstopped by the host adapter. + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "remember", "scope": "flows", "key": "k", "value": 1 + })); + let err = validate(&graph).expect_err("remember·flows must be rejected"); + match err { + ValidationError::InvalidNodeConfig { node, reason } => { + assert_eq!(node, "mem"); + assert!(reason.contains("\"flows\""), "reason: {reason}"); + assert!(reason.contains("remember"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig, got {other:?}"), + } + } + + #[test] + fn memory_rejects_forget_flows_scope() { + let graph = graph_with_memory_node(serde_json::json!({ + "operation": "forget", "scope": "flows", "key": "k" + })); + let err = validate(&graph).expect_err("forget·flows must be rejected"); + match err { + ValidationError::InvalidNodeConfig { node, reason } => { + assert_eq!(node, "mem"); + assert!(reason.contains("\"flows\""), "reason: {reason}"); + assert!(reason.contains("forget"), "reason: {reason}"); + } + other => panic!("expected InvalidNodeConfig, got {other:?}"), + } + } + #[test] fn memory_accepts_remember_flow_scope() { let graph = graph_with_memory_node(serde_json::json!({