feat(agent-harness): memory read→dedupe→write→update-index protocol — rework of #4388 (#4444)

src/openhuman/agent/harness/memory_protocol.rs

@@ -0,0 +1,386 @@
+//! 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_vector_search"
+        | "memory_chunk_context"
+        | "memory_hybrid_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<String> {
+        if !self.needs_guidance() {
+            return None;
+        }
+        let mut parts: Vec<String> = 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_vector_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());
+    }
+}

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 mod run_queue;
 pub mod sandbox_context;

src/openhuman/tinyagents/middleware.rs

@@ -30,8 +30,8 @@ use tinyagents::harness::context::RunContext;
 use tinyagents::harness::events::AgentEvent;
 use tinyagents::harness::message::{ContentBlock, Message as TaMessage};
 use tinyagents::harness::middleware::{
-    ContextualToolSelectionMiddleware, Middleware, MiddlewareToolOutcome, ToolAllowlistMiddleware,
-    ToolHandler, ToolMiddleware,
+    AgentRun, ContextualToolSelectionMiddleware, Middleware, MiddlewareToolOutcome,
+    ToolAllowlistMiddleware, ToolHandler, ToolMiddleware,
 };
 use tinyagents::harness::model::{ModelRequest, PromptSegment, SegmentRole};
 use tinyagents::harness::runtime::AgentHarness;
@@ -1440,6 +1440,143 @@ impl Middleware<()> for ArgRecoveryMiddleware {
     }
 }
 
+/// 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<crate::openhuman::agent::harness::memory_protocol::MemoryProtocolTracker>,
+    /// 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(&note);
+        }
+        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
@@ -2122,4 +2259,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
+        );
+    }
 }

src/openhuman/tinyagents/mod.rs

@@ -1060,6 +1060,12 @@ fn assemble_turn_harness(
     };
     let handle = Some(orchestration::openhuman_steering_handle(steering_run_class));
 
+    // Memory protocol (issue #4116): observe the read → dedupe → write →
+    // update-index cycle and append a corrective note when a write skips the
+    // dedupe read or leaves the index stale. Pushed first / outermost so its
+    // `after_tool` runs *after* the byte-cap truncation, keeping the note.
+    harness.push_middleware(Arc::new(middleware::MemoryProtocolMiddleware::new()));
+
     // Repeated-failure circuit breaker: pause the run when a tool returns the same
     // error `REPEATED_TOOL_FAILURE_THRESHOLD` times in a row, so a deterministic
     // security/approval denial or terminal tool error surfaces its root cause

src/openhuman/tinyagents/tests.rs

@@ -505,13 +505,14 @@ fn adapter_inventory_registers_model_tools_and_middleware() {
     let serialized = serde_json::to_string(&stable_policies).unwrap();
     assert!(serialized.contains("\"classified\":true"));
 
-    // Lifecycle middleware, in registration order: repeated-tool-failure
-    // breaker, shadow tool-exposure, prompt-cache segment + guard, cache-align +
-    // tool-output (TurnContextMiddleware::defaults), cost budget, context
-    // compression + message trim (window known + autocompact on), SDK
-    // tool-policy projection, tool-outcome capture, arg recovery.
+    // Lifecycle middleware, in registration order: memory-protocol enforcement
+    // (outermost), repeated-tool-failure breaker, shadow tool-exposure,
+    // prompt-cache segment + guard, cache-align + tool-output
+    // (TurnContextMiddleware::defaults), cost budget, context compression +
+    // message trim (window known + autocompact on), SDK tool-policy projection,
+    // tool-outcome capture, arg recovery.
     let mw = assembled.harness.middleware();
-    assert_eq!(mw.len(), 12, "lifecycle middleware inventory");
+    assert_eq!(mw.len(), 13, "lifecycle middleware inventory");
     // Around-tool wraps: approval/security + CLI/RPC-only scope gate (no
     // builder tool policy on this call).
     assert_eq!(mw.tool_middleware_len(), 2, "tool middleware inventory");
@@ -575,7 +576,7 @@ fn adapter_inventory_gates_context_middleware_on_window() {
     let mw = assembled.harness.middleware();
     assert_eq!(
         mw.len(),
-        10,
+        11,
         "compression + trim must not install without a window"
     );
     assert!(assembled.early_exit_hook.is_none());