agentmux_srv/backend/
native_memory_drift.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Out-of-band write detection for native memory files — §4.5 of
5//! docs/specs/SPEC_MEMORY_VERSION_CONTROL_AND_ARMORY_AUDIT_2026_08_19.md.
6//! Claude Code's CLI writes memory `.md` files directly to
7//! `~/.claude/projects/<hash>/memory/*.md`, with no obligation to go
8//! through `agent:memory:write_file`/`memory.write` — an agent's own
9//! general-purpose filesystem tools can (and, per this spec's own
10//! motivating incident, did) write there directly. When that happens, no
11//! RPC fires, so nothing in `native_memory_handlers.rs` or `app_api::mod`
12//! ever sees it.
13//!
14//! Two independent layers close that gap, so failure of one degrades
15//! rather than blinds the whole system:
16//!
17//! 1. **Fast path** ([`spawn_fast_path`]) — subscribe to every known
18//!    agent's memory directory via the shared `FsWatchPool`
19//!    (`backend::fs_watch`), and record a new `external_fs_write` version
20//!    on any change whose hash doesn't match the latest recorded version.
21//! 2. **Slow path** ([`reconciliation_sweep_once`]) — a periodic sweep, run
22//!    on the same cadence class as `FsWatchPool`'s own
23//!    `HEALTH_SWEEP_INTERVAL` (30s), that hashes every known agent's live
24//!    memory files and compares against the latest recorded version — the
25//!    backstop for a missed fs-watch event, a watch still in retry-backoff,
26//!    or srv having been down entirely when the write happened.
27//!
28//! Both layers call the same [`check_and_record_drift`] — the only
29//! difference between them is how they learn a file might have changed.
30//! Neither promises zero data loss (see the spec's §4.5 "precision,
31//! honestly bounded" section): two out-of-band writes to the same file
32//! between one sweep and the next collapse into one detected version, and
33//! `source_detail` never guesses a `session_id` for a drift-detected write.
34
35use std::collections::HashSet;
36use std::path::{Path, PathBuf};
37use std::sync::Arc;
38use std::time::Duration;
39
40use crate::backend::fs_watch::{FsWatchEventKind, FsWatchPool};
41use crate::backend::storage::store::Store;
42use crate::server::native_memory_handlers::list_all_memory_targets;
43
44/// Same cadence class as `FsWatchPool::HEALTH_SWEEP_INTERVAL`
45/// (`backend/fs_watch/recovery.rs`) — chosen for the same reason: frequent
46/// enough that an operator reviewing history isn't waiting long for a
47/// drifted write to surface, infrequent enough not to matter for a
48/// background sweep over what's normally a handful of small markdown files
49/// per agent.
50const SWEEP_INTERVAL: Duration = Duration::from_secs(30);
51
52/// Same cap `native_memory_handlers.rs`/`app_api::mod` already use for a
53/// memory file's live content.
54const MAX_MEMORY_FILE_BYTES: u64 = 10 * 1024 * 1024;
55
56/// Compare `live_content` against the latest recorded version of
57/// `(agent_id, filename)`; if different (including "no version recorded
58/// yet"), insert a new version tagged `source: "external_fs_write"` with
59/// `source_detail: {"detected_via": detected_via}`. Returns `Ok(true)` if a
60/// new version was recorded, `Ok(false)` if the content already matched
61/// (no drift to report).
62pub(crate) fn check_and_record_drift(
63    id_store: &Store,
64    agent_id: &str,
65    filename: &str,
66    live_content: &str,
67    detected_via: &str,
68) -> Result<bool, String> {
69    let detail = serde_json::json!({ "detected_via": detected_via }).to_string();
70    // reagent P2 on PR #2675: the fast path (per fs-watch event) and slow
71    // path (30s sweep) run as concurrent, independent tokio tasks and can
72    // both observe the same file change — a plain "read latest, then
73    // separately insert" here (two Store calls, each independently
74    // locking/unlocking) is not atomic as a compound operation, so both
75    // could read the same stale "latest" and insert duplicate
76    // external_fs_write rows for one actual change.
77    // `agent_native_memory_version_insert_if_changed` does the compare
78    // AND the insert under a single connection-lock acquisition, closing
79    // that race — see its own doc comment for the full rationale.
80    id_store
81        .agent_native_memory_version_insert_if_changed(agent_id, filename, live_content, "external_fs_write", &detail, "")
82        .map_err(|e| e.to_string())
83        .map(|v| v.is_some())
84}
85
86/// Give every verified agent's memory file with no recorded version its
87/// first one, labelled `agent_inferred` ("Agent" in Memory), before the
88/// drift detector starts. Otherwise the detector finds these ordinary files
89/// on its first sweep and records them as `external_fs_write`, "Detected
90/// outside AgentMux", as it did for every file after a new local-build
91/// channel (SPEC_MEMORY_FOLLOWS_THE_AGENT_2026_09_24.md §1.3, phase M1).
92///
93/// Covers only files that **predate tracking**: those already in the
94/// directory at the moment it is first verified — at startup (after the
95/// identity stores are attached, so an agent's account-linked directory
96/// resolves; this used to be migration m0024, which ran before that and fell
97/// back to a registry guess), or later for an agent verified mid-session.
98///
99/// Anything written after that is a change, and drift detection records it
100/// exactly as it records every direct write to the folder today — including
101/// Claude's own auto-memory writes, which never go through AgentMux. So an
102/// agent whose folder doesn't exist yet is still counted as seen: its first
103/// files are written after verification, like every later one. Pre-existing
104/// files raise no fs-watch event, so the fast path can't race this.
105/// Returns the number of versions recorded.
106pub(crate) fn backfill_newly_verified(
107    id_store: &Store,
108    targets: &[(String, PathBuf)],
109    known: &mut HashSet<String>,
110) -> usize {
111    let new: Vec<(String, PathBuf)> = targets.iter().filter(|(id, _)| !known.contains(id)).cloned().collect();
112    for (id, _) in &new {
113        known.insert(id.clone());
114    }
115    backfill_first_sight_versions_for(id_store, &new)
116}
117
118fn backfill_first_sight_versions_for(id_store: &Store, targets: &[(String, PathBuf)]) -> usize {
119    let mut recorded = 0;
120    for (agent_id, dir) in targets {
121        let Ok(entries) = std::fs::read_dir(dir) else { continue };
122        for entry in entries.flatten() {
123            let filename = entry.file_name().to_string_lossy().into_owned();
124            if !filename.ends_with(".md") || !entry.path().is_file() {
125                continue;
126            }
127            match id_store.agent_native_memory_version_latest(agent_id, &filename) {
128                Ok(None) => {}
129                Ok(Some(_)) => continue,
130                Err(e) => {
131                    tracing::warn!(agent_id, filename, error = %e, "native_memory: first-sight backfill: lookup failed");
132                    continue;
133                }
134            }
135            let content = match read_memory_file_lossy(&entry.path()) {
136                Ok(c) => c,
137                Err(e) => {
138                    tracing::warn!(agent_id, filename, error = %e, "native_memory: first-sight backfill: skipping unreadable file");
139                    continue;
140                }
141            };
142            match id_store.agent_native_memory_version_insert_if_changed(agent_id, &filename, &content, "agent_inferred", "{}", "") {
143                Ok(Some(_)) => recorded += 1,
144                Ok(None) => {}
145                Err(e) => tracing::warn!(agent_id, filename, error = %e, "native_memory: first-sight backfill: insert failed"),
146            }
147        }
148    }
149    if recorded > 0 {
150        tracing::info!(recorded, "native_memory: first-sight backfill recorded versions");
151    }
152    recorded
153}
154
155/// Read one `.md` file's content the same way the rest of the native-memory
156/// code does: lossy UTF-8. Unlike `native_memory_handlers.rs`'s own
157/// mirror-refresh path — which truncates an oversized file and separately
158/// reports the filename so an interactive caller (`bundle.export_for_agent`)
159/// can surface a warning — the drift detector has no interactive caller to
160/// warn. Silently truncating here would instead record the truncated tail
161/// as a new *version*, permanently misrepresenting content the file never
162/// actually contained as if it were authoritative (reagent P2 on PR
163/// #2675). So an oversized file is refused outright (`ErrorKind::
164/// InvalidData`, distinguishable from a transient I/O error) rather than
165/// read partially; callers log and skip it, same treatment as any other
166/// per-file read failure.
167pub(crate) fn read_memory_file_lossy(path: &Path) -> std::io::Result<String> {
168    use std::io::Read;
169    let mut buf = Vec::new();
170    let read = std::fs::File::open(path)?
171        .take(MAX_MEMORY_FILE_BYTES + 1)
172        .read_to_end(&mut buf)?;
173    if read as u64 > MAX_MEMORY_FILE_BYTES {
174        return Err(std::io::Error::new(
175            std::io::ErrorKind::InvalidData,
176            format!("file exceeds the {MAX_MEMORY_FILE_BYTES}-byte cap — refusing to record truncated content as authoritative"),
177        ));
178    }
179    Ok(String::from_utf8_lossy(&buf).into_owned())
180}
181
182/// One pass over every agent definition with a configured working
183/// directory: hash each live `.md` file in its memory dir and record drift
184/// if found. Returns the number of drifted files recorded (for logging;
185/// tests assert on it directly). Errors from an individual agent/file are
186/// logged and swallowed — one agent's bad state (e.g. a permissions issue)
187/// must not stop the sweep from covering every other agent.
188///
189/// Also detects and publishes OUT-OF-BAND DELETIONS (Codex P2, PR #2932):
190/// content-drift detection alone only ever compares files that still exist
191/// against their last recorded version — a `.md` file removed directly (not
192/// through `agent:memory:write_file`/`revert`, neither of which delete
193/// files) never surfaces there, and the fast fs-watch path deliberately
194/// ignores `FsWatchEventKind::Removed` (it exists to catch content changes,
195/// not track deletions). Without this, the grid count and an open detail
196/// view for the deleted file would stay stale indefinitely — no event ever
197/// arrives to prompt the frontend's own "the selected file is now gone"
198/// handling to fire.
199///
200/// `deleted_notified` is this sweep's own persistent, cross-tick state
201/// (threaded in from `spawn_slow_path`'s loop, same pattern the fast path's
202/// `watched_dirs`/`dir_to_agent` already use) — without it, a file that
203/// STAYS deleted would republish an event every single 30s tick forever,
204/// since "missing from disk" is true on every subsequent sweep too, not
205/// just the first one. A filename is removed from this set again the
206/// moment it reappears on disk (recreated with new content), so a later
207/// re-deletion is detected fresh rather than staying permanently suppressed.
208pub(crate) fn reconciliation_sweep_once(
209    mstore: &Store,
210    id_store: &Store,
211    broker: &crate::backend::mps::Broker,
212    deleted_notified: &mut HashSet<(String, String)>,
213) -> usize {
214    let mut drifted = 0;
215    let targets = list_all_memory_targets(mstore);
216    for (agent_id, memory_dir) in &targets {
217        drifted += sweep_one_agent_dir(id_store, agent_id, memory_dir, "reconciliation_sweep", broker);
218    }
219
220    // agent_native_memory_version_list_distinct_files() is unscoped (every
221    // agent's every file with recorded history) — the same primitive
222    // native_memory_retention.rs already uses as its own work list, not a
223    // new query pattern introduced here.
224    let Ok(distinct) = id_store.agent_native_memory_version_list_distinct_files() else {
225        return drifted;
226    };
227    for (agent_id, filename) in distinct {
228        let Some((_, memory_dir)) = targets.iter().find(|(id, _)| *id == agent_id) else {
229            // Not (or no longer) a tracked target this sweep — out of scope
230            // for this pass; list_all_memory_targets is the same
231            // authoritative source the content-drift loop above already
232            // trusts for "which agents/dirs are live right now."
233            continue;
234        };
235        let key = (agent_id.clone(), filename.clone());
236        if memory_dir.join(&filename).is_file() {
237            deleted_notified.remove(&key);
238            continue;
239        }
240        if deleted_notified.insert(key) {
241            tracing::info!(agent_id = %agent_id, filename = %filename, "native_memory_drift: detected an out-of-band deletion");
242            broker.publish_event_for::<crate::backend::mps_events::AgentMemoryChanged>(&agent_id, &(), vec![], 0);
243        }
244    }
245    drifted
246}
247
248/// Also record the provider's writes into the agent's memory record — a
249/// folder its spawn has reconciled, and whose claim it still holds alone
250/// (SPEC_MEMORY_FOLLOWS_THE_AGENT_2026_09_24.md §2.1.3). Capture only;
251/// the folder isn't touched.
252fn capture_into_record(agent_id: &str, memory_dir: &Path) {
253    let Some(fs) = crate::backend::agent_session::global_transcript_store() else { return };
254    match crate::backend::memory_reconcile::capture_while_running(fs, agent_id, memory_dir, crate::backend::memory_reconcile::CAPTURE_SETTLE) {
255        Ok(0) => {}
256        Ok(n) => tracing::info!(agent_id, recorded = n, "native_memory_drift: recorded provider writes into the memory record"),
257        Err(e) => tracing::warn!(agent_id, error = %e, "native_memory_drift: memory record capture failed; retried next sweep"),
258    }
259}
260
261fn sweep_one_agent_dir(
262    id_store: &Store,
263    agent_id: &str,
264    memory_dir: &Path,
265    detected_via: &str,
266    broker: &crate::backend::mps::Broker,
267) -> usize {
268    capture_into_record(agent_id, memory_dir);
269    let entries = match std::fs::read_dir(memory_dir) {
270        Ok(e) => e,
271        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return 0,
272        Err(e) => {
273            tracing::warn!(agent_id, dir = %memory_dir.display(), error = %e, "native_memory_drift: sweep: read_dir failed");
274            return 0;
275        }
276    };
277
278    let mut drifted = 0;
279    for entry in entries.flatten() {
280        let name = entry.file_name().to_string_lossy().into_owned();
281        if !name.ends_with(".md") {
282            continue;
283        }
284        let Ok(file_type) = entry.file_type() else { continue };
285        if !file_type.is_file() {
286            continue;
287        }
288        let content = match read_memory_file_lossy(&entry.path()) {
289            Ok(c) => c,
290            Err(e) if e.kind() == std::io::ErrorKind::InvalidData => {
291                tracing::warn!(agent_id, filename = %name, error = %e, "native_memory_drift: sweep: file exceeds size cap, skipping (not recording truncated content)");
292                continue;
293            }
294            Err(e) => {
295                tracing::warn!(agent_id, filename = %name, error = %e, "native_memory_drift: sweep: read failed, retrying next sweep");
296                continue;
297            }
298        };
299        match check_and_record_drift(id_store, agent_id, &name, &content, detected_via) {
300            Ok(true) => {
301                drifted += 1;
302                tracing::info!(agent_id, filename = %name, detected_via, "native_memory_drift: recorded an out-of-band write");
303                // Reactive Memory updates (SPEC_ARMORY_REACTIVE_UPDATES_2026_09_02.md).
304                // Only on an actual recorded change (Ok(true)) — this branch
305                // runs once per sweep tick per watched agent, and firing on
306                // every no-op tick would mean an event for every agent every
307                // 30s regardless of whether anything happened.
308                broker.publish_event_for::<crate::backend::mps_events::AgentMemoryChanged>(&agent_id, &(), vec![], 0);
309            }
310            Ok(false) => {}
311            Err(e) => tracing::warn!(agent_id, filename = %name, error = %e, "native_memory_drift: check_and_record_drift failed"),
312        }
313    }
314    drifted
315}
316
317/// Start both layers. Call once at server startup with a fully-built
318/// `AppState`'s `fs_watch_pool`/`mstore`/`id_store`. Returns immediately —
319/// both loops run as spawned background tasks for the lifetime of the
320/// process (no shutdown handle; srv itself owns the process lifetime).
321pub fn spawn(
322    fs_watch_pool: Arc<FsWatchPool>,
323    mstore: Arc<Store>,
324    id_store: Arc<Store>,
325    broker: Arc<crate::backend::mps::Broker>,
326) {
327    spawn_fast_path(fs_watch_pool, mstore.clone(), id_store.clone(), broker.clone());
328    spawn_slow_path(mstore, id_store, broker);
329}
330
331fn spawn_slow_path(mstore: Arc<Store>, id_store: Arc<Store>, broker: Arc<crate::backend::mps::Broker>) {
332    tokio::spawn(async move {
333        let mut tick = tokio::time::interval(SWEEP_INTERVAL);
334        // Persistent across ticks, same pattern as the fast path's own
335        // watched_dirs/dir_to_agent — see reconciliation_sweep_once's own
336        // doc comment on why this can't just be a fresh HashSet per call.
337        let mut deleted_notified: HashSet<(String, String)> = HashSet::new();
338        // Agents whose directory has had its first-sight backfill.
339        let mut backfilled: HashSet<String> = HashSet::new();
340        loop {
341            tick.tick().await;
342            backfill_newly_verified(&id_store, &list_all_memory_targets(&mstore), &mut backfilled);
343            reconciliation_sweep_once(&mstore, &id_store, &broker, &mut deleted_notified);
344        }
345    });
346}
347
348/// Maintains a `memory_dir -> agent_id` map, refreshed every
349/// [`SWEEP_INTERVAL`] tick (piggybacking on the same cadence as the slow
350/// path rather than requiring a separate agent-spawn/teardown hook — a
351/// newly-defined agent is picked up within one tick, not instantly, which
352/// is an acceptable, honestly-bounded lag for a fast path whose whole
353/// purpose is narrowing a gap, not eliminating it). `FsWatchPool`
354/// subscriptions have no `Drop`-based teardown (see `pool.rs` — holding or
355/// discarding a `Subscription` value doesn't affect the underlying watch),
356/// so a dir is watched for the life of the process once first seen; we
357/// only need to track which dirs we've already called `subscribe_dir` on
358/// to avoid redundant re-subscription every tick.
359fn spawn_fast_path(
360    fs_watch_pool: Arc<FsWatchPool>,
361    mstore: Arc<Store>,
362    id_store: Arc<Store>,
363    broker: Arc<crate::backend::mps::Broker>,
364) {
365    tokio::spawn(async move {
366        let mut events = fs_watch_pool.events();
367        let mut watched_dirs: HashSet<PathBuf> = HashSet::new();
368        let mut dir_to_agent: std::collections::HashMap<PathBuf, String> = std::collections::HashMap::new();
369        let mut tick = tokio::time::interval(SWEEP_INTERVAL);
370
371        loop {
372            tokio::select! {
373                _ = tick.tick() => {
374                    refresh_subscriptions(&fs_watch_pool, &mstore, &id_store, &mut watched_dirs, &mut dir_to_agent);
375                }
376                event = events.recv() => {
377                    let event = match event {
378                        Ok(e) => e,
379                        Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
380                            // "a wake signal, not a guaranteed delivery log"
381                            // (FsWatchPool's own doc) — a lagged receiver
382                            // resyncs via the next reconciliation sweep
383                            // rather than trying to catch up here.
384                            continue;
385                        }
386                        Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
387                    };
388                    if !matches!(event.kind, FsWatchEventKind::Created | FsWatchEventKind::Modified) {
389                        continue;
390                    }
391                    let Some(filename) = event.path.file_name().and_then(|n| n.to_str()) else { continue };
392                    if !filename.ends_with(".md") {
393                        continue;
394                    }
395                    let Some(parent) = event.path.parent() else { continue };
396                    let Some(agent_id) = dir_to_agent.get(parent).cloned() else { continue };
397
398                    let content = match read_memory_file_lossy(&event.path) {
399                        Ok(c) => c,
400                        Err(e) if e.kind() == std::io::ErrorKind::InvalidData => {
401                            tracing::warn!(agent_id, filename, error = %e, "native_memory_drift: fast path: file exceeds size cap, skipping (not recording truncated content)");
402                            continue;
403                        }
404                        // TOCTOU-deleted or transient I/O between the event
405                        // and this read — non-fatal, the reconciliation
406                        // sweep covers it if the write actually stuck.
407                        Err(_) => continue,
408                    };
409                    // Restructured from `if let Err(...) {} else {}` to a full
410                    // match while adding the publish below: the old `else`
411                    // branch fired on Ok(false) too (content already matched,
412                    // nothing recorded), logging "recorded an out-of-band
413                    // write" for a no-op — a pre-existing log inaccuracy
414                    // fixed as a side effect of needing this branch anyway.
415                    capture_into_record(&agent_id, parent);
416                    match check_and_record_drift(&id_store, &agent_id, filename, &content, "fs_watch") {
417                        Ok(true) => {
418                            tracing::info!(agent_id, filename, "native_memory_drift: fast path recorded an out-of-band write");
419                            // Reactive Memory updates (SPEC_ARMORY_REACTIVE_UPDATES_2026_09_02.md).
420                            broker.publish_event_for::<crate::backend::mps_events::AgentMemoryChanged>(&agent_id, &(), vec![], 0);
421                        }
422                        Ok(false) => {}
423                        Err(e) => tracing::warn!(agent_id, filename, error = %e, "native_memory_drift: fast path: check_and_record_drift failed"),
424                    }
425                }
426            }
427        }
428    });
429}
430
431fn refresh_subscriptions(
432    fs_watch_pool: &Arc<FsWatchPool>,
433    mstore: &Store,
434    id_store: &Store,
435    watched_dirs: &mut HashSet<PathBuf>,
436    dir_to_agent: &mut std::collections::HashMap<PathBuf, String>,
437) {
438    for (agent_id, memory_dir) in list_all_memory_targets(mstore) {
439        // subscribe_dir canonicalizes internally; canonicalize here too so
440        // watched_dirs/dir_to_agent key on the same form an incoming
441        // event's path will actually have (events report canonical paths —
442        // see subscribe_dir's own doc comment).
443        let canonical = memory_dir.canonicalize().unwrap_or_else(|_| memory_dir.clone());
444        if watched_dirs.insert(canonical.clone()) {
445            // Record the files already here BEFORE watching, so a write
446            // landing right after the subscription is a change against them
447            // — never the first sighting that skips their own version. The
448            // slow sweep runs the same idempotent backfill.
449            backfill_first_sight_versions_for(id_store, &[(agent_id.clone(), memory_dir.clone())]);
450            fs_watch_pool.subscribe_dir(&memory_dir);
451            dir_to_agent.insert(canonical, agent_id);
452        }
453    }
454}
455
456#[cfg(test)]
457mod tests {
458    use super::*;
459    use crate::server::native_memory_handlers::memory_dir_for_agent_by_id;
460
461    fn shared_store() -> Store {
462        Store::open_shared(std::path::Path::new(":memory:")).unwrap()
463    }
464
465    /// A pre-existing file gets one `agent_inferred` version at startup, so
466    /// the first sweep sees no drift and never labels it "Detected outside
467    /// AgentMux" (SPEC_MEMORY_FOLLOWS_THE_AGENT_2026_09_24.md §1.3, M1).
468    #[test]
469    fn first_sight_backfill_records_once_and_the_sweep_then_sees_no_drift() {
470        let store = shared_store();
471        let dir = tempfile::tempdir().unwrap();
472        std::fs::write(dir.path().join("MEMORY.md"), "pre-existing").unwrap();
473        std::fs::write(dir.path().join("notes.txt"), "not memory").unwrap();
474        let targets = vec![("agent-1".to_string(), dir.path().to_path_buf())];
475        let mut known = HashSet::new();
476
477        assert_eq!(backfill_newly_verified(&store, &targets, &mut known), 1);
478        // A file that appears later in a known directory is an outside write
479        // for drift detection to record, not another first sighting.
480        std::fs::write(dir.path().join("later.md"), "appeared later").unwrap();
481        assert_eq!(backfill_newly_verified(&store, &targets, &mut known), 0);
482        assert!(store.agent_native_memory_version_latest("agent-1", "later.md").unwrap().is_none());
483        // An agent verified mid-session gets its own first sighting.
484        let dir2 = tempfile::tempdir().unwrap();
485        std::fs::write(dir2.path().join("MEMORY.md"), "second agent").unwrap();
486        let more = vec![targets[0].clone(), ("agent-2".to_string(), dir2.path().to_path_buf())];
487        assert_eq!(backfill_newly_verified(&store, &more, &mut known), 1);
488
489        // A folder that doesn't exist yet has nothing that predates tracking:
490        // the agent is seen, and files written later are changes for drift
491        // detection, like every direct write to the folder.
492        let root = tempfile::tempdir().unwrap();
493        let not_yet = root.path().join("memory");
494        let pending = vec![("agent-3".to_string(), not_yet.clone())];
495        assert_eq!(backfill_newly_verified(&store, &pending, &mut known), 0);
496        assert!(known.contains("agent-3"));
497        std::fs::create_dir_all(&not_yet).unwrap();
498        std::fs::write(not_yet.join("MEMORY.md"), "first write").unwrap();
499        assert_eq!(backfill_newly_verified(&store, &pending, &mut known), 0);
500        let history = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
501        assert_eq!(history.len(), 1);
502        assert_eq!(history[0].source, "agent_inferred");
503
504        let drifted = check_and_record_drift(&store, "agent-1", "MEMORY.md", "pre-existing", "reconciliation_sweep").unwrap();
505        assert!(!drifted, "unchanged content is not an outside write");
506    }
507
508    /// The fast path records a folder's existing files before it starts
509    /// watching, so an edit right after subscribing is a change against
510    /// them, not their first sighting (ReAgent P1 on #3707).
511    #[tokio::test]
512    async fn the_fast_path_backfills_a_folder_before_watching_it() {
513        let _guard = crate::test_support::ISOLATED_AUTH_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
514        let id_store = shared_store();
515        let mstore = Store::open_in_memory().unwrap();
516        let config = tempfile::tempdir().unwrap();
517        let mut def = crate::backend::storage::agents::test_agent_def("agent-fp", "Agent FP", "claude", "agent", 1, "");
518        def.working_directory = "/work/fp".to_string();
519        mstore.agent_def_insert(&mut def).unwrap();
520        mstore
521            .agent_content_set(&crate::backend::storage::AgentContent {
522                agent_id: def.id.clone(),
523                content_type: "env".to_string(),
524                content: format!("CLAUDE_CONFIG_DIR={}\n", config.path().display()),
525                updated_at: 0,
526            })
527            .unwrap();
528        let dir = memory_dir_for_agent_by_id(&mstore, &def).unwrap();
529        std::fs::create_dir_all(&dir).unwrap();
530        std::fs::write(dir.join("MEMORY.md"), "before watching").unwrap();
531
532        let pool = FsWatchPool::new();
533        let mut watched = HashSet::new();
534        let mut dir_to_agent = std::collections::HashMap::new();
535        refresh_subscriptions(&pool, &mstore, &id_store, &mut watched, &mut dir_to_agent);
536
537        let v = id_store.agent_native_memory_version_latest(&def.id, "MEMORY.md").unwrap().unwrap();
538        assert_eq!(v.source, "agent_inferred");
539        assert_eq!(v.content, "before watching");
540    }
541
542    #[test]
543    fn first_observation_of_a_file_is_drift() {
544        let store = shared_store();
545        let drifted = check_and_record_drift(&store, "agent-1", "MEMORY.md", "content", "fs_watch").unwrap();
546        assert!(drifted, "a file with no prior version must count as drift");
547
548        let history = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
549        assert_eq!(history.len(), 1);
550        assert_eq!(history[0].source, "external_fs_write");
551        assert!(history[0].source_detail.contains("fs_watch"));
552    }
553
554    #[test]
555    fn unchanged_content_is_not_drift() {
556        let store = shared_store();
557        store
558            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "same", "human", "{}", "")
559            .unwrap();
560
561        let drifted = check_and_record_drift(&store, "agent-1", "MEMORY.md", "same", "fs_watch").unwrap();
562        assert!(!drifted, "identical content must not record a new version");
563
564        let history = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
565        assert_eq!(history.len(), 1, "no new version should have been inserted");
566    }
567
568    #[test]
569    fn changed_content_is_drift_and_chains_onto_the_prior_version() {
570        let store = shared_store();
571        let v1 = store
572            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v1", "human", "{}", "")
573            .unwrap();
574
575        let drifted = check_and_record_drift(&store, "agent-1", "MEMORY.md", "v2 — written outside AgentMux", "reconciliation_sweep").unwrap();
576        assert!(drifted);
577
578        let history = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
579        assert_eq!(history.len(), 2);
580        assert_eq!(history[0].parent_version_id, Some(v1.id));
581        assert_eq!(history[0].source, "external_fs_write");
582        assert!(history[0].source_detail.contains("reconciliation_sweep"));
583    }
584
585    #[test]
586    fn sweep_finds_a_file_written_directly_to_disk() {
587        let store = shared_store();
588        let tmp = tempfile::tempdir().unwrap();
589        std::fs::write(tmp.path().join("MEMORY.md"), "written outside any RPC").unwrap();
590
591        let broker = crate::backend::mps::Broker::new();
592        let drifted = sweep_one_agent_dir(&store, "agent-1", tmp.path(), "reconciliation_sweep", &broker);
593        assert_eq!(drifted, 1);
594
595        let history = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
596        assert_eq!(history.len(), 1);
597        assert_eq!(history[0].source, "external_fs_write");
598    }
599
600    // SPEC_ARMORY_REACTIVE_UPDATES_2026_09_02.md: an out-of-band write
601    // (bypassing both the WS RPC and MuxAPI surfaces entirely) must still
602    // make the Memory → Personal grid reactive — that's this whole module's reason to
603    // exist, so its own publish path needs the same regression guard the
604    // RPC handlers' publish calls get.
605    #[test]
606    fn sweep_publishes_agent_memory_changed_when_drift_is_recorded() {
607        let store = shared_store();
608        let tmp = tempfile::tempdir().unwrap();
609        std::fs::write(tmp.path().join("MEMORY.md"), "written outside any RPC").unwrap();
610
611        let (broker, client) = crate::test_support::broker_recording("agent:memory:changed:agent-drift-1");
612        sweep_one_agent_dir(&store, "agent-drift-1", tmp.path(), "reconciliation_sweep", &broker);
613
614        let events = client.received_events();
615        assert_eq!(events.len(), 1, "a real drift must publish exactly one event");
616        assert_eq!(events[0].1.event, "agent:memory:changed:agent-drift-1");
617    }
618
619    // The other half of the same guard: a sweep tick that finds NO drift
620    // (unchanged content) must NOT publish — the reconciliation sweep runs
621    // every 30s for every watched agent regardless of whether anything
622    // changed, so publishing unconditionally would fire an event storm with
623    // no corresponding real change, defeating the point of an event-driven
624    // refresh (see this module's own SPEC section on why insert_if_changed's
625    // no-op case is explicitly excluded).
626    #[test]
627    fn sweep_does_not_publish_when_content_is_unchanged() {
628        let store = shared_store();
629        let tmp = tempfile::tempdir().unwrap();
630        std::fs::write(tmp.path().join("MEMORY.md"), "same content").unwrap();
631
632        let (broker, client) = crate::test_support::broker_recording("agent:memory:changed:agent-drift-2");
633        // First sweep: genuinely new content, records a version (and would
634        // publish — unobserved here since this test is about the SECOND
635        // sweep finding nothing new).
636        sweep_one_agent_dir(&store, "agent-drift-2", tmp.path(), "reconciliation_sweep", &broker);
637        // Second sweep over the same, unchanged file — no drift to record.
638        sweep_one_agent_dir(&store, "agent-drift-2", tmp.path(), "reconciliation_sweep", &broker);
639
640        let events = client.received_events();
641        assert_eq!(events.len(), 1, "only the first sweep's real drift may publish; the second (no-op) must not");
642    }
643
644    // reagent P2 on PR #2675: read_memory_file_lossy previously truncated
645    // an oversized file and returned the truncated bytes as if they were
646    // the whole file — the sweep would then record that truncated content
647    // as a new, authoritative version. It must instead refuse to read the
648    // file at all, so no version (correct or corrupted) is recorded.
649    #[test]
650    fn sweep_refuses_to_record_an_oversized_file_as_authoritative() {
651        let store = shared_store();
652        let tmp = tempfile::tempdir().unwrap();
653        let oversized = "x".repeat((MAX_MEMORY_FILE_BYTES + 1) as usize);
654        std::fs::write(tmp.path().join("MEMORY.md"), &oversized).unwrap();
655
656        let broker = crate::backend::mps::Broker::new();
657        let drifted = sweep_one_agent_dir(&store, "agent-1", tmp.path(), "reconciliation_sweep", &broker);
658        assert_eq!(drifted, 0, "an oversized file must not be recorded as drift");
659
660        let history = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
661        assert!(history.is_empty(), "no truncated content may be recorded as a version");
662    }
663
664    #[test]
665    fn read_memory_file_lossy_reads_a_file_at_exactly_the_cap() {
666        let tmp = tempfile::tempdir().unwrap();
667        let path = tmp.path().join("MEMORY.md");
668        let at_cap = "x".repeat(MAX_MEMORY_FILE_BYTES as usize);
669        std::fs::write(&path, &at_cap).unwrap();
670
671        let content = read_memory_file_lossy(&path).unwrap();
672        assert_eq!(content.len(), MAX_MEMORY_FILE_BYTES as usize);
673    }
674
675    #[test]
676    fn read_memory_file_lossy_rejects_a_file_one_byte_over_the_cap() {
677        let tmp = tempfile::tempdir().unwrap();
678        let path = tmp.path().join("MEMORY.md");
679        let over_cap = "x".repeat((MAX_MEMORY_FILE_BYTES + 1) as usize);
680        std::fs::write(&path, &over_cap).unwrap();
681
682        let err = read_memory_file_lossy(&path).unwrap_err();
683        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
684    }
685
686    #[test]
687    fn sweep_ignores_non_md_files_and_missing_directories() {
688        let store = shared_store();
689        let tmp = tempfile::tempdir().unwrap();
690        std::fs::write(tmp.path().join("notes.txt"), "not a memory file").unwrap();
691
692        let broker = crate::backend::mps::Broker::new();
693        assert_eq!(sweep_one_agent_dir(&store, "agent-1", tmp.path(), "reconciliation_sweep", &broker), 0);
694        assert_eq!(
695            sweep_one_agent_dir(&store, "agent-1", &tmp.path().join("does-not-exist"), "reconciliation_sweep", &broker),
696            0
697        );
698    }
699
700    #[test]
701    fn sweep_scopes_versions_to_the_given_agent_id() {
702        let store = shared_store();
703        let tmp = tempfile::tempdir().unwrap();
704        std::fs::write(tmp.path().join("MEMORY.md"), "content").unwrap();
705
706        let broker = crate::backend::mps::Broker::new();
707        sweep_one_agent_dir(&store, "agent-specific", tmp.path(), "reconciliation_sweep", &broker);
708
709        assert_eq!(store.agent_native_memory_version_list("agent-specific", "MEMORY.md").unwrap().len(), 1);
710        assert_eq!(store.agent_native_memory_version_list("agent-other", "MEMORY.md").unwrap().len(), 0);
711    }
712
713    #[tokio::test]
714    async fn reconciliation_sweep_once_covers_every_agent_with_a_working_directory() {
715        // list_all_memory_targets also consults the global named-agent
716        // registry (added for reagent P1 on PR #2675) — without isolating
717        // AGENTMUX_SHARED_DIR here, this test would sweep whatever real
718        // live agents happen to be registered on the machine running the
719        // suite, not just the two synthetic ones it sets up below.
720        let _guard = crate::test_support::ISOLATED_AUTH_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
721        let prev_shared_dir = std::env::var_os("AGENTMUX_SHARED_DIR");
722        let shared = tempfile::tempdir().unwrap();
723        std::env::set_var("AGENTMUX_SHARED_DIR", shared.path());
724
725        let tmp = tempfile::NamedTempFile::new().unwrap();
726        let id_store = Store::open_shared(tmp.path()).unwrap();
727        let state = crate::server::tests::test_state();
728
729        let config_a = tempfile::tempdir().unwrap();
730        let config_b = tempfile::tempdir().unwrap();
731        for (id, dir) in [("sweep-agent-a", config_a.path()), ("sweep-agent-b", config_b.path())] {
732            let mut def = crate::backend::storage::AgentDefinition {
733                conversation_visibility: crate::backend::storage::agents::default_conversation_visibility(),
734                id: id.to_string(),
735                slug: id.to_string(),
736                name: "Test".to_string(),
737                icon: String::new(),
738                provider: "claude".to_string(),
739                description: String::new(),
740                working_directory: format!("/work/{id}"),
741                shell: String::new(),
742                provider_flags: String::new(),
743                auto_start: 0,
744                restart_on_crash: 0,
745                idle_timeout_minutes: 0,
746                created_at: 0,
747                agent_type: "host".to_string(),
748                environment: String::new(),
749                agent_bus_id: String::new(),
750                is_seeded: 0,
751                accounts: String::new(),
752                parent_id: String::new(),
753                branch_label: String::new(),
754                updated_at: 0,
755                user_hidden: 0,
756                container_image: String::new(),
757                container_volumes: "[]".to_string(),
758                container_name: String::new(),
759                use_ambient_login: 0,
760                auto_continue_enabled: 0,
761                model_vendor_base_url: String::new(),
762                memory_id: String::new(),
763            };
764            state.mstore.agent_def_insert(&mut def).unwrap();
765            state
766                .mstore
767                .agent_content_set(&crate::backend::storage::AgentContent {
768                    agent_id: id.to_string(),
769                    content_type: "env".to_string(),
770                    content: format!("CLAUDE_CONFIG_DIR={}\n", dir.display()),
771                    updated_at: 0,
772                })
773                .unwrap();
774
775            let memory_dir = memory_dir_for_agent_by_id(&state.mstore, &def).unwrap();
776            std::fs::create_dir_all(&memory_dir).unwrap();
777            std::fs::write(memory_dir.join("MEMORY.md"), format!("content for {id}")).unwrap();
778        }
779
780        let broker = crate::backend::mps::Broker::new();
781        let mut deleted_notified = HashSet::new();
782        let drifted = reconciliation_sweep_once(&state.mstore, &id_store, &broker, &mut deleted_notified);
783        assert_eq!(drifted, 2, "sweep must cover every agent with a working directory");
784        assert_eq!(id_store.agent_native_memory_version_list("sweep-agent-a", "MEMORY.md").unwrap().len(), 1);
785        assert_eq!(id_store.agent_native_memory_version_list("sweep-agent-b", "MEMORY.md").unwrap().len(), 1);
786
787        match prev_shared_dir {
788            Some(v) => std::env::set_var("AGENTMUX_SHARED_DIR", v),
789            None => std::env::remove_var("AGENTMUX_SHARED_DIR"),
790        }
791    }
792
793    // Codex P2, PR #2932: content-drift detection alone never notices a
794    // file that's gone (nothing to compare content against), and the fast
795    // path deliberately ignores FsWatchEventKind::Removed — without this,
796    // an out-of-band deletion of the file open in the Memory → Personal detail view
797    // stayed stale forever, since no event ever arrived to trigger the
798    // frontend's own "the selected file is gone" handling.
799    #[tokio::test]
800    async fn reconciliation_sweep_detects_an_out_of_band_deletion_once_then_suppresses_repeats_until_recreated() {
801        let _guard = crate::test_support::ISOLATED_AUTH_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
802        let prev_shared_dir = std::env::var_os("AGENTMUX_SHARED_DIR");
803        let shared = tempfile::tempdir().unwrap();
804        std::env::set_var("AGENTMUX_SHARED_DIR", shared.path());
805
806        let tmp = tempfile::NamedTempFile::new().unwrap();
807        let id_store = Store::open_shared(tmp.path()).unwrap();
808        let state = crate::server::tests::test_state();
809        let config = tempfile::tempdir().unwrap();
810
811        let mut def = crate::backend::storage::AgentDefinition {
812            conversation_visibility: crate::backend::storage::agents::default_conversation_visibility(),
813            id: "sweep-delete-agent".to_string(),
814            slug: "sweep-delete-agent".to_string(),
815            name: "Test".to_string(),
816            icon: String::new(),
817            provider: "claude".to_string(),
818            description: String::new(),
819            working_directory: "/work/sweep-delete-agent".to_string(),
820            shell: String::new(),
821            provider_flags: String::new(),
822            auto_start: 0,
823            restart_on_crash: 0,
824            idle_timeout_minutes: 0,
825            created_at: 0,
826            agent_type: "host".to_string(),
827            environment: String::new(),
828            agent_bus_id: String::new(),
829            is_seeded: 0,
830            accounts: String::new(),
831            parent_id: String::new(),
832            branch_label: String::new(),
833            updated_at: 0,
834            user_hidden: 0,
835            container_image: String::new(),
836            container_volumes: "[]".to_string(),
837            container_name: String::new(),
838            use_ambient_login: 0,
839            auto_continue_enabled: 0,
840            model_vendor_base_url: String::new(),
841            memory_id: String::new(),
842        };
843        state.mstore.agent_def_insert(&mut def).unwrap();
844        state
845            .mstore
846            .agent_content_set(&crate::backend::storage::AgentContent {
847                agent_id: def.id.clone(),
848                content_type: "env".to_string(),
849                content: format!("CLAUDE_CONFIG_DIR={}\n", config.path().display()),
850                updated_at: 0,
851            })
852            .unwrap();
853        let memory_dir = memory_dir_for_agent_by_id(&state.mstore, &def).unwrap();
854        std::fs::create_dir_all(&memory_dir).unwrap();
855        let file_path = memory_dir.join("MEMORY.md");
856        std::fs::write(&file_path, "content").unwrap();
857
858        let (broker, client) = crate::test_support::broker_recording("agent:memory:changed:sweep-delete-agent");
859        let mut deleted_notified = HashSet::new();
860
861        // Sweep 1: file present — establishes the recorded version (and
862        // publishes once, from ordinary content-drift, not deletion logic).
863        reconciliation_sweep_once(&state.mstore, &id_store, &broker, &mut deleted_notified);
864        assert_eq!(client.received_events().len(), 1);
865
866        // Delete the file out of band (not through write_file/revert).
867        std::fs::remove_file(&file_path).unwrap();
868
869        // Sweep 2: first sweep to observe the deletion — must publish.
870        reconciliation_sweep_once(&state.mstore, &id_store, &broker, &mut deleted_notified);
871        assert_eq!(client.received_events().len(), 2, "the sweep that first observes a deletion must publish");
872
873        // Sweep 3: file still gone — must NOT publish again (suppressed;
874        // otherwise a permanently-deleted file would fire an event every
875        // 30s tick forever).
876        reconciliation_sweep_once(&state.mstore, &id_store, &broker, &mut deleted_notified);
877        assert_eq!(client.received_events().len(), 2, "a repeat sweep over an already-reported deletion must not re-publish");
878
879        // Recreate the file — content-drift detects it as a new version
880        // (publish #3) and clears the deletion-suppression for this file.
881        std::fs::write(&file_path, "recreated content").unwrap();
882        reconciliation_sweep_once(&state.mstore, &id_store, &broker, &mut deleted_notified);
883        assert_eq!(client.received_events().len(), 3);
884
885        // Delete it again — must be detected as a FRESH deletion (publish
886        // #4), proving the suppression state was actually cleared on
887        // recreation, not permanently stuck once a file is ever deleted once.
888        std::fs::remove_file(&file_path).unwrap();
889        reconciliation_sweep_once(&state.mstore, &id_store, &broker, &mut deleted_notified);
890        assert_eq!(client.received_events().len(), 4, "a re-deletion after recreation must be detected as a fresh event");
891
892        match prev_shared_dir {
893            Some(v) => std::env::set_var("AGENTMUX_SHARED_DIR", v),
894            None => std::env::remove_var("AGENTMUX_SHARED_DIR"),
895        }
896    }
897}
agentmux_srv/backend/native_memory_drift.rs

agentmux_srv/backend/
native_memory_drift.rs
