From 6c665f4f6f309992bac97bc8550acfc222b4a14f Mon Sep 17 00:00:00 2001 From: Cyrus Gray <144336577+graycyrus@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:27:39 +0530 Subject: [PATCH] =?UTF-8?q?feat(flows):=20host=20adapter=20for=20the=20`me?= =?UTF-8?q?mory`=20node=20=E2=80=94=20in-graph=20memory=20read=20+=20flow-?= =?UTF-8?q?scoped=20write=20(#5227)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 11 +- app/src-tauri/Cargo.lock | 9 + src/openhuman/about_app/catalog_data.rs | 17 + .../agents/workflow_builder/builder_prompt.rs | 55 +- .../flows/agents/workflow_builder/prompt.md | 108 +++- src/openhuman/flows/builder_tools.rs | 8 +- src/openhuman/flows/builder_tools_tests.rs | 5 +- src/openhuman/flows/memory_tools.rs | 93 +-- src/openhuman/flows/mod.rs | 7 +- src/openhuman/flows/node_contracts.rs | 31 +- src/openhuman/flows/tools.rs | 22 +- src/openhuman/memory/tools.rs | 7 +- src/openhuman/memory/tools/flavour.rs | 245 ++++---- src/openhuman/tinyflows/caps.rs | 29 +- src/openhuman/tinyflows/memory_adapter.rs | 541 ++++++++++++++++++ .../tinyflows/memory_adapter_tests.rs | 276 +++++++++ .../tinyflows/memory_node_e2e_tests.rs | 402 +++++++++++++ src/openhuman/tinyflows/mod.rs | 18 +- src/openhuman/tools/ops.rs | 2 +- vendor/tinyflows | 2 +- 20 files changed, 1689 insertions(+), 199 deletions(-) create mode 100644 src/openhuman/tinyflows/memory_adapter.rs create mode 100644 src/openhuman/tinyflows/memory_adapter_tests.rs create mode 100644 src/openhuman/tinyflows/memory_node_e2e_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 31774b1e1..15b9c82af 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -417,6 +417,7 @@ dependencies = [ "tower", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -435,6 +436,7 @@ dependencies = [ "sync_wrapper", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -3059,7 +3061,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.57.0", + "windows-core 0.58.0", ] [[package]] @@ -7299,14 +7301,19 @@ name = "tinyflows" version = "0.5.1" dependencies = [ "async-trait", + "axum", "futures-timer", + "futures-util", + "getrandom 0.3.4", "jaq-core", "jaq-json", "jaq-std", + "reqwest 0.12.28", "serde", "serde_json", "thiserror 2.0.18", "tinyagents", + "tokio", "tracing", ] @@ -7641,6 +7648,7 @@ dependencies = [ "tokio", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -7679,6 +7687,7 @@ version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ + "log", "pin-project-lite", "tracing-attributes", "tracing-core", diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 13949d84e..94ad8b9e7 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -581,6 +581,7 @@ dependencies = [ "tower", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -599,6 +600,7 @@ dependencies = [ "sync_wrapper", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -9015,14 +9017,19 @@ name = "tinyflows" version = "0.5.1" dependencies = [ "async-trait", + "axum", "futures-timer", + "futures-util", + "getrandom 0.3.4", "jaq-core", "jaq-json", "jaq-std", + "reqwest 0.12.28", "serde", "serde_json", "thiserror 2.0.18", "tinyagents", + "tokio", "tracing", ] @@ -9356,6 +9363,7 @@ dependencies = [ "tokio", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -9394,6 +9402,7 @@ version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ + "log", "pin-project-lite", "tracing-attributes", "tracing-core", diff --git a/src/openhuman/about_app/catalog_data.rs b/src/openhuman/about_app/catalog_data.rs index 09f3c390e..427cd77bb 100644 --- a/src/openhuman/about_app/catalog_data.rs +++ b/src/openhuman/about_app/catalog_data.rs @@ -1511,6 +1511,23 @@ pub(super) const CAPABILITIES: &[Capability] = &[ status: CapabilityStatus::Beta, privacy: DERIVED_TO_BACKEND, }, + Capability { + id: "automation.flow_memory_node", + name: "Memory Node (Flows)", + domain: "flows", + category: CapabilityCategory::Automation, + description: "A `memory` node inside a saved workflow graph, giving the flow direct, \ + in-graph memory access with no agent turn involved. It can recall/search/ \ + read style-flavour/look up people from your durable, cross-flow memory \ + (read-only — a flow can never write there) or from other flows' own \ + memory (also read-only), and can remember/forget entries in its OWN \ + private, flow-scoped memory namespace — never the user's personal memory, \ + never another flow's. Every operation is gated by the flow's autonomy \ + tier; a flow-scoped write can require human approval.", + how_to: "Flows editor > add a `memory` node; set `config.operation` and `config.scope`.", + status: CapabilityStatus::Beta, + privacy: LOCAL_RAW, + }, Capability { id: "automation.view_cron_jobs", name: "View Cron Jobs", diff --git a/src/openhuman/flows/agents/workflow_builder/builder_prompt.rs b/src/openhuman/flows/agents/workflow_builder/builder_prompt.rs index 114bdfcfc..b499d8737 100644 --- a/src/openhuman/flows/agents/workflow_builder/builder_prompt.rs +++ b/src/openhuman/flows/agents/workflow_builder/builder_prompt.rs @@ -830,21 +830,25 @@ mod tests { ); } - /// The three mechanisms that DO reach memory from inside a running flow + /// The four mechanisms that DO reach memory from inside a running flow /// must all be taught, with the correct binding path for the - /// deterministic one. A native `oh:` tool result is a `ToolResult` — - /// `{ content: [{ type, text }], is_error }` — so a downstream binding - /// dereferences `.item.json.content[0].text`, not the bare - /// `.item.json.` an agent/`http_request` output would use. Getting - /// that path wrong is the same class of silent-null failure the + /// deterministic `tool_call` one. A native `oh:` tool result is a + /// `ToolResult` — `{ content: [{ type, text }], is_error }` — so a + /// downstream binding dereferences `.item.json.content[0].text`, not the + /// bare `.item.json.` an agent/`http_request` output would use. + /// Getting that path wrong is the same class of silent-null failure the /// `=`-binding rules exist to stop. #5204 added `flow_memory_agent` as - /// the third (and now PREFERRED general) route alongside the - /// deterministic `tool_call` reads and `context_scout`'s narrower niche. + /// the PREFERRED general route alongside the deterministic `tool_call` + /// reads and `context_scout`'s narrower niche; the memory-node feature + /// (issue #5226) then added the `memory` node itself as the preferred + /// choice specifically for a non-reasoning node (`condition`/`switch`) + /// that needs to branch on a recalled value. #[test] - fn standing_prompt_teaches_the_three_working_memory_read_paths() { + fn standing_prompt_teaches_the_four_working_memory_read_paths() { const STANDING_PROMPT: &str = include_str!("prompt.md"); for rule in [ + "A `memory` node", "oh:memory_recall", "oh:memory_hybrid_search", "flow_memory_agent", @@ -853,7 +857,7 @@ mod tests { ] { assert!( STANDING_PROMPT.contains(rule), - "standing prompt must teach `{rule}` — it is one of the only three \ + "standing prompt must teach `{rule}` — it is one of the four \ mechanisms that actually read memory at flow run time, or the \ binding path needed to consume one" ); @@ -861,23 +865,31 @@ mod tests { } /// Flows run on trigger data a third party can influence (an inbound - /// email, a webhook payload), so writing that into the user's personal - /// memory is deliberately not offered. `agent_memory` is NOT an escape - /// hatch here despite being a registered, `read_only` builtin: its + /// email, a webhook payload), so writing to the user's PERSONAL memory is + /// deliberately never offered — that guarantee must survive the + /// memory-node feature (issue #5226) verbatim. `agent_memory` is NOT an + /// escape hatch here despite being a registered, `read_only` builtin: its /// `memory_tree` tool inherits the trait-default `PermissionLevel::ReadOnly` /// while dispatching an `ingest_document` WRITE mode, so it survives the /// read-only tool filter in `session/builder/factory.rs` (which consults /// the argless `permission_level()`). Steering the builder there would /// hand prompt-injected trigger content a memory-write foothold — exactly /// the hole `context_scout`'s own agent.toml documents refusing. + /// + /// What DID change with #5226: a flow can now write its OWN private, + /// flow-scoped memory (`memory` node, `scope: "flow"`) — the prompt must + /// teach that too, with the "remember after the action, not before" rule, + /// so the builder stops telling users memory writes are unavailable + /// entirely and instead reaches for the real mechanism. #[test] - fn standing_prompt_states_flows_cannot_write_memory_and_avoids_agent_memory() { + fn standing_prompt_states_flows_cannot_write_user_memory_but_can_write_flow_memory() { const STANDING_PROMPT: &str = include_str!("prompt.md"); assert!( STANDING_PROMPT.contains("can never WRITE the user's memory"), "standing prompt must state plainly that a workflow cannot write the \ - user's memory, so the builder stops authoring remember/store steps" + user's PERSONAL memory, so the builder never targets scope \"user\" \ + on a remember/forget memory node" ); assert!( !STANDING_PROMPT.contains("agent_memory"), @@ -886,6 +898,19 @@ mod tests { an ingest_document write mode, so it would give prompt-injectable \ trigger data a memory-write path" ); + assert!( + STANDING_PROMPT.contains("scope: \"flow\""), + "standing prompt must teach that a workflow CAN write its own \ + flow-scoped memory via a `memory` node (scope: \"flow\") — this is \ + the real mechanism for a flow that \"remembers\" across runs, \ + replacing the old blanket \"memory writes are not available\" advice" + ); + assert!( + STANDING_PROMPT.contains("Always place the `remember` AFTER the real action"), + "standing prompt must teach commit-on-success ordering: remember AFTER \ + the action it's recording, never before, so a failed action doesn't \ + get silently marked done" + ); } #[test] diff --git a/src/openhuman/flows/agents/workflow_builder/prompt.md b/src/openhuman/flows/agents/workflow_builder/prompt.md index 64a9f6307..50983b878 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -154,7 +154,7 @@ rather than a general context recall), use `memory_hybrid_search` in its You have a machine-readable belt; use it instead of relying on memory: -- **Introspect the DSL:** `list_node_kinds` → the 12 kinds; `get_node_kind_contract +- **Introspect the DSL:** `list_node_kinds` → the 13 kinds; `get_node_kind_contract { kind }` → one kind's exact config fields, ports, an example, and its gotchas. Consult these instead of guessing config shapes (this is the source of truth; the summary below is just orientation). @@ -304,7 +304,7 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. - **Exactly ONE `trigger` node is required.** Every other node should be reachable from it; a dry-run helps catch orphans. -### The 12 node kinds +### The 13 node kinds > The authoritative, always-current config shapes, ports, examples, and gotchas > for each kind live in the `list_node_kinds` / `get_node_kind_contract { kind }` @@ -362,12 +362,23 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. memory access: it is a single completion, so it cannot look anything up and it cannot decide to. Prompting one to "recall the user's preference" does not read memory — the model simply INVENTS an answer, and the graph - still looks correct. Never author that. Three mechanisms actually work: + still looks correct. Never author that. Four mechanisms actually work: + - **A `memory` node** (`config.operation: "recall"` or `"search"`, + `config.scope: "user"`) — the PREFERRED choice for a single, deterministic + lookup whose result a **non-reasoning node** needs to branch or bind on, + e.g. a `condition` gating on whether something was already found. It is a + verb with static config, not a reasoning step, so it fires exactly once + per item and can't loop or decide what to look up next — use `tool_call + oh:memory_recall`/`oh:memory_hybrid_search` (below) only when you + specifically need that native-tool result shape instead. See "The + `memory` node" below for the full operation/scope reference. - **A `tool_call` node** with `config.slug` = `oh:memory_recall` (semantic - recall) or `oh:memory_hybrid_search` (keyword/lexical lookup). One - deterministic read at a fixed point in the graph. Its output is a native + recall) or `oh:memory_hybrid_search` (keyword/lexical lookup). Same + one-shot-read shape as the `memory` node above, but returns a native tool result, so bind downstream off - `=nodes..item.json.content[0].text` — NOT `.item.json.`. + `=nodes..item.json.content[0].text` — NOT `.item.json.`. Both + are valid; prefer the `memory` node for new graphs unless you need this + exact output shape. - **`config.agent_ref` = `flow_memory_agent`** — the PREFERRED general route: any step that needs the user's context, style, history, or people → `flow_memory_agent` via `agent_ref`, for ANY use case, not a fixed list. @@ -384,13 +395,34 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. `recommended_skills`). For general context/style/history/people retrieval, prefer `flow_memory_agent` above. - **A workflow can never WRITE the user's memory.** There is no - remember/store step, and no `agent_ref` that grants one — a flow runs on - trigger data that a third party can influence (an inbound email, a webhook - payload), so writing that into the user's memory is deliberately not - possible. If the user asks for a workflow that "remembers" something - across runs, say plainly that memory writes are not available to workflows - yet and build the rest of the flow without that step. + **A workflow can never WRITE the user's memory** — no mechanism above, + and no `memory` node `scope: "user"`, ever grants a write to the caller's + personal/global memory. `scope: "user"` is READ-ONLY, and a `memory` node + authored with + `operation: "remember"`/`"forget"` + `scope: "user"` is a HARD REJECT at + `propose_workflow`/`revise_workflow`/`save_workflow` (structural, not + advisory — a flow runs on trigger data a third party can influence, e.g. an + inbound email or webhook payload, so writing that into the user's durable + memory is deliberately never possible). + + **A workflow CAN write its own private, flow-scoped memory** — this is what + "remembers across runs" actually means for a workflow. A `memory` node with + `operation: "remember"`/`"forget"` + `scope: "flow"` reads/writes a sandbox + namespace unique to that saved flow (never the user's memory, never another + flow's). **Always place the `remember` AFTER the real action, never before** + — if the action fails, the item was never marked done, so the next run + retries it instead of silently skipping it. If the user asks for a workflow + that "remembers" something, this is the mechanism: build it with a `memory` + node at `scope: "flow"`, not by claiming memory writes are unavailable. + + **Exact "process each item once" dedup is NOT reliably expressible this + way.** Semantic `recall` ranks results by similarity, not exact key + membership, so there is no sound `recall → condition` pattern that + correctly answers "have I already handled this exact item" — don't + improvise one. A dedicated dedup primitive is deferred to a future + iteration; until it lands, tell the user the workflow can't guarantee + exactly-once processing rather than shipping a graph that looks like it + dedupes but doesn't. Use memory reads sparingly — only when the workflow genuinely needs the user's context, rather than hardcoding what memory already holds. @@ -531,6 +563,56 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. 10. **`transform`** — `config.set` = `{ key: "=expr" }`, merged onto each item. 11. **`output_parser`** — passthrough today; no config required. 12. **`sub_workflow`** — `config.workflow` = an embedded child `WorkflowGraph`. +13. **`memory`** — reads or writes host-managed memory directly, no agent turn + involved. See "The `memory` node" just below for the full reference. + +### The `memory` node + +A `memory` node is a **verb with static config**, not a reasoning step — it +fires once per item (or once per run with `execution: "once"`) and can't loop +or decide what to look up next. Its unique value over `flow_memory_agent` is +serving nodes that **cannot reason at all**: a `condition`/`switch` node can +branch on a `memory` node's recalled output, but it can't call a tool or run +an agent turn itself. + +- **`config.operation`** (required) — one of `recall` · `search` · `flavour` · + `people` · `remember` · `forget`. +- **`config.scope`** (required for `recall`/`search`/`remember`/`forget`) — + one of: + - `"user"` — the caller's durable, cross-flow memory. **READ-ONLY** — + `remember`/`forget` may never target it (hard reject at + `propose_workflow`/`revise_workflow`/`save_workflow`, before a run ever + starts). + - `"flow"` — this flow's own private memory namespace. The **only** scope + `remember`/`forget` may target. Reads/writes the SAME namespace the + `flow_memory_recall`/`flow_memory_remember` agent tools use, so a + `memory` node and a `flow_memory_agent` step in the same flow see each + other's writes. + - `"flows"` — READ-ONLY visibility across every flow's own private memory + (never the user's personal/global memory) — useful when related flows + should dedupe against each other. +- **`config.query`** (`=`-bindable, required for `recall`/`search`, optional + for `people`) — the lookup query. +- **`config.flavour`** (required for `flavour`) — a persona facet slug: one of + `communication` · `coding_style` · `stack` · `workflow` · `environment` · + `directives` · `anti_preferences`, e.g. `"communication"`. +- **`config.key`** / **`config.value`** (`=`-bindable, required for + `remember`/`forget`) — the memory key to write/delete, and the value to + store. +- **`config.limit`** / **`config.min_score`** (optional, `recall`/`search`) — + cap the result count / relevance floor. + +**Dedup is out of scope for this node.** Exact "have I already processed this +item" membership checks are not reliably expressible via semantic `recall` — +`results` is similarity-ranked, not an exact-match lookup, so a +`recall → condition` graph cannot safely gate on it. Don't author one; a +dedicated dedup primitive is deferred to a future iteration. Put `remember` +**after** the real action it records, not before — a failed action must never +be mistaken for a completed one on the next run. See the `agent` node kind's +"Reading the user's memory at run time" section above for how this node +relates to `tool_call oh:memory_recall` and `flow_memory_agent` — all three +are valid; `memory` is the right choice specifically when a non-reasoning node +needs to branch on the result. ### Expressions: the `=` / jq convention diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index 2f756f5a0..c7a20f6bc 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -2343,7 +2343,7 @@ impl Tool for ListAgentProfilesTool { // list_node_kinds / get_node_kind_contract — queryable DSL schema (F2) // ───────────────────────────────────────────────────────────────────────────── -/// `list_node_kinds`: enumerate the 12 tinyflows node kinds with a one-line +/// `list_node_kinds`: enumerate the 13 tinyflows node kinds with a one-line /// summary each. The DSL counterpart of `search_tool_catalog` for Composio /// actions — a cheap first call to orient before fetching a full contract. pub struct ListNodeKindsTool; @@ -2369,7 +2369,7 @@ impl Tool for ListNodeKindsTool { } fn description(&self) -> &str { - "List the 12 tinyflows node kinds you can put in a WorkflowGraph, each with a one-line \ + "List the 13 tinyflows node kinds you can put in a WorkflowGraph, each with a one-line \ summary and its config field names. Read-only, no args. Returns a JSON array of { kind, \ summary, required_config, optional_config }. Call get_node_kind_contract { kind } for the \ full config-field shapes, ports, an example node, and authoring gotchas of any one kind — \ @@ -2460,7 +2460,7 @@ impl Tool for GetNodeKindContractTool { "properties": { "kind": { "type": "string", - "description": "One of the 12 node kinds, e.g. 'tool_call' (from list_node_kinds).", + "description": "One of the 13 node kinds, e.g. 'tool_call' (from list_node_kinds).", "enum": crate::openhuman::flows::NODE_KINDS, } }, @@ -2488,7 +2488,7 @@ impl Tool for GetNodeKindContractTool { &contract, )?)), None => Ok(ToolResult::error(format!( - "'{kind}' is not a tinyflows node kind — call list_node_kinds for the 12 valid \ + "'{kind}' is not a tinyflows node kind — call list_node_kinds for the 13 valid \ kinds." ))), } diff --git a/src/openhuman/flows/builder_tools_tests.rs b/src/openhuman/flows/builder_tools_tests.rs index af765479a..f72ea1618 100644 --- a/src/openhuman/flows/builder_tools_tests.rs +++ b/src/openhuman/flows/builder_tools_tests.rs @@ -1945,15 +1945,16 @@ async fn save_workflow_accepts_correctly_schemad_graph() { } #[tokio::test] -async fn list_node_kinds_tool_returns_all_twelve() { +async fn list_node_kinds_tool_returns_all_thirteen() { let tool = ListNodeKindsTool::new(); let result = tool.execute(json!({})).await.unwrap(); assert!(!result.is_error, "{}", result.output()); let parsed: Value = serde_json::from_str(&result.output()).unwrap(); let kinds = parsed["node_kinds"].as_array().unwrap(); - assert_eq!(kinds.len(), 12); + assert_eq!(kinds.len(), 13); // Each entry carries a kind + summary + the config-field name lists. assert!(kinds.iter().any(|k| k["kind"] == "tool_call")); + assert!(kinds.iter().any(|k| k["kind"] == "memory")); assert!(kinds.iter().all(|k| k.get("summary").is_some())); } diff --git a/src/openhuman/flows/memory_tools.rs b/src/openhuman/flows/memory_tools.rs index a275ed39e..292a4ed53 100644 --- a/src/openhuman/flows/memory_tools.rs +++ b/src/openhuman/flows/memory_tools.rs @@ -114,6 +114,55 @@ pub fn flow_namespace(flow_id: &str) -> String { /// 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 merged across **every** flow's own `flow_` memory +/// namespace — never the user's personal/global memory, and never any +/// namespace outside the `flow_*` prefix. +/// +/// Shared by [`FlowMemoryRecallTool`]'s `scope: "flows"` arm and the +/// tinyflows `memory` node's `scope: "flows"` (`OpenHumanMemory::recall` in +/// `crate::openhuman::tinyflows::memory_adapter`) — both surfaces must see +/// identical cross-flow results, so this is the one place that walks +/// [`Memory::namespace_summaries`] and filters to `flow_*`. A per-namespace +/// recall failure is logged and skipped rather than failing the whole call, +/// so one corrupt/unavailable flow namespace can't blank out every other +/// flow's results. +pub async fn cross_flow_recall( + memory: &Arc, + query: &str, + limit: usize, + min_score: Option, +) -> anyhow::Result> { + let summaries = memory.namespace_summaries().await?; + let mut merged: Vec = Vec::new(); + for summary in summaries + .iter() + .filter(|s| s.namespace.starts_with(FLOW_MEMORY_NAMESPACE_LISTED_PREFIX)) + { + let opts = RecallOpts { + namespace: Some(summary.namespace.as_str()), + min_score, + ..RecallOpts::default() + }; + match memory.recall(query, limit, opts).await { + Ok(entries) => merged.extend(entries), + Err(e) => { + log::warn!( + "[flows:memory] cross_flow_recall 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(merged) +} + /// Read-only recall over a flow's own memory namespace, or (with /// `scope: "flows"`) across every flow's namespace. /// @@ -255,44 +304,12 @@ impl Tool for FlowMemoryRecallTool { Err(e) => Ok(ToolResult::error(format!("Flow memory recall failed: {e}"))), } } - "flows" => { - let summaries = match self.memory.namespace_summaries().await { - Ok(summaries) => summaries, - Err(e) => { - return Ok(ToolResult::error(format!( - "Failed to list flow memory namespaces: {e}" - ))) - } - }; - - let mut merged: Vec = Vec::new(); - for summary in summaries - .iter() - .filter(|s| s.namespace.starts_with(FLOW_MEMORY_NAMESPACE_LISTED_PREFIX)) - { - let opts = RecallOpts { - namespace: Some(summary.namespace.as_str()), - ..RecallOpts::default() - }; - match self.memory.recall(query, limit, opts).await { - Ok(entries) => merged.extend(entries), - Err(e) => { - log::warn!( - "[flows:memory] flow_memory_recall scope=flows failed for namespace={}: {e}", - summary.namespace - ); - } - } - } - merged.sort_by(|a, b| { - b.score - .unwrap_or(0.0) - .partial_cmp(&a.score.unwrap_or(0.0)) - .unwrap_or(std::cmp::Ordering::Equal) - }); - merged.truncate(limit); - Ok(ToolResult::success(render_entries(&merged))) - } + "flows" => match cross_flow_recall(&self.memory, query, limit, None).await { + Ok(merged) => Ok(ToolResult::success(render_entries(&merged))), + Err(e) => Ok(ToolResult::error(format!( + "Failed to list flow memory namespaces: {e}" + ))), + }, other => Ok(ToolResult::error(format!( "Unknown scope '{other}': expected 'flow' or 'flows'" ))), diff --git a/src/openhuman/flows/mod.rs b/src/openhuman/flows/mod.rs index 04dacac4f..a1b6ae410 100644 --- a/src/openhuman/flows/mod.rs +++ b/src/openhuman/flows/mod.rs @@ -51,4 +51,9 @@ pub use types::{ // `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}; +// `cross_flow_recall` is re-exported for the same reason: the tinyflows +// `memory` node's `OpenHumanMemory` adapter (`scope: "flows"` recall) must +// see byte-identical cross-flow results to `flow_memory_recall`'s own +// `scope: "flows"` arm, so both call the one implementation here rather than +// each walking `namespace_summaries` independently. +pub use memory_tools::{cross_flow_recall, flow_namespace, FLOW_MEMORY_NAMESPACE_PREFIX}; diff --git a/src/openhuman/flows/node_contracts.rs b/src/openhuman/flows/node_contracts.rs index e2e14ebd8..520bc35ae 100644 --- a/src/openhuman/flows/node_contracts.rs +++ b/src/openhuman/flows/node_contracts.rs @@ -73,11 +73,20 @@ fn apply_host_overlay(contract: NodeKindContract) -> NodeKindContract { default the path to \"json.data\" (that targets the whole payload container and \ yields one item) — probe the real array path with get_tool_output_sample instead.", ), + "memory" => contract.with_note( + "scope: \"flow\" reads/writes the SAME per-flow memory namespace the \ + flow_memory_recall / flow_memory_remember agent tools use — a memory[remember] \ + node and a flow_memory_remember tool call inside an agent node on the same flow \ + see each other's writes. Exact \"process each item once\" dedup is NOT reliably \ + expressible via semantic recall (results are similarity-ranked, not an exact \ + membership check) — that is deferred to a dedicated primitive; don't improvise a \ + recall→condition dedupe graph.", + ), _ => contract, } } -/// All 12 node-kind contracts with this host's overlay applied, in +/// All 13 node-kind contracts with this host's overlay applied, in /// [`NODE_KINDS`] order. pub fn all_node_kind_contracts() -> Vec { tinyflows::catalog::all_contracts() @@ -87,7 +96,7 @@ pub fn all_node_kind_contracts() -> Vec { } /// The overlaid contract for one node kind, or `None` if `kind` is not one of -/// the 12. +/// the 13. pub fn node_kind_contract(kind: &str) -> Option { tinyflows::catalog::contract_for(kind).map(apply_host_overlay) } @@ -137,14 +146,28 @@ mod tests { use super::*; #[test] - fn overlay_preserves_all_12_kinds() { - assert_eq!(all_node_kind_contracts().len(), 12); + fn overlay_preserves_all_13_kinds() { + assert_eq!(all_node_kind_contracts().len(), 13); for kind in NODE_KINDS { assert!(node_kind_contract(kind).is_some(), "missing {kind}"); } assert!(node_kind_contract("not_a_kind").is_none()); } + #[test] + fn memory_overlay_adds_flow_memory_coherence_facts() { + let c = node_kind_contract("memory").unwrap(); + let notes = c.notes.join("\n"); + assert!(notes.contains("flow_memory_recall"), "{notes}"); + assert!(notes.contains("flow_memory_remember"), "{notes}"); + assert!(notes.contains("SAME per-flow memory namespace"), "{notes}"); + // The dedup recipe was removed (P1 review fix): semantic recall + // cannot express exact "have I seen this key" membership, so the + // overlay must not teach a recall→condition dedupe pattern. + assert!(!notes.contains("Canonical dedupe pattern"), "{notes}"); + assert!(!notes.contains("item.json.found"), "{notes}"); + } + #[test] fn tool_call_overlay_adds_host_composio_facts() { let c = node_kind_contract("tool_call").unwrap(); diff --git a/src/openhuman/flows/tools.rs b/src/openhuman/flows/tools.rs index 257fc3f0a..685eb9dce 100644 --- a/src/openhuman/flows/tools.rs +++ b/src/openhuman/flows/tools.rs @@ -57,7 +57,7 @@ impl Tool for ProposeWorkflowTool { (to_port just stays \"main\"); routing is keyed exclusively on from_port, so a label \ on to_port instead silently turns the branch into an unconditional fan-out and is a \ hard reject. Exactly ONE \ - trigger node is required. The 12 node kinds: trigger (config.trigger_kind: manual | \ + trigger node is required. The 13 node kinds: trigger (config.trigger_kind: manual | \ schedule | webhook | app_event | form | chat_message | evaluation | system | \ execute_by_workflow; schedule needs config.schedule = {kind:\"cron\",expr,tz?} | \ {kind:\"at\",at} | {kind:\"every\",every_ms}; app_event needs config.toolkit + \ @@ -69,7 +69,15 @@ impl Tool for ProposeWorkflowTool { the matching case port, or \"default\"), transform (config.set: {key: \"=expr\"} \ merged onto each item), split_out (config.path to an array field; fans out one item per \ element), merge (fan-in passthrough, no config), output_parser (passthrough today; no \ - config required), sub_workflow (config.workflow: an embedded child WorkflowGraph). If \ + config required), sub_workflow (config.workflow: an embedded child WorkflowGraph), \ + memory (config.operation REQUIRED: recall | search | flavour | people | remember | \ + forget; config.scope for recall/remember/forget: \"user\" is READ-ONLY, \"flow\" is \ + this flow's own memory and the ONLY scope remember/forget may target, \"flows\" is \ + cross-flow READ-ONLY; config.query for recall/search; config.flavour for the flavour \ + slug; config.key/config.value for remember/forget. Place remember AFTER the real \ + action it records, never before, so a failed action never marks an item as done. \ + Exact \"process each item once\" dedup is not reliably expressible via semantic \ + recall — don't improvise a recall/condition dedupe graph). If \ validation fails, fix the graph and call this tool again." } @@ -96,7 +104,7 @@ impl Tool for ProposeWorkflowTool { "enum": [ "trigger", "agent", "tool_call", "http_request", "code", "condition", "switch", "merge", "split_out", - "transform", "output_parser", "sub_workflow" + "transform", "output_parser", "sub_workflow", "memory" ] }, "name": { "type": "string", "description": "Human-readable node name." }, @@ -467,6 +475,14 @@ fn config_hint(node: &Node) -> Option { .and_then(Value::as_str) .map(|p| format!("path: {p}")), NodeKind::SubWorkflow => Some("embedded sub-workflow".to_string()), + NodeKind::Memory => { + let operation = cfg.get("operation").and_then(Value::as_str).unwrap_or("?"); + let hint = match cfg.get("scope").and_then(Value::as_str) { + Some(scope) => format!("{operation} · {scope}"), + None => operation.to_string(), + }; + Some(truncate_hint(&hint)) + } NodeKind::Merge | NodeKind::OutputParser | NodeKind::Trigger => None, } } diff --git a/src/openhuman/memory/tools.rs b/src/openhuman/memory/tools.rs index 6e2fd5aae..445896966 100644 --- a/src/openhuman/memory/tools.rs +++ b/src/openhuman/memory/tools.rs @@ -1,5 +1,10 @@ mod doctor; -mod flavour; +// `pub(crate)` (not `mod`): the tinyflows `memory` node's `OpenHumanMemory` +// adapter (`crate::openhuman::tinyflows::memory_adapter`) reaches +// `flavour::lookup_flavour` / `flavour::FlavourLookup` directly so the node's +// `flavour` operation and `MemoryFlavourTool` share one flavoured-tree read +// path — see `lookup_flavour`'s doc comment. +pub(crate) mod flavour; mod forget; mod recall; mod store; diff --git a/src/openhuman/memory/tools/flavour.rs b/src/openhuman/memory/tools/flavour.rs index 00c19c20f..a1135be00 100644 --- a/src/openhuman/memory/tools/flavour.rs +++ b/src/openhuman/memory/tools/flavour.rs @@ -58,6 +58,141 @@ fn body_after_front_matter(content: &str) -> &str { } } +/// Outcome of [`lookup_flavour`] — split from a hard `Err` so a caller can +/// distinguish "bad input" (never reached the store) from "reached the store +/// and here's what it found (or didn't, or a lookup itself failed)". +pub(crate) enum FlavourLookup { + /// A compiled profile body, ready to hand to the agent/node. + Profile(String), + /// No profile has been built yet for this facet — not an error, just + /// empty (persona ingestion hasn't run, or produced nothing for this + /// facet yet). + NotBuilt(String), + /// The tree lookup or compile step itself failed (I/O, corrupt tree, + /// …) — distinct from `NotBuilt` because this IS an error, just one + /// discovered after `flavour_raw` was already validated. + Failed(String), +} + +/// Pure lookup shared by [`MemoryFlavourTool::execute`] and the tinyflows +/// `memory` node's `flavour` operation +/// (`OpenHumanMemory::flavour` in `crate::openhuman::tinyflows::memory_adapter`) +/// — both surfaces read the exact same flavoured-tree path, so there is only +/// one place that knows how a `flavour` slug resolves to a compiled profile. +/// +/// `Err` is reserved for input the caller should have caught before ever +/// reaching the store (empty/unknown `flavour_raw`); everything the store +/// itself can report — hit, miss, or lookup failure — comes back as `Ok` of +/// the matching [`FlavourLookup`] variant so callers can shape each case +/// (tool result vs. node output) however their surface needs. +pub(crate) fn lookup_flavour(config: &Config, flavour_raw: &str) -> Result { + let flavour_raw = flavour_raw.trim(); + if flavour_raw.is_empty() { + return Err("'flavour' cannot be empty".to_string()); + } + + let facet = PersonaFacet::parse_loose(flavour_raw).ok_or_else(|| { + format!("Unknown flavour '{flavour_raw}'. Valid flavours: {VALID_FLAVOURS}") + })?; + + let mc = memory_config_from(config, config.workspace_dir.clone()); + let scope = facet.tree_scope(); + let heading = facet.heading(); + + tracing::debug!( + target: "memory_flavour", + flavour = flavour_raw, + facet = ?facet, + "[memory_flavour] entry" + ); + + // Fast path: the compiled artifact already exists on disk with a + // non-empty body — read it directly without touching the tree store. + let abs_path = flavoured_root_abs_path(&mc, &scope); + if abs_path.is_file() { + if let Ok(content) = std::fs::read_to_string(&abs_path) { + let body = body_after_front_matter(&content); + if !body.trim().is_empty() { + tracing::debug!( + target: "memory_flavour", + flavour = flavour_raw, + body_len = body.len(), + "[memory_flavour] fast path hit: returning stripped body from disk" + ); + return Ok(FlavourLookup::Profile(body.to_string())); + } + } + } + + tracing::debug!( + target: "memory_flavour", + flavour = flavour_raw, + "[memory_flavour] fast path missed or empty, falling to tree lookup" + ); + + // Slow path: look up the flavoured tree and (re)compile its root. + match get_tree_by_scope(&mc, TreeKind::Flavoured, &scope) { + Ok(None) => { + tracing::debug!( + target: "memory_flavour", + flavour = flavour_raw, + "[memory_flavour] no flavoured tree exists yet" + ); + Ok(FlavourLookup::NotBuilt(format!( + "No profile built yet for {heading}. Run persona ingestion first, then try \ + again." + ))) + } + Ok(Some(tree)) => { + tracing::debug!( + target: "memory_flavour", + flavour = flavour_raw, + tree_id = %tree.id, + "[memory_flavour] tree found, compiling root" + ); + match compile_flavoured_root(&mc, &tree.id) { + Ok(markdown) => { + let body = body_after_front_matter(&markdown); + if body.trim().is_empty() { + Ok(FlavourLookup::NotBuilt(format!( + "No profile built yet for {heading}. Run persona ingestion \ + first, then try again." + ))) + } else { + tracing::debug!( + target: "memory_flavour", + flavour = flavour_raw, + body_len = body.len(), + "[memory_flavour] compiled profile returned" + ); + Ok(FlavourLookup::Profile(body.to_string())) + } + } + Err(err) => { + tracing::warn!( + %err, + flavour = flavour_raw, + "[memory_flavour] failed to compile flavoured profile" + ); + Ok(FlavourLookup::Failed(format!( + "Failed to compile the {heading} profile: {err}" + ))) + } + } + } + Err(err) => { + tracing::warn!( + %err, + flavour = flavour_raw, + "[memory_flavour] failed to look up flavoured tree" + ); + Ok(FlavourLookup::Failed(format!( + "Failed to look up the {heading} profile: {err}" + ))) + } + } +} + #[async_trait] impl Tool for MemoryFlavourTool { fn name(&self) -> &str { @@ -104,111 +239,13 @@ impl Tool for MemoryFlavourTool { let flavour_raw = args .get("flavour") .and_then(|v| v.as_str()) - .ok_or_else(|| anyhow::anyhow!("Missing 'flavour' parameter"))? - .trim(); - if flavour_raw.is_empty() { - return Err(anyhow::anyhow!("'flavour' cannot be empty")); - } + .ok_or_else(|| anyhow::anyhow!("Missing 'flavour' parameter"))?; - let facet = PersonaFacet::parse_loose(flavour_raw).ok_or_else(|| { - anyhow::anyhow!("Unknown flavour '{flavour_raw}'. Valid flavours: {VALID_FLAVOURS}") - })?; - - let mc = memory_config_from(&self.config, self.config.workspace_dir.clone()); - let scope = facet.tree_scope(); - let heading = facet.heading(); - - tracing::debug!( - target: "memory_flavour", - flavour = flavour_raw, - facet = ?facet, - "[memory_flavour] entry" - ); - - // Fast path: the compiled artifact already exists on disk with a - // non-empty body — read it directly without touching the tree store. - let abs_path = flavoured_root_abs_path(&mc, &scope); - if abs_path.is_file() { - if let Ok(content) = std::fs::read_to_string(&abs_path) { - let body = body_after_front_matter(&content); - if !body.trim().is_empty() { - tracing::debug!( - target: "memory_flavour", - flavour = flavour_raw, - body_len = body.len(), - "[memory_flavour] fast path hit: returning stripped body from disk" - ); - return Ok(ToolResult::success(body.to_string())); - } - } - } - - tracing::debug!( - target: "memory_flavour", - flavour = flavour_raw, - "[memory_flavour] fast path missed or empty, falling to tree lookup" - ); - - // Slow path: look up the flavoured tree and (re)compile its root. - match get_tree_by_scope(&mc, TreeKind::Flavoured, &scope) { - Ok(None) => { - tracing::debug!( - target: "memory_flavour", - flavour = flavour_raw, - "[memory_flavour] no flavoured tree exists yet" - ); - Ok(ToolResult::success(format!( - "No profile built yet for {heading}. Run persona ingestion first, then try \ - again." - ))) - } - Ok(Some(tree)) => { - tracing::debug!( - target: "memory_flavour", - flavour = flavour_raw, - tree_id = %tree.id, - "[memory_flavour] tree found, compiling root" - ); - match compile_flavoured_root(&mc, &tree.id) { - Ok(markdown) => { - let body = body_after_front_matter(&markdown); - if body.trim().is_empty() { - Ok(ToolResult::success(format!( - "No profile built yet for {heading}. Run persona ingestion \ - first, then try again." - ))) - } else { - tracing::debug!( - target: "memory_flavour", - flavour = flavour_raw, - body_len = body.len(), - "[memory_flavour] compiled profile returned" - ); - Ok(ToolResult::success(body.to_string())) - } - } - Err(err) => { - tracing::warn!( - %err, - flavour = flavour_raw, - "[memory_flavour] failed to compile flavoured profile" - ); - Ok(ToolResult::error(format!( - "Failed to compile the {heading} profile: {err}" - ))) - } - } - } - Err(err) => { - tracing::warn!( - %err, - flavour = flavour_raw, - "[memory_flavour] failed to look up flavoured tree" - ); - Ok(ToolResult::error(format!( - "Failed to look up the {heading} profile: {err}" - ))) - } + match lookup_flavour(&self.config, flavour_raw) { + Err(hard) => Err(anyhow::anyhow!(hard)), + Ok(FlavourLookup::Profile(body)) => Ok(ToolResult::success(body)), + Ok(FlavourLookup::NotBuilt(msg)) => Ok(ToolResult::success(msg)), + Ok(FlavourLookup::Failed(msg)) => Ok(ToolResult::error(msg)), } } } diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs index 47ab33352..2a6c78be8 100644 --- a/src/openhuman/tinyflows/caps.rs +++ b/src/openhuman/tinyflows/caps.rs @@ -99,7 +99,13 @@ fn model_response_to_completion_value( /// would otherwise silently override the tier's `Prompt` decision). The error /// is prefixed with [`POLICY_BLOCKED_MARKER`] so the harness's repeated-failure /// middleware recognizes it as a permanent, don't-retry refusal. -fn enforce_node_tier_gate( +/// +/// `pub(crate)` (not `http_request`/`code`-private): the `memory` node's +/// [`OpenHumanMemory`](super::memory_adapter::OpenHumanMemory) adapter reuses +/// this exact function — `CommandClass::Read` for recall/search/flavour/people, +/// `CommandClass::Write` for remember/forget — rather than growing a second +/// permission path for the new node kind. +pub(crate) fn enforce_node_tier_gate( security: &SecurityPolicy, class: CommandClass, node: &str, @@ -152,7 +158,7 @@ fn enforce_node_tier_gate( /// call only* `Workflow { require_approval: true }` origin around /// `intercept_audited`, forcing the real parking/HITL flow. `GateDecision::Allow` /// (and any other origin shape) passes through unchanged — existing behavior. -async fn gate_call_for_tier( +pub(crate) async fn gate_call_for_tier( tier_decision: GateDecision, tool_name: &str, action_summary: &str, @@ -3553,13 +3559,18 @@ impl WorkflowResolver for OpenHumanWorkflowResolver { } } -/// Builds the [`Capabilities`] bundle for one run, wiring each of the six -/// host-injected traits to a real OpenHuman adapter (see each adapter above for -/// its contract). +/// Builds the [`Capabilities`] bundle for one run, wiring each of the seven +/// host-injected traits to a real OpenHuman adapter (see each adapter above, +/// and [`super::memory_adapter::OpenHumanMemory`] for `memory`, for its +/// contract). /// /// `state_namespace` scopes the [`FlowStateStore`] KV so two saved flows that /// use the same state key never read or overwrite each other — callers pass a -/// per-flow namespace (e.g. `"flow:"`). +/// per-flow namespace (e.g. `"flow:"`). Note this is **not** the same +/// namespace `OpenHumanMemory` writes flow-scoped memory under — that one is +/// derived independently from the run's trusted origin via +/// `flows::flow_namespace`, so the two never need to agree on separator +/// conventions. pub fn build_capabilities(config: Arc, state_namespace: impl Into) -> Capabilities { let security = Arc::new(SecurityPolicy::from_config( &config.autonomy, @@ -3584,7 +3595,7 @@ pub fn build_capabilities(config: Arc, state_namespace: impl Into, state_namespace: impl Into` namespace those +//! agent tools use — both derive it from +//! [`crate::openhuman::flows::flow_namespace`], and `scope: "flows"` recall +//! delegates to the exact same [`crate::openhuman::flows::cross_flow_recall`] +//! those tools now share. A `memory[remember·flow]` node, the +//! `flow_memory_agent`, and the post-run digest subscriber therefore all read +//! and write one consistent store — never three namespace conventions that +//! happen to overlap by convention. +//! +//! **Defense-in-depth on writes.** The engine's own `validate_all` already +//! rejects `remember`/`forget` nodes authored with `scope: "user"` before a +//! run ever starts (`vendor/tinyflows/src/validate.rs`). `recall` hard-refuses +//! any `scope` it doesn't recognise; `remember` and `forget` separately +//! hard-refuse anything other than `scope: "flow"` — never `"user"`, never +//! `"flows"` — so even a compromised or out-of-band caller of this adapter +//! (bypassing the validator entirely) can never write outside a flow's own +//! sandboxed namespace. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::{json, Value}; +use tinyflows::caps::MemoryProvider; +use tinyflows::error::{EngineError, Result}; + +use crate::openhuman::agent::harness::memory_context_safety::{ + is_potentially_untrusted, wrap_untrusted_for_agent, +}; +use crate::openhuman::agent::turn_origin::{self, AgentTurnOrigin, TrustedAutomationSource}; +use crate::openhuman::approval::{redact_args, summarize_action, GateOutcome}; +use crate::openhuman::config::Config; +use crate::openhuman::flows::{cross_flow_recall, flow_namespace}; +use crate::openhuman::memory::tools::flavour::{lookup_flavour, FlavourLookup}; +use crate::openhuman::memory::{Memory, MemoryCategory, MemoryEntry, MemoryTaint, RecallOpts}; +use crate::openhuman::security::{CommandClass, SecurityPolicy}; + +use super::caps::{enforce_node_tier_gate, gate_call_for_tier}; + +/// Stable `tracing` grep prefix for every log line this adapter emits. Never +/// logs memory *content* or PII — only operation names, scopes, resolved +/// namespaces, tier-gate decisions, hit/miss, and result sizes. +const LOG_PREFIX: &str = "[memory-node-host]"; + +/// The user's durable, cross-flow memory namespace `scope: "user"` reads from +/// — the same default namespace `memory_recall`/`memory_hybrid_search` +/// resolve to when the agent doesn't name one explicitly +/// (`RecallOpts.namespace: None` falls back to this same constant inside the +/// store). Named here explicitly (rather than passing `None`) purely so it +/// shows up in the debug log. +const USER_NAMESPACE: &str = tinycortex::memory::GLOBAL_NAMESPACE; + +/// Host-injected memory access for `memory` nodes. See the module doc for the +/// security contract; see [`super::caps::OpenHumanAgentRunner`] for the +/// sibling adapter this one's tier-gate wiring mirrors. +pub struct OpenHumanMemory { + pub(crate) config: Arc, + pub(crate) security: Arc, +} + +impl OpenHumanMemory { + /// Resolves the process-global memory store, matching how every other + /// memory-backed agent tool (`flow_memory_recall`, `memory_recall`, the + /// post-run digest subscriber) reaches it — + /// `memory::ops::helpers::active_memory_client()` reuses the already- + /// initialised global client when ready, else lazily initialises it for + /// the current workspace. No adapter-local memory instance is ever + /// constructed, so there is exactly one on-disk store in play. + async fn memory(&self) -> Result> { + crate::openhuman::memory::ops::helpers::active_memory_client() + .await + .map(|client| client.memory_handle()) + .map_err(EngineError::Capability) + } + + /// The running flow's own id, from the run's trusted + /// `TrustedAutomation { Workflow }` turn origin — the ONLY authoritative + /// source. Mirrors `flows::memory_tools::trusted_flow_id`'s security + /// invariant exactly: this adapter's `scope` argument is a fixed + /// three-value enum (never a caller-supplied namespace string), but the + /// *flow id* half of the namespace must still come from the trusted run + /// context, not be re-derivable from anything model- or config-supplied, + /// so a `scope: "flow"` node can never be pointed at another flow's + /// namespace by a crafted `=`-binding. + fn trusted_flow_id(&self) -> Result { + match turn_origin::current() { + Some(AgentTurnOrigin::TrustedAutomation { + job_id, + source: TrustedAutomationSource::Workflow { .. }, + }) => Ok(job_id), + _ => Err(EngineError::Capability( + "memory node: scope \"flow\"/\"flows\" requires the node to run inside a saved \ + flow's own run (no trusted Workflow-scoped origin found)" + .to_string(), + )), + } + } + + /// [`flow_namespace`] for the running flow — the same `flow_` + /// namespace `flow_memory_recall`/`flow_memory_remember` read and write. + fn flow_memory_namespace(&self) -> Result { + self.trusted_flow_id().map(|id| flow_namespace(&id)) + } + + /// Read-side tier gate: `CommandClass::Read` is `Allow` at every autonomy + /// tier (`SecurityPolicy::gate_decision`), so this can only ever succeed + /// or `Block` outright (`ReadOnly`/`Supervised`/`Full` all `Allow` Read). + /// Still routed through [`enforce_node_tier_gate`] — rather than skipped + /// — so the tier-gate debug log line and the `Block` floor stay uniform + /// across every acting/reading adapter in this module. No + /// [`gate_call_for_tier`] round-trip: unlike `remember`/`forget`, a read + /// never needs the HITL escalation that function exists for, and — like + /// the curated-Read short-circuit in `OpenHumanTools::invoke` — routing + /// a guaranteed-`Allow` decision through `intercept_audited` anyway would + /// only reintroduce the "reads wait for approval" bug that short-circuit + /// was added to close. + fn tier_gate_read(&self, op: &str) -> Result<()> { + enforce_node_tier_gate(&self.security, CommandClass::Read, op)?; + Ok(()) + } + + /// Write-side tier gate: `CommandClass::Write` is `Block` under + /// `ReadOnly` (returns `Err` from [`enforce_node_tier_gate`] before this + /// function is even entered — see call sites), `Prompt` under + /// `Supervised`, and `Allow` under `Full`. Unlike [`Self::tier_gate_read`] + /// this ALWAYS calls [`gate_call_for_tier`] — never short-circuits on + /// `Allow` — because a `Full`-tier run can still have the *flow's own* + /// `require_approval: true` toggle set, and only `intercept_audited` + /// (reached via `gate_call_for_tier`) consults that. Mirrors + /// `OpenHumanCode::run`'s gating exactly. + async fn tier_gate_write(&self, op: &str, action: &Value) -> Result<()> { + let tool_name = format!("flows_memory_{op}"); + let tier_decision = enforce_node_tier_gate(&self.security, CommandClass::Write, op)?; + let summary = summarize_action(&tool_name, action); + let redacted = redact_args(action); + let (outcome, _audit_id) = + gate_call_for_tier(tier_decision, &tool_name, &summary, redacted).await; + if let GateOutcome::Deny { reason } = outcome { + tracing::warn!(target: "flows", op, "{LOG_PREFIX} write: approval gate denied"); + return Err(EngineError::Capability(reason)); + } + Ok(()) + } + + /// Shapes a batch of [`MemoryEntry`] hits into the node's `recall`/ + /// `search` output: `{ scope, query, results: [{ id, key, text, score, + /// namespace, category }] }`. Each entry's `text` passes through + /// [`wrap_untrusted_for_agent`] when [`is_potentially_untrusted`] flags + /// it — a connector-synced row (email, Slack, …) reaching downstream + /// `agent`/`condition` bindings must carry the same untrusted-source + /// marking a normal agent-prompt recall would give it. + fn shape_recall_result(scope: &str, query: &str, entries: &[MemoryEntry]) -> Value { + let results: Vec = entries + .iter() + .map(|entry| { + let text = if is_potentially_untrusted(entry) { + let hint = entry.namespace.as_deref().unwrap_or(scope); + wrap_untrusted_for_agent(&entry.content, hint) + } else { + entry.content.clone() + }; + json!({ + "id": entry.id, + "key": entry.key, + "text": text, + "score": entry.score, + "namespace": entry.namespace, + "category": entry.category.to_string(), + }) + }) + .collect(); + json!({ "scope": scope, "query": query, "results": results }) + } +} + +#[async_trait] +impl MemoryProvider for OpenHumanMemory { + /// Backs both `recall` and `search` (`opts.operation` distinguishes them + /// only for the `tracing::debug!` logs below — [`Self::shape_recall_result`] + /// returns `{ scope, query, results }` with no `operation` field, so the + /// two ops are otherwise indistinguishable in the response; both + /// currently route through the same [`Memory::recall`] call, as there is + /// no separate hybrid-search path reachable through the generic + /// `Arc` trait object this adapter holds). + async fn recall(&self, scope: &str, query: &str, opts: Value) -> Result { + let operation: &str = opts + .get("operation") + .and_then(Value::as_str) + .unwrap_or("recall"); + #[allow(clippy::cast_possible_truncation)] + let limit = opts + .get("limit") + .and_then(Value::as_u64) + .map_or(5, |v| v as usize); + let min_score = opts.get("min_score").and_then(Value::as_f64); + + tracing::debug!( + target: "flows", + operation, + scope, + query_chars = query.chars().count(), + limit, + ?min_score, + "{LOG_PREFIX} recall: entry" + ); + + self.tier_gate_read(operation)?; + + let entries = match scope { + "user" => { + let memory = self.memory().await?; + let recall_opts = RecallOpts { + namespace: Some(USER_NAMESPACE), + min_score, + ..RecallOpts::default() + }; + tracing::debug!( + target: "flows", + operation, + namespace = USER_NAMESPACE, + "{LOG_PREFIX} recall: querying user-scope namespace" + ); + memory + .recall(query, limit, recall_opts) + .await + .map_err(|e| { + EngineError::Capability(format!("memory node: recall failed: {e}")) + })? + } + "flow" => { + let namespace = self.flow_memory_namespace()?; + let memory = self.memory().await?; + let recall_opts = RecallOpts { + namespace: Some(namespace.as_str()), + min_score, + ..RecallOpts::default() + }; + tracing::debug!( + target: "flows", + operation, + namespace = %namespace, + "{LOG_PREFIX} recall: querying this flow's own namespace" + ); + memory + .recall(query, limit, recall_opts) + .await + .map_err(|e| { + EngineError::Capability(format!("memory node: recall failed: {e}")) + })? + } + "flows" => { + // Read-only, and — via `cross_flow_recall` — confined to the + // shared `flow_*` namespace prefix. This is intentionally + // NOT a broad cross-namespace scan: it can never reach the + // user's personal/global memory, only other flows' own + // automation output, exactly like `flow_memory_recall`'s + // `scope: "flows"` arm it delegates to. + let memory = self.memory().await?; + tracing::debug!( + target: "flows", + operation, + "{LOG_PREFIX} recall: querying cross-flow (flow_* prefix only)" + ); + cross_flow_recall(&memory, query, limit, min_score) + .await + .map_err(|e| { + EngineError::Capability(format!( + "memory node: cross-flow recall failed: {e}" + )) + })? + } + other => { + return Err(EngineError::Capability(format!( + "memory node: unknown scope \"{other}\" (expected \"user\", \"flow\", or \"flows\")" + ))); + } + }; + + tracing::debug!( + target: "flows", + operation, + scope, + hit_count = entries.len(), + "{LOG_PREFIX} recall: store returned" + ); + + Ok(Self::shape_recall_result(scope, query, &entries)) + } + + /// Delegates to [`lookup_flavour`] — the exact same flavoured-tree read + /// path `memory_flavour` (`MemoryFlavourTool`, #5175) uses. + async fn flavour(&self, slug: &str) -> Result { + tracing::debug!(target: "flows", flavour = slug, "{LOG_PREFIX} flavour: entry"); + self.tier_gate_read("flavour")?; + + match lookup_flavour(&self.config, slug) { + Err(hard) => { + tracing::debug!(target: "flows", flavour = slug, "{LOG_PREFIX} flavour: rejected (bad slug)"); + Err(EngineError::Capability(format!("memory node: {hard}"))) + } + Ok(FlavourLookup::Profile(body)) => { + tracing::debug!( + target: "flows", + flavour = slug, + profile_chars = body.chars().count(), + "{LOG_PREFIX} flavour: hit" + ); + Ok(json!({ "flavour": slug, "found": true, "profile": body })) + } + Ok(FlavourLookup::NotBuilt(message)) => { + tracing::debug!(target: "flows", flavour = slug, "{LOG_PREFIX} flavour: no profile built yet"); + Ok( + json!({ "flavour": slug, "found": false, "profile": Value::Null, "message": message }), + ) + } + Ok(FlavourLookup::Failed(message)) => { + tracing::warn!(target: "flows", flavour = slug, "{LOG_PREFIX} flavour: lookup failed"); + Err(EngineError::Capability(format!("memory node: {message}"))) + } + } + } + + /// Delegates to the same op the `people_list` agent tool calls + /// (`people::rpc::handle_list` over the current-context `PeopleStore`), + /// then narrows the ranked listing to entries whose display name, + /// primary email/phone, or any handle value contains `query` + /// (case-insensitive substring — this adapter has no access to a richer + /// query DSL than that). + async fn people(&self, query: Option<&str>) -> Result { + let query = query.map(str::trim).filter(|q| !q.is_empty()); + tracing::debug!(target: "flows", has_query = query.is_some(), "{LOG_PREFIX} people: entry"); + self.tier_gate_read("people")?; + + let store = crate::core::runtime::context::CoreContext::current() + .ok_or_else(|| { + EngineError::Capability( + "memory node: people store unavailable: core context not initialized" + .to_string(), + ) + })? + .people() + .map_err(|e| { + EngineError::Capability(format!("memory node: people store unavailable: {e}")) + })?; + + const DEFAULT_PEOPLE_LIMIT: usize = 100; + let outcome = crate::openhuman::people::rpc::handle_list(&store, DEFAULT_PEOPLE_LIMIT) + .await + .map_err(EngineError::Capability)?; + + let shaped = match query { + None => outcome.value, + Some(q) => filter_people_by_query(outcome.value, q), + }; + + tracing::debug!( + target: "flows", + result_count = shaped.get("people").and_then(serde_json::Value::as_array).map_or(0, Vec::len), + "{LOG_PREFIX} people: done" + ); + Ok(shaped) + } + + /// Writes `value` under `key` into this flow's own `flow_` + /// namespace — see the module doc's defense-in-depth note. Never + /// reachable for `scope != "flow"`, regardless of what the caller + /// passes: this is checked here independently of the engine's own + /// `validate_all` rejection of `scope: "user"` writes. + async fn remember(&self, scope: &str, key: &str, value: Value) -> Result<()> { + if scope != "flow" { + tracing::warn!( + target: "flows", + scope, + "{LOG_PREFIX} remember: REFUSED — scope must be \"flow\" (defense-in-depth; a \ + remember/forget node targeting scope \"user\" is already a hard reject in \ + tinyflows' own validate_all, but this adapter never trusts that alone)" + ); + return Err(EngineError::Capability(format!( + "memory node: remember only supports scope \"flow\" (got \"{scope}\") — writing \ + to the user's personal memory, or to the read-only \"flows\" scope, is never \ + permitted from a memory node" + ))); + } + let key = key.trim(); + if key.is_empty() { + return Err(EngineError::Capability( + "memory node: remember requires a non-empty key".to_string(), + )); + } + + // Secret check MUST run before the tier gate / HITL approval prompt + // below: `tier_gate_write` can park the run for human approval + // (`gate_call_for_tier`), and a likely-secret value must be refused + // up front rather than spend that approval round-trip on a write + // that was always going to be rejected (review fix — see #5227). + let content = value_to_content(&value); + if crate::openhuman::memory_store::safety::has_likely_secret(&content) { + tracing::warn!( + target: "flows", + key_chars = key.chars().count(), + content_chars = content.chars().count(), + "{LOG_PREFIX} remember: REFUSED — content looks like a secret" + ); + return Err(EngineError::Capability( + "memory node: refusing to store content that looks like a secret".to_string(), + )); + } + + let action = json!({ "operation": "remember", "scope": scope, "key": key }); + self.tier_gate_write("remember", &action).await?; + + let namespace = self.flow_memory_namespace()?; + let memory = self.memory().await?; + memory + .store_with_taint( + &namespace, + key, + &content, + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .map_err(|e| EngineError::Capability(format!("memory node: remember failed: {e}")))?; + + tracing::debug!( + target: "flows", + namespace = %namespace, + key_chars = key.chars().count(), + content_chars = content.chars().count(), + "{LOG_PREFIX} remember: stored" + ); + Ok(()) + } + + /// Deletes `key` from this flow's own `flow_` namespace — same + /// `scope`-lockdown as [`Self::remember`]. + async fn forget(&self, scope: &str, key: &str) -> Result<()> { + if scope != "flow" { + tracing::warn!( + target: "flows", + scope, + "{LOG_PREFIX} forget: REFUSED — scope must be \"flow\" (defense-in-depth)" + ); + return Err(EngineError::Capability(format!( + "memory node: forget only supports scope \"flow\" (got \"{scope}\") — the \ + user's personal memory and the read-only \"flows\" scope are never reachable \ + from a memory node" + ))); + } + let key = key.trim(); + if key.is_empty() { + return Err(EngineError::Capability( + "memory node: forget requires a non-empty key".to_string(), + )); + } + + let action = json!({ "operation": "forget", "scope": scope, "key": key }); + self.tier_gate_write("forget", &action).await?; + + let namespace = self.flow_memory_namespace()?; + let memory = self.memory().await?; + let removed = memory + .forget(&namespace, key) + .await + .map_err(|e| EngineError::Capability(format!("memory node: forget failed: {e}")))?; + + tracing::debug!( + target: "flows", + namespace = %namespace, + key_chars = key.chars().count(), + removed, + "{LOG_PREFIX} forget: done" + ); + Ok(()) + } +} + +/// Renders a `remember` node's `value` (arbitrary JSON — a plain string when +/// the author wrote a literal, or any resolved `=`-expression result +/// otherwise) into the `&str` content [`Memory::store_with_taint`] persists. +/// A JSON string is stored verbatim (not re-quoted); anything else is +/// serialized to its compact JSON form so structured values round-trip +/// losslessly through recall. +fn value_to_content(value: &Value) -> String { + match value { + Value::String(s) => s.clone(), + other => other.to_string(), + } +} + +/// Narrows a `{"people": [...]}` listing (as returned by +/// `people::rpc::handle_list`) to entries whose display name, primary email, +/// primary phone, or any handle value contains `query` — case-insensitive +/// substring match. Malformed/missing fields on an entry just don't match +/// (never panics on an unexpected shape). +fn filter_people_by_query(listing: Value, query: &str) -> Value { + let needle = query.to_ascii_lowercase(); + let people = listing + .get("people") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let matches = |field: Option<&str>| { + field + .map(|s| s.to_ascii_lowercase().contains(&needle)) + .unwrap_or(false) + }; + let filtered: Vec = people + .into_iter() + .filter(|person| { + matches(person.get("display_name").and_then(Value::as_str)) + || matches(person.get("primary_email").and_then(Value::as_str)) + || matches(person.get("primary_phone").and_then(Value::as_str)) + || person + .get("handles") + .and_then(Value::as_array) + .is_some_and(|handles| { + handles + .iter() + .any(|h| matches(h.get("value").and_then(Value::as_str))) + }) + }) + .collect(); + json!({ "people": filtered }) +} + +#[cfg(test)] +#[path = "memory_adapter_tests.rs"] +mod tests; diff --git a/src/openhuman/tinyflows/memory_adapter_tests.rs b/src/openhuman/tinyflows/memory_adapter_tests.rs new file mode 100644 index 000000000..13b1c7085 --- /dev/null +++ b/src/openhuman/tinyflows/memory_adapter_tests.rs @@ -0,0 +1,276 @@ +use super::*; +use crate::openhuman::security::{AutonomyLevel, POLICY_BLOCKED_MARKER}; +use tempfile::TempDir; + +fn security(autonomy: AutonomyLevel) -> Arc { + Arc::new(SecurityPolicy { + autonomy, + ..SecurityPolicy::default() + }) +} + +/// A `Config` rooted at a fresh tempdir — `lookup_flavour` (via +/// `flavour`) touches disk under `workspace_dir`, so every test needs its +/// own isolated root rather than sharing whatever `Config::default()`'s +/// workspace_dir happens to resolve to. The `TempDir` guard must outlive +/// the adapter, so callers keep it alive for the test's duration. +fn test_config() -> (TempDir, Arc) { + let tmp = TempDir::new().unwrap(); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + (tmp, Arc::new(cfg)) +} + +fn adapter(autonomy: AutonomyLevel) -> (TempDir, OpenHumanMemory) { + let (tmp, config) = test_config(); + ( + tmp, + OpenHumanMemory { + config, + security: security(autonomy), + }, + ) +} + +// ── scope lockdown (defense-in-depth) ──────────────────────────────── + +#[tokio::test] +async fn remember_rejects_user_scope() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter.remember("user", "k", json!("v")).await.unwrap_err(); + assert!(err.to_string().contains("only supports scope \"flow\"")); +} + +#[tokio::test] +async fn remember_rejects_flows_scope() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter + .remember("flows", "k", json!("v")) + .await + .unwrap_err(); + assert!(err.to_string().contains("only supports scope \"flow\"")); +} + +#[tokio::test] +async fn forget_rejects_user_scope() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter.forget("user", "k").await.unwrap_err(); + assert!(err.to_string().contains("only supports scope \"flow\"")); +} + +#[tokio::test] +async fn forget_rejects_flows_scope() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter.forget("flows", "k").await.unwrap_err(); + assert!(err.to_string().contains("only supports scope \"flow\"")); +} + +#[tokio::test] +async fn remember_rejects_empty_key_even_at_flow_scope() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter + .remember("flow", " ", json!("v")) + .await + .unwrap_err(); + assert!(err.to_string().contains("non-empty key")); +} + +// ── recall scope validation ─────────────────────────────────────────── + +#[tokio::test] +async fn recall_rejects_unknown_scope() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter + .recall("nonsense", "q", json!({"operation": "recall"})) + .await + .unwrap_err(); + assert!(err.to_string().contains("unknown scope")); +} + +// ── flow-id trust boundary ───────────────────────────────────────────── + +#[tokio::test] +async fn remember_flow_scope_without_trusted_origin_errs() { + // No `turn_origin::current()` scoped — not running inside a flow. + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter.remember("flow", "k", json!("v")).await.unwrap_err(); + assert!(err.to_string().contains("trusted Workflow-scoped origin")); +} + +// ── secret check runs BEFORE the HITL approval gate (P1 review fix) ── + +/// Regression for the review finding that `has_likely_secret` ran AFTER +/// `tier_gate_write` (which can park the run for human approval via +/// `gate_call_for_tier`) — a user could approve a write that then silently +/// failed the secret heuristic, wasting the approval round-trip. +/// +/// This is asserted indirectly but unambiguously: with NO trusted +/// `TrustedAutomation { Workflow }` origin scoped, `remember`'s later steps +/// (the tier-gate write summary is fine, but `flow_memory_namespace` — +/// reached only AFTER the tier gate — requires one and errors +/// `"trusted Workflow-scoped origin"` if missing, see +/// `remember_flow_scope_without_trusted_origin_errs` above). If the secret +/// check now runs strictly first, a secret-shaped value must fail with the +/// secret-rejection message instead of ever reaching that later trusted-origin +/// check — proving the reject happens before both the approval gate AND the +/// flow-id resolution that follows it. +#[tokio::test] +async fn remember_rejects_secret_before_reaching_trusted_origin_or_approval_gate() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter + .remember("flow", "k", json!("api_key=abc123")) + .await + .unwrap_err(); + assert!( + err.to_string().contains("looks like a secret"), + "expected the secret rejection to fire first, got: {err}" + ); + assert!( + !err.to_string().contains("trusted Workflow-scoped origin"), + "secret check must short-circuit before flow-id/approval resolution, got: {err}" + ); +} + +/// Same proof under `Supervised` autonomy, where `CommandClass::Write` is +/// `GateDecision::Prompt` and — with a real `ApprovalGate` installed — +/// `tier_gate_write` would park for human approval. The secret rejection +/// must still win the race: it never even calls into the tier gate. +#[tokio::test] +async fn remember_rejects_secret_under_supervised_autonomy_without_approval_round_trip() { + let (_tmp, adapter) = adapter(AutonomyLevel::Supervised); + let err = adapter + .remember("flow", "k", json!("Bearer abcdefghijklmnopqrstuvwxyz")) + .await + .unwrap_err(); + assert!( + err.to_string().contains("looks like a secret"), + "expected the secret rejection to fire before any approval prompt, got: {err}" + ); +} + +#[tokio::test] +async fn recall_flow_scope_without_trusted_origin_errs() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter + .recall("flow", "q", json!({"operation": "recall"})) + .await + .unwrap_err(); + assert!(err.to_string().contains("trusted Workflow-scoped origin")); +} + +// ── tier gate ─────────────────────────────────────────────────────────── + +#[tokio::test] +async fn recall_blocked_in_readonly_autonomy_never_touches_flow_id() { + // ReadOnly + CommandClass::Read is still Allow (Read is always + // allowed) — so this proves the tier gate runs and lets a genuine + // read through even under the most restrictive tier, distinct from + // the write path below. + let (_tmp, adapter) = adapter(AutonomyLevel::ReadOnly); + // No trusted origin, so this will fail on the flow-id lookup, not on + // the tier gate — proving the tier gate (Read => Allow) did not + // block it first. + let err = adapter + .recall("flow", "q", json!({"operation": "recall"})) + .await + .unwrap_err(); + assert!(err.to_string().contains("trusted Workflow-scoped origin")); +} + +#[tokio::test] +async fn remember_blocked_in_readonly_autonomy() { + let (_tmp, adapter) = adapter(AutonomyLevel::ReadOnly); + let err = adapter.remember("flow", "k", json!("v")).await.unwrap_err(); + assert!(err.to_string().contains(POLICY_BLOCKED_MARKER)); +} + +#[tokio::test] +async fn forget_blocked_in_readonly_autonomy() { + let (_tmp, adapter) = adapter(AutonomyLevel::ReadOnly); + let err = adapter.forget("flow", "k").await.unwrap_err(); + assert!(err.to_string().contains(POLICY_BLOCKED_MARKER)); +} + +// ── flavour / people delegate cleanly with no trusted origin required ── + +#[tokio::test] +async fn flavour_unknown_slug_errs() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let err = adapter.flavour("not-a-real-flavour").await.unwrap_err(); + assert!(err.to_string().contains("Unknown flavour")); +} + +#[tokio::test] +async fn flavour_valid_slug_with_no_tree_yet_reports_not_found() { + let (_tmp, adapter) = adapter(AutonomyLevel::Full); + let result = adapter.flavour("coding_style").await.unwrap(); + assert_eq!(result["found"], json!(false)); + assert_eq!(result["profile"], Value::Null); +} + +#[tokio::test] +async fn flavour_blocked_in_readonly_autonomy_still_reaches_lookup() { + // Read is Allow at every tier, so ReadOnly must behave identically + // to Full for a read-only operation like `flavour`. + let (_tmp, adapter) = adapter(AutonomyLevel::ReadOnly); + let result = adapter.flavour("coding_style").await.unwrap(); + assert_eq!(result["found"], json!(false)); +} + +// ── value_to_content / filter_people_by_query (pure helpers) ────────── + +#[test] +fn value_to_content_keeps_strings_verbatim() { + assert_eq!( + value_to_content(&json!("already published")), + "already published" + ); +} + +#[test] +fn value_to_content_serializes_non_strings() { + assert_eq!(value_to_content(&json!({"id": 42})), "{\"id\":42}"); + assert_eq!(value_to_content(&json!(42)), "42"); + assert_eq!(value_to_content(&json!(true)), "true"); +} + +#[test] +fn filter_people_by_query_matches_display_name_case_insensitively() { + let listing = json!({ + "people": [ + {"display_name": "Ada Lovelace", "primary_email": null, "primary_phone": null, "handles": []}, + {"display_name": "Grace Hopper", "primary_email": null, "primary_phone": null, "handles": []}, + ] + }); + let filtered = filter_people_by_query(listing, "ada"); + let people = filtered["people"].as_array().unwrap(); + assert_eq!(people.len(), 1); + assert_eq!(people[0]["display_name"], json!("Ada Lovelace")); +} + +#[test] +fn filter_people_by_query_matches_handle_values() { + let listing = json!({ + "people": [ + { + "display_name": "Someone", + "primary_email": null, + "primary_phone": null, + "handles": [{"kind": "email", "value": "someone@example.com"}] + }, + ] + }); + let filtered = filter_people_by_query(listing, "example.com"); + assert_eq!(filtered["people"].as_array().unwrap().len(), 1); +} + +#[test] +fn filter_people_by_query_no_match_returns_empty() { + let listing = json!({ + "people": [ + {"display_name": "Ada Lovelace", "primary_email": null, "primary_phone": null, "handles": []}, + ] + }); + let filtered = filter_people_by_query(listing, "zzz-no-match"); + assert!(filtered["people"].as_array().unwrap().is_empty()); +} diff --git a/src/openhuman/tinyflows/memory_node_e2e_tests.rs b/src/openhuman/tinyflows/memory_node_e2e_tests.rs new file mode 100644 index 000000000..c8a1aa4d3 --- /dev/null +++ b/src/openhuman/tinyflows/memory_node_e2e_tests.rs @@ -0,0 +1,402 @@ +//! End-to-end tests proving the `memory` node round-trips through the REAL +//! stack: the `tinyflows` engine executing a *compiled graph*, dispatching +//! through the real [`OpenHumanMemory`](super::memory_adapter::OpenHumanMemory) +//! host adapter — wired via [`build_capabilities`], NOT +//! `tinyflows::caps::mock::MockMemory` — against a real, on-disk `Memory` +//! store. +//! +//! **The gap this closes.** Two layers of unit coverage already exist: +//! - `memory_adapter_tests.rs` calls `OpenHumanMemory`'s methods directly +//! (scope lockdown, tier gate, trust boundary) but never through a compiled +//! graph or the engine, and never against the real store (every test there +//! uses a fresh empty `Config::workspace_dir` and only exercises the +//! error paths). +//! - `vendor/tinyflows/src/nodes/integration/memory.rs`'s own tests drive +//! `MemoryNode` through `compile`/`run`, but always against +//! `tinyflows::caps::mock::MockMemory` — the crate's own mock, not this +//! host's real adapter. +//! +//! Neither proves the actual wiring a saved flow run uses in production: +//! `flows::ops::flows_run` compiles a graph, calls +//! [`build_capabilities`] for a real `Capabilities` bundle, and hands both to +//! `tinyflows::engine::run`. This file reproduces exactly that path (see +//! [`trigger_to_memory`] / [`workflow_origin`]) so a regression in the +//! engine → `Capabilities::memory` → `OpenHumanMemory` → `Memory` store chain +//! fails a test, not just a live flow run. +//! +//! **Layer entered: full engine-run**, not a node-executor shortcut. Every +//! test below drives `tinyflows::compiler::compile` + `tinyflows::engine::run` +//! over a real [`build_capabilities`] bundle, under the same +//! `AgentTurnOrigin::TrustedAutomation { source: Workflow { .. }, .. }` +//! task-local scope `flows::ops::workflow_origin` scopes around a real +//! `flows_run` — the ONLY thing not reproduced is `flows::ops`'s own +//! run-bookkeeping (draft persistence, checkpointing, run-history rows), +//! which has nothing to do with whether the `memory` node's adapter wiring +//! works and is covered separately by `flows::ops_tests`. +//! +//! **Real store, not a stub.** `memory` here is the process-global +//! `MemoryClient` (`crate::openhuman::memory::global`), bound to the shared +//! temp workspace `memory::ops::test_support::ensure_shared_memory_client` +//! already uses for every other `memory::ops` real-store test — the SAME +//! on-disk `UnifiedMemory`-backed store `flows_run` writes to in production. +//! Serialized against sibling tests with `GLOBAL_MEMORY_TEST_LOCK`, exactly +//! like `memory::ops::documents`/`memory::ops::sync`/etc already do (see +//! [`lock_shared_memory`]). + +use std::sync::Arc; + +use serde_json::{json, Value}; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; + +use crate::openhuman::agent::turn_origin::{self, AgentTurnOrigin, TrustedAutomationSource}; +use crate::openhuman::config::Config; +use crate::openhuman::flows::flow_namespace; +use crate::openhuman::flows::memory_tools::FlowMemoryRecallTool; +use crate::openhuman::security::AutonomyLevel; +use crate::openhuman::tools::traits::Tool; + +use super::build_capabilities; + +// ── fixtures ──────────────────────────────────────────────────────────── + +/// Binds the process-global memory client to the shared test workspace and +/// holds the cross-test serialization lock for the caller's whole test body. +/// +/// One on-disk SQLite store is shared across every test thread in this +/// binary (`memory::global` is a process-global `OnceLock`), so concurrent +/// `init`/read/write from sibling tests races on schema init and can bleed +/// data across tests. `GLOBAL_MEMORY_TEST_LOCK` + `ensure_shared_memory_client` +/// is the crate's existing, proven pattern for this (see +/// `memory::ops::documents::tests::ensure_memory_client` and +/// `composio::ops_tests::init_memory_client`) — reused verbatim here rather +/// than inventing a third variant. +async fn lock_shared_memory() -> tokio::sync::MutexGuard<'static, ()> { + let guard = crate::openhuman::memory::ops::GLOBAL_MEMORY_TEST_LOCK + .lock() + .await; + crate::openhuman::memory::ops::ensure_shared_memory_client(); + guard +} + +/// A `Config` rooted at its own fresh tempdir, with autonomy raised to +/// `Full`. +/// +/// `remember`/`forget` are `CommandClass::Write` in +/// `OpenHumanMemory::tier_gate_write`, which needs a non-`Block` tier +/// decision to clear `enforce_node_tier_gate` at all. `Full` also keeps the +/// write from depending on whether some earlier test in this shared binary +/// has installed a global `ApprovalGate` — see [`workflow_origin`]'s doc for +/// why `require_approval: false` makes this safe either way. +fn full_autonomy_config() -> (tempfile::TempDir, Arc) { + let tmp = tempfile::TempDir::new().unwrap(); + let mut cfg = Config { + workspace_dir: tmp.path().join("workspace"), + action_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Config::default() + }; + cfg.autonomy.level = AutonomyLevel::Full; + (tmp, Arc::new(cfg)) +} + +/// A trusted, saved-flow run origin for `flow_id` — the ONLY source +/// `OpenHumanMemory::trusted_flow_id` (and `flow_memory_recall`/ +/// `flow_memory_remember`'s own `trusted_flow_id`) accepts for +/// `scope: "flow"`/`"flows"`. Mirrors what `flows::ops::flows_run` scopes +/// around a real run. +/// +/// `require_approval: false` matches a saved flow's default. It also means +/// this test's write can never park on approval even if some other test in +/// this shared binary has installed a global `ApprovalGate`: +/// `ApprovalGate::intercept_audited` allows a +/// `Workflow { require_approval: false }` origin unconditionally (see +/// `caps::gate_call_for_tier`'s doc comment) — so this test's outcome does +/// not depend on test execution order. +fn workflow_origin(flow_id: &str) -> AgentTurnOrigin { + AgentTurnOrigin::TrustedAutomation { + job_id: flow_id.to_string(), + source: TrustedAutomationSource::Workflow { + require_approval: false, + }, + } +} + +fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: Vec::new(), + position: None, + } +} + +fn edge(from: &str, to: &str) -> Edge { + Edge { + from_node: from.to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +/// A minimal `trigger -> memory` graph — the same two-node shape +/// `vendor/tinyflows/src/nodes/integration/memory.rs`'s own tests use, kept +/// identical here so any behavioral difference between running against +/// `MockMemory` and running against the real `OpenHumanMemory` adapter is +/// attributable to the adapter, not to graph-shape differences. +fn trigger_to_memory(config: Value) -> WorkflowGraph { + WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger, Value::Null), + node("mem", NodeKind::Memory, config), + ], + edges: vec![edge("t", "mem")], + ..Default::default() + } +} + +fn unique_flow_id(prefix: &str) -> String { + format!("{prefix}-{}", uuid::Uuid::new_v4()) +} + +// ── 1 & 2. flow-scope round-trip through the real engine, plus coherence +// with the sibling `flow_memory_recall` agent tool ───────────────────────── + +#[tokio::test] +async fn memory_node_remember_then_recall_round_trips_through_the_real_engine_and_adapter() { + let _serial = lock_shared_memory().await; + let (_tmp, config) = full_autonomy_config(); + let flow_id = unique_flow_id("e2e-roundtrip"); + let caps = build_capabilities(config, format!("flow:{flow_id}")); + + // ── remember: a real graph, compiled by the real validator/compiler and + // run through the real engine, dispatching through the real + // OpenHumanMemory adapter (build_capabilities' `memory` slot). ── + let remember_graph = trigger_to_memory(json!({ + "operation": "remember", + "scope": "flow", + "key": "item-42", + "value": "Digest already sent item-42 to subscribers" + })); + let compiled_remember = + tinyflows::compiler::compile(&remember_graph).expect("compile remember graph"); + + let remember_outcome = turn_origin::with_origin( + workflow_origin(&flow_id), + tinyflows::engine::run(&compiled_remember, Value::Null, &caps), + ) + .await + .expect("remember run should complete against the real store"); + + assert!( + remember_outcome.pending_approvals.is_empty(), + "Full autonomy + require_approval:false must never park this write" + ); + assert_eq!( + remember_outcome.output["nodes"]["mem"]["items"][0]["json"]["json"]["ok"], + json!(true), + "unexpected remember output shape: {}", + remember_outcome.output + ); + assert_eq!( + remember_outcome.output["nodes"]["mem"]["items"][0]["json"]["json"]["key"], + json!("item-42") + ); + + // ── recall: a SECOND compiled graph, a SECOND engine run, same flow_id + // — proving the write from the first run is durably visible via the real + // on-disk store, not merely echoed back within one call. ── + let recall_graph = trigger_to_memory(json!({ + "operation": "recall", + "scope": "flow", + "query": "item-42" + })); + let compiled_recall = + tinyflows::compiler::compile(&recall_graph).expect("compile recall graph"); + + let recall_outcome = turn_origin::with_origin( + workflow_origin(&flow_id), + tinyflows::engine::run(&compiled_recall, Value::Null, &caps), + ) + .await + .expect("recall run should complete against the real store"); + + assert!(recall_outcome.pending_approvals.is_empty()); + let results = recall_outcome.output["nodes"]["mem"]["items"][0]["json"]["json"]["results"] + .as_array() + .cloned() + .unwrap_or_default(); + assert!( + results + .iter() + .any(|hit| hit["text"].as_str().is_some_and(|t| t.contains("item-42"))), + "expected the remembered content back from the real flow_ namespace, got: {results:?}" + ); + + // ── coherence (#5176): the SAME write is visible through the sibling + // `flow_memory_recall` agent tool for the same flow_id — proving one + // shared store, not two namespace conventions that happen to overlap by + // convention (see memory_adapter.rs's module doc). ── + let memory = crate::openhuman::memory::global::client_if_ready() + .expect("global memory client must be initialized by lock_shared_memory") + .memory_handle(); + let recall_tool = FlowMemoryRecallTool::new(memory); + let tool_result = turn_origin::with_origin( + workflow_origin(&flow_id), + recall_tool.execute(json!({ "query": "item-42", "flow_id": flow_id })), + ) + .await + .expect("flow_memory_recall tool call should not error at the harness level"); + assert!( + !tool_result.is_error, + "flow_memory_recall reported an error: {}", + tool_result.output() + ); + assert!( + tool_result.output().contains("item-42"), + "expected the memory-node's write to be visible via flow_memory_recall \ + (same underlying store), got: {}", + tool_result.output() + ); +} + +// ── 3. security invariant end-to-end: scope:"user" writes are rejected, +// and the user's real memory store is never touched ──────────────────────── + +#[tokio::test] +async fn memory_node_remember_user_scope_is_rejected_and_never_touches_user_memory() { + let _serial = lock_shared_memory().await; + let (_tmp, config) = full_autonomy_config(); + let flow_id = unique_flow_id("e2e-security"); + let caps = build_capabilities(config, format!("flow:{flow_id}")); + + // A unique key so this assertion can't collide with real content any + // other test in this shared workspace may have written under + // GLOBAL_NAMESPACE. + let forbidden_key = format!("forbidden-{}", uuid::Uuid::new_v4()); + + // ── (a) validate-time rejection: tinyflows' own structural validator + // rejects a `remember`/`scope: "user"` node BEFORE compile ever + // succeeds, so a graph shaped this way can never even reach a run. ── + let user_scope_graph = trigger_to_memory(json!({ + "operation": "remember", + "scope": "user", + "key": forbidden_key, + "value": "must never be written to user memory" + })); + let compile_err = tinyflows::compiler::compile(&user_scope_graph) + .expect_err("scope: \"user\" on a remember node must be rejected at validate/compile time"); + assert!( + compile_err.to_string().to_lowercase().contains("user"), + "expected the validator's scope:\"user\" rejection, got: {compile_err}" + ); + + // ── (b) defense-in-depth: even bypassing tinyflows' validator entirely + // and calling straight through to the adapter build_capabilities wired + // (the exact instance a real run would dispatch to), OpenHumanMemory's + // own remember() hard-refuses anything but scope: "flow". ── + let direct_err = turn_origin::with_origin( + workflow_origin(&flow_id), + caps.memory + .as_ref() + .expect("build_capabilities must wire a memory capability") + .remember("user", &forbidden_key, json!("must never be written")), + ) + .await + .expect_err("the adapter itself must independently refuse scope: \"user\""); + assert!(direct_err + .to_string() + .contains("only supports scope \"flow\"")); + + // ── (c) the user's real, durable GLOBAL_NAMESPACE store is untouched by + // either attempt above. ── + let memory = crate::openhuman::memory::global::client_if_ready() + .expect("global memory client must be initialized by lock_shared_memory") + .memory_handle(); + let entry = memory + .get(tinycortex::memory::GLOBAL_NAMESPACE, &forbidden_key) + .await + .expect("get should not error"); + assert!( + entry.is_none(), + "the memory node must never write to the user's GLOBAL_NAMESPACE store, found: {entry:?}" + ); +} + +// ── 4. dry_run_workflow still works with a memory node: MockMemory returns +// shaped data without ever touching the real store ───────────────────────── + +#[tokio::test] +async fn memory_node_dry_run_uses_mock_memory_and_never_touches_the_real_store() { + let _serial = lock_shared_memory().await; + let flow_id = unique_flow_id("e2e-dryrun"); + + let memory = crate::openhuman::memory::global::client_if_ready() + .expect("global memory client must be initialized by lock_shared_memory") + .memory_handle(); + + // Nothing under this flow's namespace exists yet. + assert!(memory + .get(&flow_namespace(&flow_id), "item-42") + .await + .unwrap() + .is_none()); + + // The SAME round-trip graph shape as the real-adapter test above, but + // run against `tinyflows::caps::mock::mock_capabilities()` — exactly + // what `DryRunWorkflowTool::execute` wires (`Capabilities::memory` + // defaults to `MockMemory` there; see the crate's own doc comment on + // `mock_capabilities`). + let mock_caps = tinyflows::caps::mock::mock_capabilities(); + + let remember_graph = trigger_to_memory(json!({ + "operation": "remember", + "scope": "flow", + "key": "item-42", + "value": "should never reach the real store in a dry run" + })); + let compiled_remember = + tinyflows::compiler::compile(&remember_graph).expect("compile remember graph"); + let remember_outcome = tinyflows::engine::run(&compiled_remember, Value::Null, &mock_caps) + .await + .expect("dry-run remember should succeed against MockMemory"); + assert_eq!( + remember_outcome.output["nodes"]["mem"]["items"][0]["json"]["json"]["ok"], + json!(true) + ); + + // The real store never saw this write — MockMemory::remember is a no-op. + assert!( + memory + .get(&flow_namespace(&flow_id), "item-42") + .await + .unwrap() + .is_none(), + "dry_run_workflow's MockMemory must never touch the real on-disk store" + ); + + // recall through the same mock returns MockMemory's fixed shaped echo — + // proving a graph containing a `memory` node still dry-runs cleanly end + // to end, without ever reaching the real adapter or store. + let recall_graph = trigger_to_memory(json!({ + "operation": "recall", + "scope": "flow", + "query": "item-42" + })); + let compiled_recall = + tinyflows::compiler::compile(&recall_graph).expect("compile recall graph"); + let recall_outcome = tinyflows::engine::run(&compiled_recall, Value::Null, &mock_caps) + .await + .expect("dry-run recall should succeed against MockMemory"); + let results = recall_outcome.output["nodes"]["mem"]["items"][0]["json"]["json"]["results"] + .as_array() + .cloned() + .unwrap_or_default(); + assert!( + results.iter().any(|hit| hit["id"] == json!("mem_1")), + "expected MockMemory's fixed shaped echo (not real-store content), got: {results:?}" + ); +} diff --git a/src/openhuman/tinyflows/mod.rs b/src/openhuman/tinyflows/mod.rs index 327652127..cd1a1d8ba 100644 --- a/src/openhuman/tinyflows/mod.rs +++ b/src/openhuman/tinyflows/mod.rs @@ -2,10 +2,13 @@ //! (an external, host-agnostic crate — validate → compile → run on //! `tinyagents`) to real OpenHuman services. //! -//! This module is export-focused. The five capability adapters plus the two -//! run entry points — [`build_capabilities`] and [`open_flow_checkpointer`], -//! re-exported below — live in [`caps`]; run observability logging lives in -//! [`observability`]; post-run Langfuse export of a run's durable graph +//! This module is export-focused. Six of the seven capability adapters plus +//! the two run entry points — [`build_capabilities`] and +//! [`open_flow_checkpointer`], re-exported below — live in [`caps`]; the +//! `memory` node's `MemoryProvider` adapter (`OpenHumanMemory`) lives in its +//! own [`memory_adapter`] module, per this repo's ~500-line file-size +//! convention (`caps.rs` is already large); run observability logging lives +//! in [`observability`]; post-run Langfuse export of a run's durable graph //! observations lives in [`langfuse_export`]. The `flows::` domain //! (`src/openhuman/flows/ops.rs`) calls [`build_capabilities`] / //! [`open_flow_checkpointer`] to drive a run and @@ -13,8 +16,15 @@ pub mod caps; pub mod langfuse_export; +pub mod memory_adapter; +/// End-to-end coverage for the `memory` node through the REAL engine + real +/// `OpenHumanMemory` adapter + real store — see the module doc there for why +/// this lives apart from `tests.rs`'s general capability-seam smoke tests. +#[cfg(test)] +mod memory_node_e2e_tests; pub mod observability; #[cfg(test)] mod tests; pub use caps::{build_capabilities, open_flow_checkpointer}; +pub use memory_adapter::OpenHumanMemory; diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 47957f01b..2352363b6 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -367,7 +367,7 @@ pub fn all_tools_with_runtime( // toolkits a flow still needs (Phase 5, item 19). Read-only. #[cfg(feature = "flows")] Box::new(ListConnectableToolkitsTool::new(config.clone())), - // Queryable DSL schema (F2): enumerate the 12 node kinds and fetch one + // Queryable DSL schema (F2): enumerate the 13 node kinds and fetch one // kind's full config-field/port/example/gotcha contract — the DSL // analogue of search_tool_catalog + get_tool_contract, so an agent need // not rely on prompt prose or memory for node config shapes. Read-only. diff --git a/vendor/tinyflows b/vendor/tinyflows index fb24363ae..ff9950186 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit fb24363aea921f957958bc8f4aeb5b0a244e41c7 +Subproject commit ff9950186ac66775a554c80aa420920b4f39e4de