diff --git a/src/openhuman/agent/harness/memory_protocol.rs b/src/openhuman/agent/harness/memory_protocol.rs new file mode 100644 index 0000000000..c130695ee1 --- /dev/null +++ b/src/openhuman/agent/harness/memory_protocol.rs @@ -0,0 +1,381 @@ +//! 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`, 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 +//! (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 call into a [`MemoryOp`], keyed by name and — for the two +/// multi-purpose tools — the arguments. +/// +/// 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 — 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 (the split-out ingest tool). + "memory_store" | "memory_forget" | "memory_tree_ingest_document" => MemoryOp::Write, + // 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_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 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 + /// 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::*; + 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("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("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_tree", &json!({ "mode": "search_entities" })), + MemoryOp::IndexRead + ); + + // 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", &no_args()), + Default::default() + ); + + 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", &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", &no_args()); + 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", &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", &no_args()); + 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", &no_args()); + t.observe_tool("memory_store", &no_args()); + // Intervening non-memory tool calls don't clear the obligation. + t.observe_tool("send_message", &no_args()); + 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", &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", &no_args()); + 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", &no_args()).needs_guidance()); + assert!(!t.observe_tool("send_message", &no_args()).needs_guidance()); + assert!(t + .observe_tool("memory_recall", &no_args()) + .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..e00eaa91e7 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,145 @@ 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, + /// 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 { + pub fn new() -> Self { + Self { + tracker: std::sync::Mutex::new( + crate::openhuman::agent::harness::memory_protocol::MemoryProtocolTracker::new(), + ), + pending_ops: std::sync::Mutex::new(std::collections::HashMap::new()), + } + } +} + +impl Default for MemoryProtocolMiddleware { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl Middleware<()> for MemoryProtocolMiddleware { + fn name(&self) -> &str { + "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() { + return Ok(()); + } + let observation = { + let mut tracker = match self.tracker.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + tracker.observe(op) + }; + 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 +1824,159 @@ 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; + + /// 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 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: {}", + 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 read = run_cycle(&mw, "memory_recall", json!({}), "no dupes", None).await; + assert!( + !read.content.contains(MEMORY_PROTOCOL_MARKER), + "a read is not annotated" + ); + + 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. + assert!(!write + .content + .contains("without first reading the memory index")); + + 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 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)); + 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(); + 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 second = run_cycle(&mw, "memory_store", json!({}), "b", None).await; + 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..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))); @@ -389,6 +393,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