From 1243ede87016aea575d3f51566a517150dc2d249 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 1 Jul 2026 23:39:39 +0530 Subject: [PATCH 1/4] Enforce the memory read-dedupe-write-update-index protocol (#4116) Agents were told to follow a read-index -> dedupe -> write -> update-index cycle around durable memory but it was never enforced, so writes landed without a dedupe read (duplicating entries) and update_memory_md was skipped (leaving MEMORY.md drifting from the store). Add a MemoryProtocolTracker state machine and drive it from a tinyagents MemoryProtocolMiddleware: on each successful memory op it advances the cycle and, on a write, appends a corrective note to the tool result the model reads (flagging a missing dedupe read or a drifting index), then warns at run end when a write was never followed by update_memory_md. Same structured-correction-surfaced-to-the-model pattern as the unknown-tool recovery. Registered outermost so its guidance survives the after-tool byte cap. --- .../agent/harness/memory_protocol.rs | 293 ++++++++++++++++++ src/openhuman/agent/harness/mod.rs | 1 + src/openhuman/tinyagents/middleware.rs | 178 ++++++++++- src/openhuman/tinyagents/mod.rs | 10 + 4 files changed, 481 insertions(+), 1 deletion(-) create mode 100644 src/openhuman/agent/harness/memory_protocol.rs diff --git a/src/openhuman/agent/harness/memory_protocol.rs b/src/openhuman/agent/harness/memory_protocol.rs new file mode 100644 index 0000000000..c0f2bcbc79 --- /dev/null +++ b/src/openhuman/agent/harness/memory_protocol.rs @@ -0,0 +1,293 @@ +//! Memory-protocol enforcement state machine (issue #4116). +//! +//! Agents are instructed to follow a **read-index → dedupe → write → +//! update-index** cycle when they mutate durable memory: +//! +//! 1. Read the memory index (`memory_recall` / a `memory_tree_*` query, or +//! equivalently the `MEMORY.md` index) to check for near-duplicates *before* +//! creating an entry. +//! 2. Write the entry (`memory_store`, `memory_forget`, `memory_tree_ingest_document`). +//! 3. Call `update_memory_md` afterward so the `MEMORY.md` index stays in sync +//! with the underlying store. +//! +//! The protocol was previously described to the model but never enforced, so it +//! was followed inconsistently — agents wrote entries without a dedupe read +//! (creating duplicates) and skipped `update_memory_md` (so the index drifted). +//! +//! This module is the pure, side-effect-free state machine that observes the +//! sequence of memory tool calls in a session and reports two violations: +//! +//! - **missing index read** — a write not preceded by an index read this cycle. +//! - **index drift** — a write that was never followed by `update_memory_md` +//! (detected at the next write, and at run end via [`MemoryProtocolTracker::pending_index_update`]). +//! +//! The tinyagents seam ([`crate::openhuman::tinyagents::middleware`]) drives this +//! tracker from its `after_tool` / `after_agent` hooks and surfaces the guidance +//! back to the model as a corrective note appended to the tool result — the same +//! "structured correction surfaced to the model" pattern used for unknown-tool +//! recovery (#4118). Keeping the logic here (free of tinyagents types) lets the +//! read → dedupe → write → update-index contract be unit-tested directly. + +/// Marker prefixed to every corrective note so downstream code (and tests) can +/// recognise memory-protocol guidance in a tool result. +pub const MEMORY_PROTOCOL_MARKER: &str = "[memory-protocol]"; + +/// Classification of a tool call for the memory protocol. Everything the model +/// can call is one of these; non-memory tools are [`MemoryOp::Other`] and never +/// affect protocol state. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MemoryOp { + /// Reading the memory index to check for duplicates (satisfies the "read + /// index / dedupe" step of the cycle). + IndexRead, + /// A durable memory mutation that should be preceded by a dedupe read and + /// followed by an index update. + Write, + /// Writing the `MEMORY.md` index back into sync (the closing step). + IndexUpdate, + /// Any tool that is not part of the memory protocol. + Other, +} + +/// Classify a tool by name into a [`MemoryOp`]. +/// +/// Name-based (the classifier runs before arguments matter): the memory tool +/// surface is a small, stable set of `memory_*` / `update_memory_md` names. +pub fn classify_memory_op(tool_name: &str) -> MemoryOp { + match tool_name { + // The index-sync step. + "update_memory_md" => MemoryOp::IndexUpdate, + // Durable mutations: create an entry, delete an entry, or ingest a + // document into the memory tree. + "memory_store" | "memory_forget" | "memory_tree_ingest_document" => MemoryOp::Write, + // Dedupe reads: recall/search over stored memory, or any read-only walk + // of the memory tree. These let the agent check for near-duplicates + // before writing. + "memory_recall" + | "memory_search" + | "memory_tree" + | "memory_tree_query_source" + | "memory_tree_search_entities" + | "memory_tree_fetch_leaves" + | "memory_tree_drill_down" + | "memory_tree_cover_window" => MemoryOp::IndexRead, + _ => MemoryOp::Other, + } +} + +/// What a single observed memory op means for the protocol. Returned by +/// [`MemoryProtocolTracker::observe`] so the caller can decide whether — and +/// what — to surface back to the model. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct MemoryProtocolObservation { + /// The observed op was a durable memory write. + pub was_write: bool, + /// A write happened without a dedupe/index read earlier in this cycle. + pub missing_index_read: bool, + /// A write happened while a previous write was still awaiting + /// `update_memory_md` — the index is drifting from the store. + pub index_drift: bool, +} + +impl MemoryProtocolObservation { + /// Whether this observation warrants a corrective note back to the model. + /// Every write gets one (the forward "call `update_memory_md`" reminder is + /// itself the enforcement of the closing step); reads and other ops don't. + pub fn needs_guidance(&self) -> bool { + self.was_write + } + + /// Render the corrective note appended to the tool result, or `None` when no + /// guidance is warranted. The wording escalates with the violations detected. + pub fn guidance(&self, tool_name: &str) -> Option { + if !self.needs_guidance() { + return None; + } + let mut parts: Vec = Vec::new(); + if self.missing_index_read { + parts.push(format!( + "`{tool_name}` wrote to memory without first reading the memory index to check for \ + duplicates. Before creating entries, recall existing memory (e.g. `memory_recall`) \ + so you don't store a near-duplicate." + )); + } + if self.index_drift { + parts.push( + "A previous memory write was never followed by `update_memory_md`, so the MEMORY.md \ + index is drifting from stored memory. Reconcile it now." + .to_string(), + ); + } + // The always-on closing-step reminder: keep the index in sync. + parts.push( + "After mutating memory, call `update_memory_md` to keep the MEMORY.md index in sync." + .to_string(), + ); + Some(format!("{MEMORY_PROTOCOL_MARKER} {}", parts.join(" "))) + } +} + +/// Per-session tracker of the memory-protocol cycle. One instance lives for the +/// duration of a turn/run (held by the tinyagents middleware) and observes the +/// ordered sequence of *successful* memory tool calls. +/// +/// The cycle is `read → write → update-index`, and it repeats: an +/// `update_memory_md` closes a cycle and arms the next one (so the following +/// write again expects a fresh dedupe read). +#[derive(Debug, Default)] +pub struct MemoryProtocolTracker { + /// A dedupe/index read has occurred since the last cycle reset. + saw_index_read: bool, + /// A write has occurred that has not yet been followed by `update_memory_md`. + pending_index_update: bool, +} + +impl MemoryProtocolTracker { + /// Fresh tracker with no observed ops. + pub fn new() -> Self { + Self::default() + } + + /// Observe one **successful** memory op and advance the state machine, + /// returning what it means for the protocol. Callers must only pass ops that + /// actually succeeded — a failed `memory_store` neither creates an entry nor + /// obliges an index update. + pub fn observe(&mut self, op: MemoryOp) -> MemoryProtocolObservation { + match op { + MemoryOp::IndexRead => { + self.saw_index_read = true; + MemoryProtocolObservation::default() + } + MemoryOp::Write => { + let obs = MemoryProtocolObservation { + was_write: true, + missing_index_read: !self.saw_index_read, + index_drift: self.pending_index_update, + }; + self.pending_index_update = true; + obs + } + MemoryOp::IndexUpdate => { + // The index is back in sync; arm the next cycle so its write + // expects a fresh dedupe read. + self.pending_index_update = false; + self.saw_index_read = false; + MemoryProtocolObservation::default() + } + MemoryOp::Other => MemoryProtocolObservation::default(), + } + } + + /// Classify `tool_name` and [`observe`](Self::observe) it in one step. + pub fn observe_tool(&mut self, tool_name: &str) -> MemoryProtocolObservation { + self.observe(classify_memory_op(tool_name)) + } + + /// Whether a memory write is still awaiting `update_memory_md`. Checked at + /// run end to detect a write that was never followed by an index update. + pub fn pending_index_update(&self) -> bool { + self.pending_index_update + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn classifies_the_memory_tool_surface() { + assert_eq!(classify_memory_op("update_memory_md"), MemoryOp::IndexUpdate); + assert_eq!(classify_memory_op("memory_store"), MemoryOp::Write); + assert_eq!(classify_memory_op("memory_forget"), MemoryOp::Write); + assert_eq!( + classify_memory_op("memory_tree_ingest_document"), + MemoryOp::Write + ); + assert_eq!(classify_memory_op("memory_recall"), MemoryOp::IndexRead); + assert_eq!(classify_memory_op("memory_search"), MemoryOp::IndexRead); + assert_eq!( + classify_memory_op("memory_tree_search_entities"), + MemoryOp::IndexRead + ); + assert_eq!(classify_memory_op("send_message"), MemoryOp::Other); + assert_eq!(classify_memory_op("file_write"), MemoryOp::Other); + } + + #[test] + fn full_cycle_reports_no_violation() { + let mut t = MemoryProtocolTracker::new(); + assert_eq!(t.observe_tool("memory_recall"), Default::default()); + + let write = t.observe_tool("memory_store"); + assert!(write.was_write); + assert!(!write.missing_index_read, "read preceded the write"); + assert!(!write.index_drift); + assert!(t.pending_index_update()); + + // Closing the cycle clears the pending index update. + assert_eq!(t.observe_tool("update_memory_md"), Default::default()); + assert!(!t.pending_index_update()); + } + + #[test] + fn write_without_index_read_is_flagged() { + let mut t = MemoryProtocolTracker::new(); + let obs = t.observe_tool("memory_store"); + assert!(obs.was_write); + assert!(obs.missing_index_read, "no dedupe read preceded the write"); + assert!(obs.needs_guidance()); + let note = obs.guidance("memory_store").expect("guidance for a write"); + assert!(note.starts_with(MEMORY_PROTOCOL_MARKER)); + assert!(note.contains("without first reading the memory index")); + assert!(note.contains("update_memory_md")); + } + + #[test] + fn write_not_followed_by_update_is_detected_at_next_write() { + let mut t = MemoryProtocolTracker::new(); + t.observe_tool("memory_recall"); + let first = t.observe_tool("memory_store"); + assert!(!first.index_drift); + assert!(t.pending_index_update()); + + // A second write with no intervening update_memory_md: the index is + // drifting from the store. + let second = t.observe_tool("memory_store"); + assert!(second.index_drift, "prior write never synced the index"); + let note = second.guidance("memory_store").unwrap(); + assert!(note.contains("drifting")); + } + + #[test] + fn pending_index_update_survives_until_update_at_run_end() { + let mut t = MemoryProtocolTracker::new(); + t.observe_tool("memory_recall"); + t.observe_tool("memory_store"); + // Intervening non-memory tool calls don't clear the obligation. + t.observe_tool("send_message"); + assert!( + t.pending_index_update(), + "index update still owed at run end" + ); + } + + #[test] + fn update_arms_a_fresh_cycle_that_expects_a_new_read() { + let mut t = MemoryProtocolTracker::new(); + t.observe_tool("memory_recall"); + t.observe_tool("memory_store"); + t.observe_tool("update_memory_md"); + + // Next cycle: a write with no fresh read is flagged again. + let obs = t.observe_tool("memory_store"); + assert!(obs.missing_index_read, "each cycle needs its own dedupe read"); + } + + #[test] + fn reads_and_other_ops_need_no_guidance() { + let mut t = MemoryProtocolTracker::new(); + assert!(!t.observe_tool("memory_recall").needs_guidance()); + assert!(!t.observe_tool("send_message").needs_guidance()); + assert!(t.observe_tool("memory_recall").guidance("memory_recall").is_none()); + } +} diff --git a/src/openhuman/agent/harness/mod.rs b/src/openhuman/agent/harness/mod.rs index db9ac0b6fa..5d700fadcd 100644 --- a/src/openhuman/agent/harness/mod.rs +++ b/src/openhuman/agent/harness/mod.rs @@ -32,6 +32,7 @@ pub(crate) mod graph; mod instructions; pub(crate) mod memory_context; pub(crate) mod memory_context_safety; +pub(crate) mod memory_protocol; pub(crate) mod parse; pub(crate) mod payload_summarizer; pub mod run_queue; diff --git a/src/openhuman/tinyagents/middleware.rs b/src/openhuman/tinyagents/middleware.rs index a6ea26b9dd..1a491ec384 100644 --- a/src/openhuman/tinyagents/middleware.rs +++ b/src/openhuman/tinyagents/middleware.rs @@ -29,7 +29,7 @@ use tinyagents::error::Result as TaResult; use tinyagents::harness::context::RunContext; use tinyagents::harness::message::{ContentBlock, Message as TaMessage}; use tinyagents::harness::middleware::{ - Middleware, MiddlewareToolOutcome, ToolHandler, ToolMiddleware, + AgentRun, Middleware, MiddlewareToolOutcome, ToolHandler, ToolMiddleware, }; use tinyagents::harness::model::ModelRequest; use tinyagents::harness::runtime::AgentHarness; @@ -990,6 +990,102 @@ impl Middleware<()> for ArgRecoveryMiddleware { } } +/// `after_tool` + `after_agent`: enforce the memory protocol (issue #4116). +/// +/// Agents are told to follow a **read-index → dedupe → write → update-index** +/// cycle around durable memory, but the contract was never enforced, so it was +/// followed inconsistently: writes landed without a dedupe read (duplicating +/// entries) and `update_memory_md` was skipped (so `MEMORY.md` drifted from the +/// store). This middleware observes the ordered sequence of *successful* memory +/// tool calls via [`MemoryProtocolTracker`] and, on each memory write, appends a +/// corrective note to the tool result so the model is nudged back onto the +/// protocol — the same "structured correction surfaced to the model" pattern the +/// unknown-tool recovery (#4118) uses. At run end it warns when a write was never +/// followed by an index update (the index is left stale). +/// +/// Only *successful* ops advance the state machine — a failed `memory_store` +/// neither creates an entry nor obliges an index update. Non-memory tools are +/// ignored, so this is a no-op on turns that never touch memory. +pub struct MemoryProtocolMiddleware { + tracker: std::sync::Mutex, +} + +impl MemoryProtocolMiddleware { + pub fn new() -> Self { + Self { + tracker: std::sync::Mutex::new( + crate::openhuman::agent::harness::memory_protocol::MemoryProtocolTracker::new(), + ), + } + } +} + +impl Default for MemoryProtocolMiddleware { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl Middleware<()> for MemoryProtocolMiddleware { + fn name(&self) -> &str { + "memory_protocol" + } + + async fn after_tool( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + result: &mut TaToolResult, + ) -> TaResult<()> { + // Only successful memory ops advance the protocol — a failed write did + // not mutate memory and must not demand an index update. + if result.error.is_some() { + return Ok(()); + } + let observation = { + let mut tracker = match self.tracker.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + tracker.observe_tool(&result.name) + }; + if let Some(note) = observation.guidance(&result.name) { + tracing::debug!( + tool = result.name.as_str(), + missing_index_read = observation.missing_index_read, + index_drift = observation.index_drift, + "[tinyagents::mw] memory-protocol guidance appended to tool result" + ); + if !result.content.is_empty() { + result.content.push_str("\n\n"); + } + result.content.push_str(¬e); + } + Ok(()) + } + + async fn after_agent( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + _run: &mut AgentRun, + ) -> TaResult<()> { + let pending = self + .tracker + .lock() + .map(|tracker| tracker.pending_index_update()) + .unwrap_or(false); + if pending { + tracing::warn!( + "[tinyagents::mw] memory-protocol: run ended with a memory write that was never \ + followed by update_memory_md — the MEMORY.md index is left stale" + ); + } + Ok(()) + } +} + /// `before_model`: enforce OpenHuman's daily/monthly cost budgets **before** a /// model call spends (issue #4249, Phase 5). Reads the global /// [`CostTracker`](crate::openhuman::cost) and, when cost budgets are configured @@ -1685,4 +1781,84 @@ mod tests { // Unknown tool defaults to no external effect (nothing to gate). assert!(!mw.has_external_effect("missing", &json!({}))); } + + // ── MemoryProtocolMiddleware (issue #4116) ────────────────────────────── + + use crate::openhuman::agent::harness::memory_protocol::MEMORY_PROTOCOL_MARKER; + + #[tokio::test] + async fn memory_write_without_index_read_gets_a_corrective_note() { + let mw = MemoryProtocolMiddleware::new(); + let mut result = tool_result("memory_store", "stored entry 42"); + mw.after_tool(&mut ctx(), &(), &mut result).await.unwrap(); + assert!( + result.content.contains(MEMORY_PROTOCOL_MARKER), + "a write with no preceding dedupe read should be annotated: {}", + result.content + ); + assert!(result.content.contains("without first reading the memory index")); + assert!(result.content.contains("update_memory_md")); + // The original tool output is preserved, guidance is appended. + assert!(result.content.starts_with("stored entry 42")); + } + + #[tokio::test] + async fn full_cycle_read_then_write_then_update_only_reminds_on_the_write() { + let mw = MemoryProtocolMiddleware::new(); + + let mut read = tool_result("memory_recall", "no near-duplicates found"); + mw.after_tool(&mut ctx(), &(), &mut read).await.unwrap(); + assert!( + !read.content.contains(MEMORY_PROTOCOL_MARKER), + "a read is not annotated" + ); + + let mut write = tool_result("memory_store", "stored"); + mw.after_tool(&mut ctx(), &(), &mut write).await.unwrap(); + assert!(write.content.contains(MEMORY_PROTOCOL_MARKER)); + // The read preceded the write, so no missing-read complaint — just the + // forward "sync the index" reminder. + assert!(!write.content.contains("without first reading the memory index")); + + let mut update = tool_result("update_memory_md", "index updated"); + mw.after_tool(&mut ctx(), &(), &mut update).await.unwrap(); + assert!( + !update.content.contains(MEMORY_PROTOCOL_MARKER), + "closing the cycle needs no guidance" + ); + } + + #[tokio::test] + async fn failed_memory_write_does_not_advance_the_protocol() { + let mw = MemoryProtocolMiddleware::new(); + let mut failed = failing_result("memory_store", "disk full"); + mw.after_tool(&mut ctx(), &(), &mut failed).await.unwrap(); + // A failed write is not annotated and leaves nothing pending, so a later + // run-end sweep must not warn about a stale index. + assert!(!failed.content.contains(MEMORY_PROTOCOL_MARKER)); + let mut run = AgentRun::new(); + // after_agent is a no-op warn path; it must not error. + mw.after_agent(&mut ctx(), &(), &mut run).await.unwrap(); + } + + #[tokio::test] + async fn second_write_without_an_update_flags_index_drift() { + let mw = MemoryProtocolMiddleware::new(); + for _ in 0..1 { + let mut read = tool_result("memory_recall", "checked"); + mw.after_tool(&mut ctx(), &(), &mut read).await.unwrap(); + } + let mut first = tool_result("memory_store", "a"); + mw.after_tool(&mut ctx(), &(), &mut first).await.unwrap(); + assert!(!first.content.contains("drifting")); + + // No update_memory_md between the two writes → the index is drifting. + let mut second = tool_result("memory_store", "b"); + mw.after_tool(&mut ctx(), &(), &mut second).await.unwrap(); + assert!( + second.content.contains("drifting"), + "a second unsynced write should flag index drift: {}", + second.content + ); + } } diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index a41937ff8d..2a68d657f8 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -389,6 +389,16 @@ pub async fn run_turn_via_tinyagents_shared( // `before_model` hooks run cache-align → microcompact → compress → trim. // Capture the autocompaction opt-out before `install` consumes `context_mw`. let autocompact_enabled = context_mw.autocompact_enabled; + + // Memory-protocol enforcement (issue #4116): observe the read → dedupe → + // write → update-index cycle across this run's memory tool calls and surface + // a corrective note back to the model on a write that skipped the dedupe read + // or left the MEMORY.md index unsynced. Registered FIRST (outermost) so its + // `after_tool` runs LAST — after the byte-cap (`ToolOutputMiddleware`, pushed + // by `install` below) truncates — and its guidance survives into the tool + // result the model reads. A no-op on turns that never touch memory. + harness.push_middleware(Arc::new(middleware::MemoryProtocolMiddleware::new())); + context_mw.install(&mut harness, &tool_sets); // Pre-call cost budget gate (issue #4249, Phase 5): fail before a model call From 8a0f0f14d7c3967afffa52acb8bbd4eabf29dba5 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 1 Jul 2026 23:48:51 +0530 Subject: [PATCH 2/4] Apply rustfmt to memory-protocol enforcement --- src/openhuman/agent/harness/memory_protocol.rs | 15 ++++++++++++--- src/openhuman/tinyagents/middleware.rs | 11 ++++++++--- 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/src/openhuman/agent/harness/memory_protocol.rs b/src/openhuman/agent/harness/memory_protocol.rs index c0f2bcbc79..07de1d5237 100644 --- a/src/openhuman/agent/harness/memory_protocol.rs +++ b/src/openhuman/agent/harness/memory_protocol.rs @@ -196,7 +196,10 @@ mod tests { #[test] fn classifies_the_memory_tool_surface() { - assert_eq!(classify_memory_op("update_memory_md"), MemoryOp::IndexUpdate); + assert_eq!( + classify_memory_op("update_memory_md"), + MemoryOp::IndexUpdate + ); assert_eq!(classify_memory_op("memory_store"), MemoryOp::Write); assert_eq!(classify_memory_op("memory_forget"), MemoryOp::Write); assert_eq!( @@ -280,7 +283,10 @@ mod tests { // Next cycle: a write with no fresh read is flagged again. let obs = t.observe_tool("memory_store"); - assert!(obs.missing_index_read, "each cycle needs its own dedupe read"); + assert!( + obs.missing_index_read, + "each cycle needs its own dedupe read" + ); } #[test] @@ -288,6 +294,9 @@ mod tests { let mut t = MemoryProtocolTracker::new(); assert!(!t.observe_tool("memory_recall").needs_guidance()); assert!(!t.observe_tool("send_message").needs_guidance()); - assert!(t.observe_tool("memory_recall").guidance("memory_recall").is_none()); + assert!(t + .observe_tool("memory_recall") + .guidance("memory_recall") + .is_none()); } } diff --git a/src/openhuman/tinyagents/middleware.rs b/src/openhuman/tinyagents/middleware.rs index 1a491ec384..f0366f9952 100644 --- a/src/openhuman/tinyagents/middleware.rs +++ b/src/openhuman/tinyagents/middleware.rs @@ -1007,7 +1007,8 @@ impl Middleware<()> for ArgRecoveryMiddleware { /// neither creates an entry nor obliges an index update. Non-memory tools are /// ignored, so this is a no-op on turns that never touch memory. pub struct MemoryProtocolMiddleware { - tracker: std::sync::Mutex, + tracker: + std::sync::Mutex, } impl MemoryProtocolMiddleware { @@ -1796,7 +1797,9 @@ mod tests { "a write with no preceding dedupe read should be annotated: {}", result.content ); - assert!(result.content.contains("without first reading the memory index")); + assert!(result + .content + .contains("without first reading the memory index")); assert!(result.content.contains("update_memory_md")); // The original tool output is preserved, guidance is appended. assert!(result.content.starts_with("stored entry 42")); @@ -1818,7 +1821,9 @@ mod tests { assert!(write.content.contains(MEMORY_PROTOCOL_MARKER)); // The read preceded the write, so no missing-read complaint — just the // forward "sync the index" reminder. - assert!(!write.content.contains("without first reading the memory index")); + assert!(!write + .content + .contains("without first reading the memory index")); let mut update = tool_result("update_memory_md", "index updated"); mw.after_tool(&mut ctx(), &(), &mut update).await.unwrap(); From 92789f2e6ecff73582b2a5d2f95b1084ff995662 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 1 Jul 2026 23:49:25 +0530 Subject: [PATCH 3/4] Simplify drift test: drop single-iteration loop --- src/openhuman/tinyagents/middleware.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/openhuman/tinyagents/middleware.rs b/src/openhuman/tinyagents/middleware.rs index f0366f9952..fa8ece47ef 100644 --- a/src/openhuman/tinyagents/middleware.rs +++ b/src/openhuman/tinyagents/middleware.rs @@ -1849,10 +1849,8 @@ mod tests { #[tokio::test] async fn second_write_without_an_update_flags_index_drift() { let mw = MemoryProtocolMiddleware::new(); - for _ in 0..1 { - let mut read = tool_result("memory_recall", "checked"); - mw.after_tool(&mut ctx(), &(), &mut read).await.unwrap(); - } + let mut read = tool_result("memory_recall", "checked"); + mw.after_tool(&mut ctx(), &(), &mut read).await.unwrap(); let mut first = tool_result("memory_store", "a"); mw.after_tool(&mut ctx(), &(), &mut first).await.unwrap(); assert!(!first.content.contains("drifting")); From e4a2ee8d99d39513beed6c0f3ad59abccaf403c2 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Thu, 2 Jul 2026 00:07:04 +0530 Subject: [PATCH 4/4] Address review: arg-aware memory-op classification + wire both entrypoints - Classify memory_tree(mode=ingest_document) as a Write and update_memory_md only as an index-update when file=MEMORY.md (a SKILL.md edit no longer masks a stale MEMORY.md index). Arguments are captured in the middleware's before_tool hook (keyed by call id) since the tool result carries none. - Install MemoryProtocolMiddleware on the run_turn_via_tinyagents entrypoint too, so both harness-build paths enforce the cycle consistently. - Extend unit + middleware tests for the SKILL.md and memory_tree-ingest cases. --- .../agent/harness/memory_protocol.rs | 167 +++++++++++++----- src/openhuman/tinyagents/middleware.rs | 149 ++++++++++++++-- src/openhuman/tinyagents/mod.rs | 4 + 3 files changed, 259 insertions(+), 61 deletions(-) diff --git a/src/openhuman/agent/harness/memory_protocol.rs b/src/openhuman/agent/harness/memory_protocol.rs index 07de1d5237..c130695ee1 100644 --- a/src/openhuman/agent/harness/memory_protocol.rs +++ b/src/openhuman/agent/harness/memory_protocol.rs @@ -3,12 +3,13 @@ //! Agents are instructed to follow a **read-index → dedupe → write → //! update-index** cycle when they mutate durable memory: //! -//! 1. Read the memory index (`memory_recall` / a `memory_tree_*` query, or +//! 1. Read the memory index (`memory_recall` / a `memory_tree` query, or //! equivalently the `MEMORY.md` index) to check for near-duplicates *before* //! creating an entry. -//! 2. Write the entry (`memory_store`, `memory_forget`, `memory_tree_ingest_document`). -//! 3. Call `update_memory_md` afterward so the `MEMORY.md` index stays in sync -//! with the underlying store. +//! 2. Write the entry (`memory_store`, `memory_forget`, or a document ingest via +//! `memory_tree_ingest_document` / `memory_tree` with `mode: "ingest_document"`). +//! 3. Call `update_memory_md` (targeting `MEMORY.md`) afterward so the index +//! stays in sync with the underlying store. //! //! The protocol was previously described to the model but never enforced, so it //! was followed inconsistently — agents wrote entries without a dedupe read @@ -49,23 +50,44 @@ pub enum MemoryOp { Other, } -/// Classify a tool by name into a [`MemoryOp`]. +/// Classify a tool call into a [`MemoryOp`], keyed by name and — for the two +/// multi-purpose tools — the arguments. /// -/// Name-based (the classifier runs before arguments matter): the memory tool -/// surface is a small, stable set of `memory_*` / `update_memory_md` names. -pub fn classify_memory_op(tool_name: &str) -> MemoryOp { +/// Two tools are polymorphic and cannot be classified by name alone: +/// - `update_memory_md` edits either `MEMORY.md` **or** `SKILL.md` +/// (`src/openhuman/tools/impl/filesystem/update_memory_md.rs`); only a +/// `MEMORY.md` edit reconciles the memory index, so a `SKILL.md` edit is not +/// an [`MemoryOp::IndexUpdate`] and must not close the cycle. +/// - the consolidated `memory_tree` tool (`src/openhuman/memory/query/mod.rs`) +/// is a read in every `mode` except `ingest_document`, which writes a document +/// into the tree — a durable mutation. +/// +/// The arguments are captured at `before_tool` time and correlated to the result +/// by call id (the tool result itself carries no arguments). +pub fn classify_memory_op(tool_name: &str, arguments: &serde_json::Value) -> MemoryOp { + let arg_str = |key: &str| arguments.get(key).and_then(|v| v.as_str()); match tool_name { - // The index-sync step. - "update_memory_md" => MemoryOp::IndexUpdate, + // The index-sync step — but only for the MEMORY.md index. The same tool + // can edit SKILL.md, which does not reconcile the memory index and so is + // a no-op for this protocol. + "update_memory_md" => match arg_str("file") { + Some("MEMORY.md") => MemoryOp::IndexUpdate, + _ => MemoryOp::Other, + }, // Durable mutations: create an entry, delete an entry, or ingest a - // document into the memory tree. + // document into the memory tree (the split-out ingest tool). "memory_store" | "memory_forget" | "memory_tree_ingest_document" => MemoryOp::Write, - // Dedupe reads: recall/search over stored memory, or any read-only walk - // of the memory tree. These let the agent check for near-duplicates - // before writing. + // Consolidated memory_tree tool: `ingest_document` writes; every other + // mode is a read-only retrieval. + "memory_tree" => match arg_str("mode") { + Some("ingest_document") => MemoryOp::Write, + _ => MemoryOp::IndexRead, + }, + // Dedupe reads: recall/search over stored memory, or a read-only walk of + // the memory tree. These let the agent check for near-duplicates before + // writing. "memory_recall" | "memory_search" - | "memory_tree" | "memory_tree_query_source" | "memory_tree_search_entities" | "memory_tree_fetch_leaves" @@ -178,9 +200,14 @@ impl MemoryProtocolTracker { } } - /// Classify `tool_name` and [`observe`](Self::observe) it in one step. - pub fn observe_tool(&mut self, tool_name: &str) -> MemoryProtocolObservation { - self.observe(classify_memory_op(tool_name)) + /// Classify a tool call (name + arguments) and [`observe`](Self::observe) it + /// in one step. + pub fn observe_tool( + &mut self, + tool_name: &str, + arguments: &serde_json::Value, + ) -> MemoryProtocolObservation { + self.observe(classify_memory_op(tool_name, arguments)) } /// Whether a memory write is still awaiting `update_memory_md`. Checked at @@ -193,49 +220,101 @@ impl MemoryProtocolTracker { #[cfg(test)] mod tests { use super::*; + use serde_json::json; + + /// Empty arguments — for the name-only tools where arguments are irrelevant. + fn no_args() -> serde_json::Value { + json!({}) + } #[test] fn classifies_the_memory_tool_surface() { + let a = no_args(); + assert_eq!(classify_memory_op("memory_store", &a), MemoryOp::Write); + assert_eq!(classify_memory_op("memory_forget", &a), MemoryOp::Write); + assert_eq!( + classify_memory_op("memory_tree_ingest_document", &a), + MemoryOp::Write + ); + assert_eq!(classify_memory_op("memory_recall", &a), MemoryOp::IndexRead); + assert_eq!(classify_memory_op("memory_search", &a), MemoryOp::IndexRead); assert_eq!( - classify_memory_op("update_memory_md"), + classify_memory_op("memory_tree_search_entities", &a), + MemoryOp::IndexRead + ); + assert_eq!(classify_memory_op("send_message", &a), MemoryOp::Other); + assert_eq!(classify_memory_op("file_write", &a), MemoryOp::Other); + } + + #[test] + fn update_memory_md_only_closes_the_cycle_for_the_memory_index() { + // A MEMORY.md edit reconciles the index; a SKILL.md edit does not and so + // must not close the cycle or clear a pending write. + assert_eq!( + classify_memory_op("update_memory_md", &json!({ "file": "MEMORY.md" })), MemoryOp::IndexUpdate ); - assert_eq!(classify_memory_op("memory_store"), MemoryOp::Write); - assert_eq!(classify_memory_op("memory_forget"), MemoryOp::Write); assert_eq!( - classify_memory_op("memory_tree_ingest_document"), + classify_memory_op("update_memory_md", &json!({ "file": "SKILL.md" })), + MemoryOp::Other + ); + + // A SKILL.md update after a memory write leaves the index still owed. + let mut t = MemoryProtocolTracker::new(); + t.observe_tool("memory_recall", &no_args()); + t.observe_tool("memory_store", &no_args()); + t.observe_tool("update_memory_md", &json!({ "file": "SKILL.md" })); + assert!( + t.pending_index_update(), + "a SKILL.md edit must not mask the stale MEMORY.md index" + ); + } + + #[test] + fn consolidated_memory_tree_ingest_is_a_write() { + // Every mode is a read except `ingest_document`, which writes. + assert_eq!( + classify_memory_op("memory_tree", &json!({ "mode": "ingest_document" })), MemoryOp::Write ); - assert_eq!(classify_memory_op("memory_recall"), MemoryOp::IndexRead); - assert_eq!(classify_memory_op("memory_search"), MemoryOp::IndexRead); assert_eq!( - classify_memory_op("memory_tree_search_entities"), + classify_memory_op("memory_tree", &json!({ "mode": "search_entities" })), MemoryOp::IndexRead ); - assert_eq!(classify_memory_op("send_message"), MemoryOp::Other); - assert_eq!(classify_memory_op("file_write"), MemoryOp::Other); + + // An ingest via the consolidated tool obliges an index update. + let mut t = MemoryProtocolTracker::new(); + let obs = t.observe_tool("memory_tree", &json!({ "mode": "ingest_document" })); + assert!(obs.was_write, "ingest_document mode is a durable write"); + assert!(t.pending_index_update()); } #[test] fn full_cycle_reports_no_violation() { let mut t = MemoryProtocolTracker::new(); - assert_eq!(t.observe_tool("memory_recall"), Default::default()); + assert_eq!( + t.observe_tool("memory_recall", &no_args()), + Default::default() + ); - let write = t.observe_tool("memory_store"); + let write = t.observe_tool("memory_store", &no_args()); assert!(write.was_write); assert!(!write.missing_index_read, "read preceded the write"); assert!(!write.index_drift); assert!(t.pending_index_update()); // Closing the cycle clears the pending index update. - assert_eq!(t.observe_tool("update_memory_md"), Default::default()); + assert_eq!( + t.observe_tool("update_memory_md", &json!({ "file": "MEMORY.md" })), + Default::default() + ); assert!(!t.pending_index_update()); } #[test] fn write_without_index_read_is_flagged() { let mut t = MemoryProtocolTracker::new(); - let obs = t.observe_tool("memory_store"); + let obs = t.observe_tool("memory_store", &no_args()); assert!(obs.was_write); assert!(obs.missing_index_read, "no dedupe read preceded the write"); assert!(obs.needs_guidance()); @@ -248,14 +327,14 @@ mod tests { #[test] fn write_not_followed_by_update_is_detected_at_next_write() { let mut t = MemoryProtocolTracker::new(); - t.observe_tool("memory_recall"); - let first = t.observe_tool("memory_store"); + t.observe_tool("memory_recall", &no_args()); + let first = t.observe_tool("memory_store", &no_args()); assert!(!first.index_drift); assert!(t.pending_index_update()); // A second write with no intervening update_memory_md: the index is // drifting from the store. - let second = t.observe_tool("memory_store"); + let second = t.observe_tool("memory_store", &no_args()); assert!(second.index_drift, "prior write never synced the index"); let note = second.guidance("memory_store").unwrap(); assert!(note.contains("drifting")); @@ -264,10 +343,10 @@ mod tests { #[test] fn pending_index_update_survives_until_update_at_run_end() { let mut t = MemoryProtocolTracker::new(); - t.observe_tool("memory_recall"); - t.observe_tool("memory_store"); + t.observe_tool("memory_recall", &no_args()); + t.observe_tool("memory_store", &no_args()); // Intervening non-memory tool calls don't clear the obligation. - t.observe_tool("send_message"); + t.observe_tool("send_message", &no_args()); assert!( t.pending_index_update(), "index update still owed at run end" @@ -277,12 +356,12 @@ mod tests { #[test] fn update_arms_a_fresh_cycle_that_expects_a_new_read() { let mut t = MemoryProtocolTracker::new(); - t.observe_tool("memory_recall"); - t.observe_tool("memory_store"); - t.observe_tool("update_memory_md"); + t.observe_tool("memory_recall", &no_args()); + t.observe_tool("memory_store", &no_args()); + t.observe_tool("update_memory_md", &json!({ "file": "MEMORY.md" })); // Next cycle: a write with no fresh read is flagged again. - let obs = t.observe_tool("memory_store"); + let obs = t.observe_tool("memory_store", &no_args()); assert!( obs.missing_index_read, "each cycle needs its own dedupe read" @@ -292,10 +371,10 @@ mod tests { #[test] fn reads_and_other_ops_need_no_guidance() { let mut t = MemoryProtocolTracker::new(); - assert!(!t.observe_tool("memory_recall").needs_guidance()); - assert!(!t.observe_tool("send_message").needs_guidance()); + assert!(!t.observe_tool("memory_recall", &no_args()).needs_guidance()); + assert!(!t.observe_tool("send_message", &no_args()).needs_guidance()); assert!(t - .observe_tool("memory_recall") + .observe_tool("memory_recall", &no_args()) .guidance("memory_recall") .is_none()); } diff --git a/src/openhuman/tinyagents/middleware.rs b/src/openhuman/tinyagents/middleware.rs index fa8ece47ef..e00eaa91e7 100644 --- a/src/openhuman/tinyagents/middleware.rs +++ b/src/openhuman/tinyagents/middleware.rs @@ -1009,6 +1009,16 @@ impl Middleware<()> for ArgRecoveryMiddleware { pub struct MemoryProtocolMiddleware { tracker: std::sync::Mutex, + /// call_id → classified op, captured in `before_tool` (the tool result carries + /// no arguments, yet `update_memory_md` and `memory_tree` can only be + /// classified from their `file` / `mode` argument). Correlated back by + /// `result.call_id` in `after_tool`. + pending_ops: std::sync::Mutex< + std::collections::HashMap< + String, + crate::openhuman::agent::harness::memory_protocol::MemoryOp, + >, + >, } impl MemoryProtocolMiddleware { @@ -1017,6 +1027,7 @@ impl MemoryProtocolMiddleware { tracker: std::sync::Mutex::new( crate::openhuman::agent::harness::memory_protocol::MemoryProtocolTracker::new(), ), + pending_ops: std::sync::Mutex::new(std::collections::HashMap::new()), } } } @@ -1033,12 +1044,43 @@ impl Middleware<()> for MemoryProtocolMiddleware { "memory_protocol" } + async fn before_tool( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + call: &mut TaToolCall, + ) -> TaResult<()> { + // Classify with the arguments in hand (the result won't carry them) and + // stash the op keyed by call id. Only memory-relevant ops are stored, so + // the map stays empty on turns that never touch memory. + let op = crate::openhuman::agent::harness::memory_protocol::classify_memory_op( + &call.name, + &call.arguments, + ); + if op != crate::openhuman::agent::harness::memory_protocol::MemoryOp::Other { + if let Ok(mut ops) = self.pending_ops.lock() { + ops.insert(call.id.clone(), op); + } + } + Ok(()) + } + async fn after_tool( &self, _ctx: &mut RunContext<()>, _state: &(), result: &mut TaToolResult, ) -> TaResult<()> { + // Consume the op captured for this call (removing it so the map can't + // grow unbounded). Absent → a non-memory tool: nothing to enforce. + let op = self + .pending_ops + .lock() + .ok() + .and_then(|mut ops| ops.remove(&result.call_id)); + let Some(op) = op else { + return Ok(()); + }; // Only successful memory ops advance the protocol — a failed write did // not mutate memory and must not demand an index update. if result.error.is_some() { @@ -1049,7 +1091,7 @@ impl Middleware<()> for MemoryProtocolMiddleware { Ok(guard) => guard, Err(poisoned) => poisoned.into_inner(), }; - tracker.observe_tool(&result.name) + tracker.observe(op) }; if let Some(note) = observation.guidance(&result.name) { tracing::debug!( @@ -1787,11 +1829,32 @@ mod tests { use crate::openhuman::agent::harness::memory_protocol::MEMORY_PROTOCOL_MARKER; + /// Drive one full tool cycle through the middleware: `before_tool` (captures + /// the arguments the result won't carry) then `after_tool`, correlated by a + /// shared call id. Returns the (possibly annotated) result. + async fn run_cycle( + mw: &MemoryProtocolMiddleware, + name: &str, + args: serde_json::Value, + content: &str, + error: Option<&str>, + ) -> TaToolResult { + let mut call = TaToolCall { + id: "c1".into(), + name: name.into(), + arguments: args, + }; + mw.before_tool(&mut ctx(), &(), &mut call).await.unwrap(); + let mut result = tool_result(name, content); // call_id "c1" matches + result.error = error.map(|e| e.to_string()); + mw.after_tool(&mut ctx(), &(), &mut result).await.unwrap(); + result + } + #[tokio::test] async fn memory_write_without_index_read_gets_a_corrective_note() { let mw = MemoryProtocolMiddleware::new(); - let mut result = tool_result("memory_store", "stored entry 42"); - mw.after_tool(&mut ctx(), &(), &mut result).await.unwrap(); + let result = run_cycle(&mw, "memory_store", json!({}), "stored entry 42", None).await; assert!( result.content.contains(MEMORY_PROTOCOL_MARKER), "a write with no preceding dedupe read should be annotated: {}", @@ -1809,15 +1872,13 @@ mod tests { async fn full_cycle_read_then_write_then_update_only_reminds_on_the_write() { let mw = MemoryProtocolMiddleware::new(); - let mut read = tool_result("memory_recall", "no near-duplicates found"); - mw.after_tool(&mut ctx(), &(), &mut read).await.unwrap(); + let read = run_cycle(&mw, "memory_recall", json!({}), "no dupes", None).await; assert!( !read.content.contains(MEMORY_PROTOCOL_MARKER), "a read is not annotated" ); - let mut write = tool_result("memory_store", "stored"); - mw.after_tool(&mut ctx(), &(), &mut write).await.unwrap(); + let write = run_cycle(&mw, "memory_store", json!({}), "stored", None).await; assert!(write.content.contains(MEMORY_PROTOCOL_MARKER)); // The read preceded the write, so no missing-read complaint — just the // forward "sync the index" reminder. @@ -1825,19 +1886,76 @@ mod tests { .content .contains("without first reading the memory index")); - let mut update = tool_result("update_memory_md", "index updated"); - mw.after_tool(&mut ctx(), &(), &mut update).await.unwrap(); + let update = run_cycle( + &mw, + "update_memory_md", + json!({ "file": "MEMORY.md" }), + "index updated", + None, + ) + .await; assert!( !update.content.contains(MEMORY_PROTOCOL_MARKER), "closing the cycle needs no guidance" ); } + #[tokio::test] + async fn skill_md_update_does_not_close_the_memory_cycle() { + let mw = MemoryProtocolMiddleware::new(); + run_cycle(&mw, "memory_recall", json!({}), "checked", None).await; + run_cycle(&mw, "memory_store", json!({}), "stored", None).await; + // update_memory_md targeting SKILL.md must NOT reconcile the MEMORY.md + // index, so the stale-index warning is still owed at run end. + run_cycle( + &mw, + "update_memory_md", + json!({ "file": "SKILL.md" }), + "skill updated", + None, + ) + .await; + let mut run = AgentRun::new(); + // Still pending → after_agent takes its warn path without erroring. + mw.after_agent(&mut ctx(), &(), &mut run).await.unwrap(); + // A following write reports drift, proving pending was not cleared. + let next = run_cycle(&mw, "memory_store", json!({}), "again", None).await; + assert!( + next.content.contains("drifting"), + "SKILL.md update must not mask the stale MEMORY.md index: {}", + next.content + ); + } + + #[tokio::test] + async fn consolidated_memory_tree_ingest_is_treated_as_a_write() { + let mw = MemoryProtocolMiddleware::new(); + let ingest = run_cycle( + &mw, + "memory_tree", + json!({ "mode": "ingest_document" }), + "ingested", + None, + ) + .await; + assert!( + ingest.content.contains(MEMORY_PROTOCOL_MARKER), + "memory_tree ingest_document is a write and must be annotated: {}", + ingest.content + ); + } + #[tokio::test] async fn failed_memory_write_does_not_advance_the_protocol() { let mw = MemoryProtocolMiddleware::new(); - let mut failed = failing_result("memory_store", "disk full"); - mw.after_tool(&mut ctx(), &(), &mut failed).await.unwrap(); + let failed = run_cycle( + &mw, + "memory_store", + json!({}), + "disk full", + Some("disk full"), + ) + .await; // A failed write is not annotated and leaves nothing pending, so a later // run-end sweep must not warn about a stale index. assert!(!failed.content.contains(MEMORY_PROTOCOL_MARKER)); @@ -1849,15 +1967,12 @@ mod tests { #[tokio::test] async fn second_write_without_an_update_flags_index_drift() { let mw = MemoryProtocolMiddleware::new(); - let mut read = tool_result("memory_recall", "checked"); - mw.after_tool(&mut ctx(), &(), &mut read).await.unwrap(); - let mut first = tool_result("memory_store", "a"); - mw.after_tool(&mut ctx(), &(), &mut first).await.unwrap(); + run_cycle(&mw, "memory_recall", json!({}), "checked", None).await; + let first = run_cycle(&mw, "memory_store", json!({}), "a", None).await; assert!(!first.content.contains("drifting")); // No update_memory_md between the two writes → the index is drifting. - let mut second = tool_result("memory_store", "b"); - mw.after_tool(&mut ctx(), &(), &mut second).await.unwrap(); + let second = run_cycle(&mw, "memory_store", json!({}), "b", None).await; assert!( second.content.contains("drifting"), "a second unsynced write should flag index drift: {}", diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index 2a68d657f8..eb4f8a30b4 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -230,6 +230,10 @@ pub async fn run_turn_via_tinyagents( harness .register_model(model, Arc::new(provider_model)) .set_default_model(model); + // Memory-protocol enforcement (issue #4116) — installed on this entrypoint + // too so both harness-build paths enforce the read → dedupe → write → + // update-index cycle consistently. No-op unless memory tools are called. + harness.push_middleware(Arc::new(middleware::MemoryProtocolMiddleware::new())); let tool_count = resolved_tools.len(); for tool in resolved_tools { harness.register_tool(Arc::new(ToolAdapter::new(tool)));