agentmux_srv/backend/storage/
agent_native_memory_versions.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Append-only version history for native memory content. See
5//! `db_agent_native_memory_versions` in migrations.rs
6//! (`SHARED_STORE_SCHEMA_VERSION` v8 / `OBJECT_SCHEMA_VERSION` v24 /
7//! `IDENTITY_STORE_SCHEMA_VERSION` v3) and
8//! docs/specs/SPEC_MEMORY_VERSION_CONTROL_AND_ARMORY_AUDIT_2026_08_19.md for
9//! the full design.
10//!
11//! Additive to `db_agent_native_memory` (the current-value mirror in
12//! `agent_native_memory.rs`) — never read on that module's hot path
13//! (`list`/`read_file`), only by the `agent:memory:history`/`diff`/`revert`
14//! RPCs. A version row is inserted on every `agent:memory:write_file` call
15//! (source-tagged; see the spec's §4.1) and, separately, by
16//! `agent:memory:revert` (source `"revert"`, never overwriting or deleting
17//! a prior row — a revert is always a new version, git-revert-style).
18
19use rusqlite::params;
20use sha2::{Digest, Sha256};
21
22use super::error::StoreError;
23use super::store::Store;
24
25/// One version of one memory file's content, including its full body.
26#[derive(Debug, Clone, PartialEq)]
27pub struct NativeMemoryVersion {
28    pub id: String,
29    pub agent_id: String,
30    pub filename: String,
31    pub content: String,
32    pub content_hash: String,
33    /// The version this one superseded, or `None` for the first version
34    /// ever recorded for this `(agent_id, filename)`.
35    pub parent_version_id: Option<String>,
36    /// `"human"` | `"agent_inferred"` | `"jekt"` | `"external_fs_write"` |
37    /// `"revert"` — not a Rust enum: this column is read/written by
38    /// forward-compatible callers (new sources can be added without a
39    /// schema migration), and validated at the RPC layer instead.
40    pub source: String,
41    /// JSON blob — jekt marker fields when `source == "jekt"`, detection
42    /// method when `source == "external_fs_write"`, `"{}"` otherwise.
43    pub source_detail: String,
44    pub session_id: String,
45    pub created_at: i64,
46}
47
48/// A version's metadata without its full content — the list-view shape,
49/// mirroring `NativeMemoryMirrorRow`'s own list_meta/read split in
50/// `agent_native_memory.rs` (a history view renders many rows at once; a
51/// diff/revert call reads exactly one or two full bodies).
52#[derive(Debug, Clone, PartialEq)]
53pub struct NativeMemoryVersionSummary {
54    pub id: String,
55    pub content_hash: String,
56    pub parent_version_id: Option<String>,
57    pub source: String,
58    pub source_detail: String,
59    pub session_id: String,
60    pub created_at: i64,
61}
62
63use agentmux_common::time::now_ms;
64
65/// SHA-256 hex digest of a memory file's content — used both to populate
66/// `content_hash` on insert and, later, by §4.5 drift detection to compare
67/// a live file's current hash against the last recorded version without
68/// storing the live content redundantly.
69pub(crate) fn content_hash(content: &str) -> String {
70    let mut hasher = Sha256::new();
71    hasher.update(content.as_bytes());
72    hex::encode(hasher.finalize())
73}
74
75impl Store {
76    /// Record a new version of `(agent_id, filename)`'s content, chained
77    /// onto whatever version was previously latest for this file (`None`
78    /// for the first write ever recorded). Always inserts, even when
79    /// `content` is byte-identical to the previous version — simplicity
80    /// over cleverness; dedup-by-hash is a possible future optimization,
81    /// not this one.
82    ///
83    /// reagent P2 (re-review): the read-latest-then-insert sequence is a
84    /// single connection-lock acquisition, not two — an earlier revision
85    /// released the lock between reading `parent_version_id` and
86    /// inserting, so two truly concurrent calls for the same `(agent_id,
87    /// filename)` (e.g. two panes of the same agent identity writing at
88    /// once) could both read the same latest id and insert sibling
89    /// versions instead of a linear chain.
90    ///
91    /// reagent P2 (second re-review, PR #2674): an in-process `Mutex` only
92    /// serializes callers sharing this one `Store`/`Connection` — it does
93    /// nothing for two SEPARATE `srv` processes (e.g. two channels/
94    /// instances of AgentMux) each holding their own `Connection` to the
95    /// same shared-store SQLite file. Those two connections' SELECT-latest
96    /// and INSERT could still interleave across processes, producing
97    /// sibling versions with the same `parent_version_id`. Wrapping the
98    /// read+insert in a `BEGIN IMMEDIATE` transaction closes that gap at
99    /// the SQLite level: it acquires the RESERVED lock up front (before
100    /// the SELECT runs), so a second connection's own `BEGIN IMMEDIATE`
101    /// blocks (up to `busy_timeout`, already configured at open) until
102    /// this one commits — true cross-process mutual exclusion, unlike a
103    /// plain deferred transaction (whose lock isn't acquired until the
104    /// first write, after the read has already happened).
105    pub fn agent_native_memory_version_insert(
106        &self,
107        agent_id: &str,
108        filename: &str,
109        content: &str,
110        source: &str,
111        source_detail: &str,
112        session_id: &str,
113    ) -> Result<NativeMemoryVersion, StoreError> {
114        let id = uuid::Uuid::new_v4().to_string();
115        let hash = content_hash(content);
116        let created_at = now_ms();
117
118        let mut conn = self.conn.lock().unwrap();
119        let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
120        let parent_version_id: Option<String> = {
121            let mut stmt = tx.prepare(
122                "SELECT id FROM db_agent_native_memory_versions
123                 WHERE agent_id = ?1 AND filename = ?2
124                 ORDER BY created_at DESC, rowid DESC
125                 LIMIT 1",
126            )?;
127            match stmt.query_row(params![agent_id, filename], |row| row.get::<_, String>(0)) {
128                Ok(id) => Some(id),
129                Err(rusqlite::Error::QueryReturnedNoRows) => None,
130                Err(e) => return Err(e.into()),
131            }
132        };
133        tx.execute(
134            "INSERT INTO db_agent_native_memory_versions
135                 (id, agent_id, filename, content, content_hash, parent_version_id, source, source_detail, session_id, created_at)
136             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
137            params![
138                id,
139                agent_id,
140                filename,
141                content,
142                hash,
143                parent_version_id,
144                source,
145                source_detail,
146                session_id,
147                created_at,
148            ],
149        )?;
150        tx.commit()?;
151
152        Ok(NativeMemoryVersion {
153            id,
154            agent_id: agent_id.to_string(),
155            filename: filename.to_string(),
156            content: content.to_string(),
157            content_hash: hash,
158            parent_version_id,
159            source: source.to_string(),
160            source_detail: source_detail.to_string(),
161            session_id: session_id.to_string(),
162            created_at,
163        })
164    }
165
166    /// Atomic compare-and-insert: within a SINGLE connection-lock
167    /// acquisition, reads the latest version's hash for `(agent_id,
168    /// filename)` and inserts a new version only if `content`'s hash
169    /// differs (or no version exists yet). Returns `Ok(None)` when nothing
170    /// was inserted (content unchanged).
171    ///
172    /// Exists specifically for `native_memory_drift.rs`'s two concurrent
173    /// detectors (fast fs-watch path, slow reconciliation sweep) — reagent
174    /// P2 on PR #2675: a separate `agent_native_memory_version_latest`
175    /// call followed by a separate `agent_native_memory_version_insert`
176    /// call (each independently acquiring and releasing the connection
177    /// lock) is NOT atomic as a *compound* operation, even though each
178    /// individual call is serialized — two concurrent callers can both
179    /// observe the same stale "latest" between their own read and write,
180    /// producing duplicate version rows for one actual out-of-band change.
181    /// A caller that doesn't need this compare-and-swap guarantee (e.g. an
182    /// explicit RPC write, where every call is expected to record a new
183    /// version regardless — see `agent_native_memory_version_insert`'s own
184    /// doc) should keep using the plain method above instead.
185    ///
186    /// Same cross-connection hardening as `agent_native_memory_version_
187    /// insert` above (reagent P2 on PR #2674, applied here proactively for
188    /// the identical read-then-insert shape): the read+insert runs inside
189    /// a single `BEGIN IMMEDIATE` transaction, not just a single connection
190    /// lock, so two separate `srv` processes/channels racing this same
191    /// compare-and-insert can't both observe the same stale "latest" either.
192    pub fn agent_native_memory_version_insert_if_changed(
193        &self,
194        agent_id: &str,
195        filename: &str,
196        content: &str,
197        source: &str,
198        source_detail: &str,
199        session_id: &str,
200    ) -> Result<Option<NativeMemoryVersion>, StoreError> {
201        let hash = content_hash(content);
202        let mut conn = self.conn.lock().unwrap();
203        let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
204
205        let latest: Option<(String, String)> = {
206            let mut stmt = tx.prepare(
207                "SELECT id, content_hash FROM db_agent_native_memory_versions
208                 WHERE agent_id = ?1 AND filename = ?2
209                 ORDER BY created_at DESC, rowid DESC
210                 LIMIT 1",
211            )?;
212            match stmt.query_row(params![agent_id, filename], |row| {
213                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
214            }) {
215                Ok(v) => Some(v),
216                Err(rusqlite::Error::QueryReturnedNoRows) => None,
217                Err(e) => return Err(e.into()),
218            }
219        };
220
221        if let Some((_, latest_hash)) = &latest {
222            if *latest_hash == hash {
223                return Ok(None);
224            }
225        }
226
227        let id = uuid::Uuid::new_v4().to_string();
228        let parent_version_id = latest.map(|(id, _)| id);
229        let created_at = now_ms();
230        tx.execute(
231            "INSERT INTO db_agent_native_memory_versions
232                 (id, agent_id, filename, content, content_hash, parent_version_id, source, source_detail, session_id, created_at)
233             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
234            params![
235                id,
236                agent_id,
237                filename,
238                content,
239                hash,
240                parent_version_id,
241                source,
242                source_detail,
243                session_id,
244                created_at,
245            ],
246        )?;
247        tx.commit()?;
248
249        Ok(Some(NativeMemoryVersion {
250            id,
251            agent_id: agent_id.to_string(),
252            filename: filename.to_string(),
253            content: content.to_string(),
254            content_hash: hash,
255            parent_version_id,
256            source: source.to_string(),
257            source_detail: source_detail.to_string(),
258            session_id: session_id.to_string(),
259            created_at,
260        }))
261    }
262
263    /// List every version of `(agent_id, filename)`, newest first — no
264    /// content (list-view; see [`NativeMemoryVersionSummary`]'s own doc).
265    pub fn agent_native_memory_version_list(
266        &self,
267        agent_id: &str,
268        filename: &str,
269    ) -> Result<Vec<NativeMemoryVersionSummary>, StoreError> {
270        let conn = self.conn.lock().unwrap();
271        let mut stmt = conn.prepare(
272            "SELECT id, content_hash, parent_version_id, source, source_detail, session_id, created_at
273             FROM db_agent_native_memory_versions
274             WHERE agent_id = ?1 AND filename = ?2
275             ORDER BY created_at DESC, rowid DESC",
276        )?;
277        let rows = stmt
278            .query_map(params![agent_id, filename], |row| {
279                Ok(NativeMemoryVersionSummary {
280                    id: row.get(0)?,
281                    content_hash: row.get(1)?,
282                    parent_version_id: row.get(2)?,
283                    source: row.get(3)?,
284                    source_detail: row.get(4)?,
285                    session_id: row.get(5)?,
286                    created_at: row.get(6)?,
287                })
288            })?
289            .collect::<Result<Vec<_>, _>>()?;
290        Ok(rows)
291    }
292
293    /// The most recently recorded version of `(agent_id, filename)`, or
294    /// `None` if no version has ever been recorded. Used both to derive a
295    /// new write's `parent_version_id` and (§4.5 drift detection, not yet
296    /// built) to compare a live file's current hash against the last known
297    /// version.
298    pub fn agent_native_memory_version_latest(
299        &self,
300        agent_id: &str,
301        filename: &str,
302    ) -> Result<Option<NativeMemoryVersion>, StoreError> {
303        let conn = self.conn.lock().unwrap();
304        let mut stmt = conn.prepare(
305            "SELECT id, content, content_hash, parent_version_id, source, source_detail, session_id, created_at
306             FROM db_agent_native_memory_versions
307             WHERE agent_id = ?1 AND filename = ?2
308             ORDER BY created_at DESC, rowid DESC
309             LIMIT 1",
310        )?;
311        let agent_id_owned = agent_id.to_string();
312        let filename_owned = filename.to_string();
313        match stmt.query_row(params![agent_id, filename], |row| {
314            Ok(NativeMemoryVersion {
315                id: row.get(0)?,
316                agent_id: agent_id_owned.clone(),
317                filename: filename_owned.clone(),
318                content: row.get(1)?,
319                content_hash: row.get(2)?,
320                parent_version_id: row.get(3)?,
321                source: row.get(4)?,
322                source_detail: row.get(5)?,
323                session_id: row.get(6)?,
324                created_at: row.get(7)?,
325            })
326        }) {
327            Ok(v) => Ok(Some(v)),
328            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
329            Err(e) => Err(e.into()),
330        }
331    }
332
333    /// Every version of every file of `agent_id`, with content, oldest
334    /// first — the one-time import into the agent's memory record
335    /// (SPEC_MEMORY_FOLLOWS_THE_AGENT_2026_09_24.md §2.1.1).
336    pub fn agent_native_memory_versions_for_agent(&self, agent_id: &str) -> Result<Vec<NativeMemoryVersion>, StoreError> {
337        let conn = self.conn.lock().unwrap();
338        let mut stmt = conn.prepare(
339            "SELECT id, filename, content, content_hash, parent_version_id, source, source_detail, session_id, created_at
340             FROM db_agent_native_memory_versions
341             WHERE agent_id = ?1
342             ORDER BY created_at ASC, rowid ASC",
343        )?;
344        let rows = stmt
345            .query_map(params![agent_id], |row| {
346                Ok(NativeMemoryVersion {
347                    id: row.get(0)?,
348                    agent_id: agent_id.to_string(),
349                    filename: row.get(1)?,
350                    content: row.get(2)?,
351                    content_hash: row.get(3)?,
352                    parent_version_id: row.get(4)?,
353                    source: row.get(5)?,
354                    source_detail: row.get(6)?,
355                    session_id: row.get(7)?,
356                    created_at: row.get(8)?,
357                })
358            })?
359            .collect::<Result<Vec<_>, _>>()?;
360        Ok(rows)
361    }
362
363    /// Every distinct `(agent_id, filename)` pair with at least one recorded
364    /// version — the retention sweep's (`native_memory_retention.rs`) work
365    /// list, so it doesn't need its own separate source of "which files
366    /// have history" (the version table itself is authoritative for that).
367    pub fn agent_native_memory_version_list_distinct_files(&self) -> Result<Vec<(String, String)>, StoreError> {
368        let conn = self.conn.lock().unwrap();
369        let mut stmt = conn.prepare("SELECT DISTINCT agent_id, filename FROM db_agent_native_memory_versions")?;
370        let rows = stmt
371            .query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))?
372            .collect::<Result<Vec<_>, _>>()?;
373        Ok(rows)
374    }
375
376    /// Prune old versions of `(agent_id, filename)` per the hybrid
377    /// age + min-count-floor retention policy
378    /// (`native_memory_retention.rs`, spec §7.1): a version is deleted only
379    /// if it is BOTH older than `max_age_ms` AND ranked beyond the
380    /// `min_keep` most-recent versions for this file — so a rarely-touched
381    /// file never loses its entire history purely because every version
382    /// happens to be old, and a hyperactive file never grows unbounded
383    /// purely because every version happens to be recent. Returns the
384    /// number of rows deleted.
385    /// Repair the mislabeled rollout backfill — see
386    /// `migrations::m0024_native_memory_backfill_from_fs` for the full
387    /// story. In short: `m0023` was supposed to tag every pre-existing
388    /// memory file `agent_inferred`, but it iterated the
389    /// `db_agent_native_memory` MIRROR, which is populated lazily and is
390    /// empty on a typical install — so it backfilled nothing, and the
391    /// drift sweep (which reads the real filesystem) claimed those same
392    /// files milliseconds later as `external_fs_write`, i.e. "detected
393    /// outside AgentMux — provenance unknown". Innocent files, alarming
394    /// label.
395    ///
396    /// The predicate is deliberately narrow, because relabeling a
397    /// provenance claim is not something to do speculatively. All four
398    /// must hold:
399    ///
400    /// - `source = 'external_fs_write'` and `detected_via =
401    ///   'reconciliation_sweep'` — the slow path, which is the only one
402    ///   that can observe a file it has never seen before at startup;
403    /// - `parent_version_id IS NULL` — a FIRST-EVER version for that
404    ///   `(agent_id, filename)`, i.e. genuine first sight rather than an
405    ///   observed change to something already tracked;
406    /// - `created_at` inside a short window starting at `m0023`'s own
407    ///   `applied_at` — the load-bearing clause, since a memory file
408    ///   created out-of-band next week is also first-sight-via-sweep and
409    ///   that one is a real external write whose label must stand.
410    ///
411    /// ## The limit of the window, stated plainly
412    ///
413    /// Inside the window this cannot tell the rollout burst from a
414    /// genuinely new out-of-band file, and will relabel both (ReAgent P2
415    /// on PR #3055 — an earlier revision of this comment claimed an
416    /// absolute guarantee it does not deliver). Nothing in the row or the
417    /// file distinguishes them: mtime is no help either, since a burst
418    /// row's file may have been legitimately edited since.
419    ///
420    /// What makes the trade acceptable is the shape of the two errors, not
421    /// their likelihood alone. Getting it wrong costs one just-created file
422    /// a benign "Agent" label instead of "detected outside AgentMux", on an
423    /// install where the user is watching a fresh upgrade. Not acting costs
424    /// every pre-existing memory file on every affected install a permanent
425    /// security-flavored warning it did not earn — which is the bug this
426    /// repairs. The window is kept as small as the evidence allows so the
427    /// ambiguous region stays tiny.
428    ///
429    /// `source_detail` keeps the original `detected_via` and records that
430    /// the row was relabeled, so the audit trail says what happened rather
431    /// than quietly presenting the new claim as if it were the original
432    /// observation.
433    ///
434    /// Returns the number of rows relabeled. A no-op (returns 0) when
435    /// `m0023` never ran — nothing to repair on a fresh install.
436    pub fn agent_native_memory_version_relabel_rollout_first_sight(
437        &self,
438    ) -> Result<usize, StoreError> {
439        // The sweep's `tokio::time::interval` fires immediately on its
440        // first tick, so the burst lands milliseconds after migrations
441        // finish — 61ms on the install this was diagnosed from. A minute is
442        // three orders of magnitude of slack for a loaded machine's first
443        // sweep, while keeping the ambiguous region (see above) short.
444        const ROLLOUT_WINDOW_MS: i64 = 60 * 1000;
445        let conn = self.conn.lock().unwrap();
446        // strftime() parses m0023's ISO-8601 `applied_at` (offset included)
447        // to epoch seconds; the version table stores epoch millis.
448        let updated = conn.execute(
449            "UPDATE db_agent_native_memory_versions
450                SET source = 'agent_inferred',
451                    source_detail = json_object(
452                        'relabeled_from', 'external_fs_write',
453                        'detected_via', json_extract(source_detail, '$.detected_via'),
454                        'reason', 'rollout_first_sight'
455                    )
456              WHERE source = 'external_fs_write'
457                AND json_extract(source_detail, '$.detected_via') = 'reconciliation_sweep'
458                AND parent_version_id IS NULL
459                AND created_at >= (
460                    SELECT strftime('%s', applied_at) * 1000 FROM db_migrations
461                     WHERE id = '0023_native_memory_versions_backfill')
462                AND created_at <= (
463                    SELECT strftime('%s', applied_at) * 1000 + ?1 FROM db_migrations
464                     WHERE id = '0023_native_memory_versions_backfill')",
465            params![ROLLOUT_WINDOW_MS],
466        )?;
467        Ok(updated)
468    }
469
470    pub fn agent_native_memory_version_prune(
471        &self,
472        agent_id: &str,
473        filename: &str,
474        min_keep: u32,
475        max_age_ms: i64,
476    ) -> Result<usize, StoreError> {
477        let cutoff = now_ms() - max_age_ms;
478        let conn = self.conn.lock().unwrap();
479        let deleted = conn.execute(
480            "DELETE FROM db_agent_native_memory_versions
481             WHERE id IN (
482                 SELECT id FROM (
483                     SELECT id, created_at,
484                            ROW_NUMBER() OVER (ORDER BY created_at DESC, rowid DESC) AS rn
485                     FROM db_agent_native_memory_versions
486                     WHERE agent_id = ?1 AND filename = ?2
487                 )
488                 WHERE rn > ?3 AND created_at < ?4
489             )",
490            params![agent_id, filename, min_keep, cutoff],
491        )?;
492        Ok(deleted)
493    }
494
495    /// Read one version's full content by id — for diff/revert, which need
496    /// exactly one or two full bodies, not a whole file's history.
497    pub fn agent_native_memory_version_get(
498        &self,
499        version_id: &str,
500    ) -> Result<Option<NativeMemoryVersion>, StoreError> {
501        let conn = self.conn.lock().unwrap();
502        let mut stmt = conn.prepare(
503            "SELECT id, agent_id, filename, content, content_hash, parent_version_id, source, source_detail, session_id, created_at
504             FROM db_agent_native_memory_versions
505             WHERE id = ?1",
506        )?;
507        match stmt.query_row(params![version_id], |row| {
508            Ok(NativeMemoryVersion {
509                id: row.get(0)?,
510                agent_id: row.get(1)?,
511                filename: row.get(2)?,
512                content: row.get(3)?,
513                content_hash: row.get(4)?,
514                parent_version_id: row.get(5)?,
515                source: row.get(6)?,
516                source_detail: row.get(7)?,
517                session_id: row.get(8)?,
518                created_at: row.get(9)?,
519            })
520        }) {
521            Ok(v) => Ok(Some(v)),
522            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
523            Err(e) => Err(e.into()),
524        }
525    }
526}
527
528#[cfg(test)]
529mod tests {
530    use super::*;
531
532    fn shared_store() -> Store {
533        Store::open_shared(std::path::Path::new(":memory:")).unwrap()
534    }
535
536    /// Record that `m0023` ran at `applied_at` (ISO-8601). The relabel
537    /// window is anchored to this row, so it is what makes a version
538    /// "part of the rollout burst" or not.
539    fn record_m0023(store: &Store, applied_at: &str) {
540        store
541            .conn
542            .lock()
543            .unwrap()
544            .execute(
545                "INSERT INTO db_migrations (id, applied_at, duration_ms, scope)
546                 VALUES ('0023_native_memory_versions_backfill', ?1, 0, 'global')",
547                params![applied_at],
548            )
549            .unwrap();
550    }
551
552    /// An ISO-8601 stamp for "now", so versions inserted by a test land
553    /// inside the rollout window.
554    fn now_iso() -> String {
555        let secs = agentmux_common::time::now_secs_u64();
556        let conn = rusqlite::Connection::open_in_memory().unwrap();
557        conn.query_row(
558            "SELECT strftime('%Y-%m-%dT%H:%M:%S+00:00', ?1, 'unixepoch')",
559            params![secs as i64],
560            |r| r.get::<_, String>(0),
561        )
562        .unwrap()
563    }
564
565    /// The sweep's own row shape for a file it has never seen before.
566    fn sweep_first_sight(store: &Store, agent: &str, file: &str) {
567        store
568            .agent_native_memory_version_insert(
569                agent, file, "pre-existing content", "external_fs_write",
570                r#"{"detected_via":"reconciliation_sweep"}"#, "",
571            )
572            .unwrap();
573    }
574
575    #[test]
576    fn relabels_the_rollout_first_sight_burst() {
577        // The real-world case: m0023 backfilled nothing (empty mirror), and
578        // the sweep claimed the pre-existing files milliseconds later.
579        let store = shared_store();
580        record_m0023(&store, &now_iso());
581        sweep_first_sight(&store, "agent-1", "MEMORY.md");
582
583        assert_eq!(store.agent_native_memory_version_relabel_rollout_first_sight().unwrap(), 1);
584
585        let v = &store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap()[0];
586        assert_eq!(v.source, "agent_inferred");
587        // The audit trail must still say what actually happened.
588        assert!(v.source_detail.contains("relabeled_from"), "{}", v.source_detail);
589        assert!(v.source_detail.contains("reconciliation_sweep"), "{}", v.source_detail);
590    }
591
592    #[test]
593    fn leaves_a_genuine_external_write_outside_the_window_alone() {
594        // A memory file created out-of-band after the upgrade is also
595        // first-sight-via-sweep. Once it is outside the window that one is a
596        // real external write and its label must stand — the single most
597        // important thing this repair must not get wrong.
598        let store = shared_store();
599        record_m0023(&store, "2020-01-01T00:00:00+00:00");
600        sweep_first_sight(&store, "agent-1", "MEMORY.md");
601
602        assert_eq!(store.agent_native_memory_version_relabel_rollout_first_sight().unwrap(), 0);
603        assert_eq!(
604            store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap()[0].source,
605            "external_fs_write"
606        );
607    }
608
609    #[test]
610    fn relabels_a_genuinely_new_file_inside_the_window_too() {
611        // Pins a KNOWN, accepted limitation rather than a desired property
612        // (ReAgent P2 on PR #3055): inside the window the burst and a
613        // genuinely new out-of-band file are indistinguishable, so this one
614        // is relabeled too. Documented in the method's own doc comment; the
615        // mitigation is that the window is small, not that the case cannot
616        // happen. Here so nobody reads the guarantee as absolute, and so
617        // narrowing it later is a deliberate change with a failing test
618        // rather than a silent one.
619        let store = shared_store();
620        record_m0023(&store, &now_iso());
621        sweep_first_sight(&store, "agent-1", "BRAND-NEW.md");
622
623        assert_eq!(store.agent_native_memory_version_relabel_rollout_first_sight().unwrap(), 1);
624    }
625
626    #[test]
627    fn leaves_an_observed_change_to_a_tracked_file_alone() {
628        // Not first sight: the file was already versioned, so the sweep
629        // observed a real out-of-band CHANGE. parent_version_id is set, and
630        // the row must keep its label.
631        let store = shared_store();
632        record_m0023(&store, &now_iso());
633        store
634            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v1", "human", "{}", "")
635            .unwrap();
636        sweep_first_sight(&store, "agent-1", "MEMORY.md");
637
638        assert_eq!(store.agent_native_memory_version_relabel_rollout_first_sight().unwrap(), 0);
639    }
640
641    #[test]
642    fn leaves_fs_watch_detections_alone() {
643        // Only the slow sweep can observe a never-before-seen file at
644        // startup; a live fs-watch event means something wrote while we
645        // were watching.
646        let store = shared_store();
647        record_m0023(&store, &now_iso());
648        store
649            .agent_native_memory_version_insert(
650                "agent-1", "MEMORY.md", "c", "external_fs_write",
651                r#"{"detected_via":"fs_watch"}"#, "",
652            )
653            .unwrap();
654
655        assert_eq!(store.agent_native_memory_version_relabel_rollout_first_sight().unwrap(), 0);
656    }
657
658    #[test]
659    fn is_a_no_op_when_m0023_never_ran() {
660        // Fresh install: db_migrations exists but has no m0023 row, so the
661        // window subquery is NULL and nothing matches. Must not error.
662        let store = shared_store();
663        sweep_first_sight(&store, "agent-1", "MEMORY.md");
664        assert_eq!(store.agent_native_memory_version_relabel_rollout_first_sight().unwrap(), 0);
665    }
666
667    #[test]
668    fn is_idempotent() {
669        let store = shared_store();
670        record_m0023(&store, &now_iso());
671        sweep_first_sight(&store, "agent-1", "MEMORY.md");
672        assert_eq!(store.agent_native_memory_version_relabel_rollout_first_sight().unwrap(), 1);
673        // Second run finds nothing: the rows it would match are no longer
674        // external_fs_write.
675        assert_eq!(store.agent_native_memory_version_relabel_rollout_first_sight().unwrap(), 0);
676    }
677
678    #[test]
679    fn first_write_has_no_parent() {
680        let store = shared_store();
681        let v = store
682            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v1", "human", "{}", "sess-1")
683            .unwrap();
684        assert_eq!(v.parent_version_id, None);
685        assert_eq!(v.content, "v1");
686        assert_eq!(v.content_hash, content_hash("v1"));
687    }
688
689    #[test]
690    fn second_write_chains_onto_the_first() {
691        let store = shared_store();
692        let v1 = store
693            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v1", "human", "{}", "")
694            .unwrap();
695        let v2 = store
696            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v2", "agent_inferred", "{}", "")
697            .unwrap();
698        assert_eq!(v2.parent_version_id, Some(v1.id));
699    }
700
701    #[test]
702    fn a_no_op_write_still_records_a_new_version() {
703        // Simplicity over cleverness (spec §4.1's own test-plan note): two
704        // writes with byte-identical content still produce two version rows,
705        // chained onto each other — no dedup-by-hash in v1.
706        let store = shared_store();
707        let v1 = store
708            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "same", "human", "{}", "")
709            .unwrap();
710        let v2 = store
711            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "same", "human", "{}", "")
712            .unwrap();
713        assert_ne!(v1.id, v2.id);
714        assert_eq!(v2.parent_version_id, Some(v1.id));
715        assert_eq!(v1.content_hash, v2.content_hash);
716    }
717
718    // Regression for reagent P2 on PR #2674 (second re-review): an
719    // in-process Mutex only serializes callers sharing one Store/Connection
720    // — it does nothing for two SEPARATE connections to the same
721    // shared-store sqlite file (the real-world case: two srv processes,
722    // e.g. two AgentMux channels/instances, both writing the same agent's
723    // same memory file at close to the same moment). Opens the SAME
724    // on-disk file via two independent `Store` instances (simulating two
725    // processes) and fires one truly-concurrent insert from each — without
726    // the `BEGIN IMMEDIATE` fix, both connections' SELECT-latest could read
727    // the same pre-existing "latest" and each insert a version with that
728    // same parent, instead of one correctly chaining onto the other.
729    //
730    // Run several rounds rather than one shot — a race that depends on
731    // exact timing is more convincing (and less likely to pass by luck on
732    // a fast machine) when it holds up repeatedly, not just once.
733    #[test]
734    fn concurrent_inserts_from_two_separate_connections_form_a_linear_chain() {
735        let tmp = tempfile::NamedTempFile::new().unwrap();
736        let path = tmp.path().to_path_buf();
737        let store_a = std::sync::Arc::new(Store::open_shared(&path).unwrap());
738        let store_b = std::sync::Arc::new(Store::open_shared(&path).unwrap());
739
740        for round in 0..8 {
741            let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
742            let (ba, bb) = (barrier.clone(), barrier.clone());
743            let (sa, sb) = (store_a.clone(), store_b.clone());
744            let filename = format!("MEMORY-{round}.md");
745            let (fa, fb) = (filename.clone(), filename.clone());
746
747            let ha = std::thread::spawn(move || {
748                ba.wait();
749                sa.agent_native_memory_version_insert("agent-1", &fa, "A", "human", "{}", "").unwrap()
750            });
751            let hb = std::thread::spawn(move || {
752                bb.wait();
753                sb.agent_native_memory_version_insert("agent-1", &fb, "B", "human", "{}", "").unwrap()
754            });
755            let va = ha.join().unwrap();
756            let vb = hb.join().unwrap();
757
758            // Exactly one must chain onto the other — never both `None`
759            // parents (both saw an empty table) and never two unrelated
760            // versions with no parent/child relationship between them.
761            let a_onto_b = va.parent_version_id == Some(vb.id.clone());
762            let b_onto_a = vb.parent_version_id == Some(va.id.clone());
763            assert!(
764                a_onto_b != b_onto_a,
765                "round {round}: exactly one of A/B must chain onto the other — A.parent={:?} B.parent={:?}",
766                va.parent_version_id, vb.parent_version_id,
767            );
768        }
769    }
770
771    #[test]
772    fn list_returns_newest_first() {
773        let store = shared_store();
774        store
775            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v1", "human", "{}", "")
776            .unwrap();
777        store
778            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v2", "human", "{}", "")
779            .unwrap();
780        let list = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
781        assert_eq!(list.len(), 2);
782        // Newest (v2, no children) first.
783        assert!(list[0].content_hash == content_hash("v2"));
784        assert!(list[1].content_hash == content_hash("v1"));
785    }
786
787    #[test]
788    fn list_scopes_to_the_requested_agent_and_filename_only() {
789        let store = shared_store();
790        store
791            .agent_native_memory_version_insert("agent-1", "a.md", "x", "human", "{}", "")
792            .unwrap();
793        store
794            .agent_native_memory_version_insert("agent-1", "b.md", "y", "human", "{}", "")
795            .unwrap();
796        store
797            .agent_native_memory_version_insert("agent-2", "a.md", "z", "human", "{}", "")
798            .unwrap();
799
800        let list = store.agent_native_memory_version_list("agent-1", "a.md").unwrap();
801        assert_eq!(list.len(), 1);
802        assert_eq!(list[0].content_hash, content_hash("x"));
803    }
804
805    #[test]
806    fn list_omits_content_by_design() {
807        let store = shared_store();
808        store
809            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "some content", "human", "{}", "")
810            .unwrap();
811        // NativeMemoryVersionSummary has no `content` field at all — this
812        // test documents that as intentional (compile-time enforced), not
813        // just a runtime assertion. If this ever fails to compile because a
814        // future refactor added a content field, that regressed the
815        // list-view/full-body split agent_native_memory.rs's own docs
816        // established as the pattern to follow.
817        let list = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
818        assert_eq!(list.len(), 1);
819    }
820
821    #[test]
822    fn latest_returns_none_when_no_version_exists() {
823        let store = shared_store();
824        assert_eq!(
825            store.agent_native_memory_version_latest("agent-1", "MEMORY.md").unwrap(),
826            None
827        );
828    }
829
830    #[test]
831    fn get_reads_a_specific_version_by_id_with_full_content() {
832        let store = shared_store();
833        let v1 = store
834            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v1", "human", "{}", "")
835            .unwrap();
836        store
837            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v2", "human", "{}", "")
838            .unwrap();
839
840        // get() must return v1's content even though it's no longer latest.
841        let got = store.agent_native_memory_version_get(&v1.id).unwrap();
842        assert_eq!(got, Some(v1));
843    }
844
845    #[test]
846    fn get_returns_none_for_an_unknown_id() {
847        let store = shared_store();
848        assert_eq!(store.agent_native_memory_version_get("nope").unwrap(), None);
849    }
850
851    #[test]
852    fn source_and_source_detail_round_trip_exactly() {
853        let store = shared_store();
854        let detail = r#"{"FROM":"github-consumer","TIER":"sensitive","TRUST":"network-claimed"}"#;
855        let v = store
856            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "content", "jekt", detail, "sess-42")
857            .unwrap();
858        assert_eq!(v.source, "jekt");
859        assert_eq!(v.source_detail, detail);
860        assert_eq!(v.session_id, "sess-42");
861
862        let list = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
863        assert_eq!(list[0].source, "jekt");
864        assert_eq!(list[0].source_detail, detail);
865    }
866
867    #[test]
868    fn insert_if_changed_inserts_on_first_observation() {
869        let store = shared_store();
870        let v = store
871            .agent_native_memory_version_insert_if_changed("agent-1", "MEMORY.md", "content", "external_fs_write", "{}", "")
872            .unwrap();
873        assert!(v.is_some());
874    }
875
876    #[test]
877    fn insert_if_changed_is_a_no_op_for_identical_content() {
878        let store = shared_store();
879        store
880            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "same", "human", "{}", "")
881            .unwrap();
882        let result = store
883            .agent_native_memory_version_insert_if_changed("agent-1", "MEMORY.md", "same", "external_fs_write", "{}", "")
884            .unwrap();
885        assert_eq!(result, None);
886        assert_eq!(store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap().len(), 1);
887    }
888
889    #[test]
890    fn insert_if_changed_inserts_for_different_content() {
891        let store = shared_store();
892        let v1 = store
893            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "v1", "human", "{}", "")
894            .unwrap();
895        let v2 = store
896            .agent_native_memory_version_insert_if_changed("agent-1", "MEMORY.md", "v2", "external_fs_write", "{}", "")
897            .unwrap()
898            .expect("different content must insert");
899        assert_eq!(v2.parent_version_id, Some(v1.id));
900    }
901
902    #[test]
903    fn insert_if_changed_only_the_first_of_two_racing_identical_calls_inserts() {
904        // Regression for reagent P2 on PR #2675: the fast path and slow path
905        // can both observe the same drift concurrently. Simulating that
906        // race directly (rather than with real concurrency, which SQLite's
907        // single-writer-lock would serialize anyway) — the point of this
908        // test is the *logical* compare-and-swap: two sequential calls with
909        // the SAME new content must produce exactly one version, not two.
910        let store = shared_store();
911        let first = store
912            .agent_native_memory_version_insert_if_changed("agent-1", "MEMORY.md", "drifted content", "external_fs_write", "{}", "")
913            .unwrap();
914        let second = store
915            .agent_native_memory_version_insert_if_changed("agent-1", "MEMORY.md", "drifted content", "external_fs_write", "{}", "")
916            .unwrap();
917        assert!(first.is_some());
918        assert_eq!(second, None, "a second call observing the same already-recorded content must be a no-op");
919        assert_eq!(store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap().len(), 1);
920    }
921
922    const DAY_MS: i64 = 24 * 60 * 60 * 1000;
923
924    /// Test-only helper: backdate a version's `created_at` directly (the
925    /// public insert API always stamps `now_ms()` — retention policy can
926    /// only be exercised against real elapsed time otherwise, which would
927    /// make these tests either slow or flaky).
928    fn backdate(store: &Store, version_id: &str, days_ago: i64) {
929        let conn = store.conn.lock().unwrap();
930        conn.execute(
931            "UPDATE db_agent_native_memory_versions SET created_at = ?1 WHERE id = ?2",
932            params![now_ms() - days_ago * DAY_MS, version_id],
933        )
934        .unwrap();
935    }
936
937    #[test]
938    fn prune_leaves_a_file_with_fewer_versions_than_the_floor_untouched() {
939        let store = shared_store();
940        for i in 0..5 {
941            let v = store
942                .agent_native_memory_version_insert("agent-1", "MEMORY.md", &format!("v{i}"), "human", "{}", "")
943                .unwrap();
944            backdate(&store, &v.id, 1000); // ancient, but under the count floor
945        }
946        let deleted = store.agent_native_memory_version_prune("agent-1", "MEMORY.md", 50, 90 * DAY_MS).unwrap();
947        assert_eq!(deleted, 0);
948        assert_eq!(store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap().len(), 5);
949    }
950
951    #[test]
952    fn prune_keeps_the_min_keep_newest_versions_even_when_all_are_old() {
953        let store = shared_store();
954        for i in 0..8 {
955            let v = store
956                .agent_native_memory_version_insert("agent-1", "MEMORY.md", &format!("v{i}"), "human", "{}", "")
957                .unwrap();
958            backdate(&store, &v.id, 1000);
959        }
960        let deleted = store.agent_native_memory_version_prune("agent-1", "MEMORY.md", 3, 90 * DAY_MS).unwrap();
961        assert_eq!(deleted, 5, "floor of 3 must survive out of 8, even though every version is ancient");
962        let remaining = store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap();
963        assert_eq!(remaining.len(), 3);
964        // The 3 survivors must be the newest 3 (v5, v6, v7 — insertion order).
965        assert_eq!(remaining[0].content_hash, content_hash("v7"));
966        assert_eq!(remaining[1].content_hash, content_hash("v6"));
967        assert_eq!(remaining[2].content_hash, content_hash("v5"));
968    }
969
970    #[test]
971    fn prune_leaves_a_version_beyond_the_floor_alone_if_it_is_still_young() {
972        let store = shared_store();
973        // 5 versions, all inserted "now" (age 0) — none old enough to prune,
974        // even with a floor of 1 that would otherwise expose 4 of them.
975        for i in 0..5 {
976            store
977                .agent_native_memory_version_insert("agent-1", "MEMORY.md", &format!("v{i}"), "human", "{}", "")
978                .unwrap();
979        }
980        let deleted = store.agent_native_memory_version_prune("agent-1", "MEMORY.md", 1, 90 * DAY_MS).unwrap();
981        assert_eq!(deleted, 0, "young versions must survive regardless of the count floor");
982        assert_eq!(store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap().len(), 5);
983    }
984
985    #[test]
986    fn prune_deletes_only_versions_both_beyond_the_floor_and_past_the_age_cutoff() {
987        let store = shared_store();
988        // 5 old versions (beyond a floor of 2, past the age cutoff) + 2 kept
989        // by the floor (also old, but protected by min_keep) — mixing both
990        // guards in one file to prove they combine with AND, not OR.
991        let mut ids = Vec::new();
992        for i in 0..7 {
993            let v = store
994                .agent_native_memory_version_insert("agent-1", "MEMORY.md", &format!("v{i}"), "human", "{}", "")
995                .unwrap();
996            backdate(&store, &v.id, 100); // past a 90-day cutoff
997            ids.push(v.id);
998        }
999        let deleted = store.agent_native_memory_version_prune("agent-1", "MEMORY.md", 2, 90 * DAY_MS).unwrap();
1000        assert_eq!(deleted, 5);
1001        assert_eq!(store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap().len(), 2);
1002    }
1003
1004    #[test]
1005    fn prune_scopes_to_the_requested_agent_and_filename_only() {
1006        let store = shared_store();
1007        let v_a = store
1008            .agent_native_memory_version_insert("agent-1", "MEMORY.md", "a", "human", "{}", "")
1009            .unwrap();
1010        backdate(&store, &v_a.id, 200);
1011        let v_b = store
1012            .agent_native_memory_version_insert("agent-2", "MEMORY.md", "b", "human", "{}", "")
1013            .unwrap();
1014        backdate(&store, &v_b.id, 200);
1015
1016        let deleted = store.agent_native_memory_version_prune("agent-1", "MEMORY.md", 0, 90 * DAY_MS).unwrap();
1017        assert_eq!(deleted, 1);
1018        assert!(store.agent_native_memory_version_list("agent-1", "MEMORY.md").unwrap().is_empty());
1019        assert_eq!(store.agent_native_memory_version_list("agent-2", "MEMORY.md").unwrap().len(), 1, "a different agent's version must be untouched");
1020    }
1021
1022    #[test]
1023    fn list_distinct_files_returns_each_agent_filename_pair_once() {
1024        let store = shared_store();
1025        store.agent_native_memory_version_insert("agent-1", "a.md", "v1", "human", "{}", "").unwrap();
1026        store.agent_native_memory_version_insert("agent-1", "a.md", "v2", "human", "{}", "").unwrap();
1027        store.agent_native_memory_version_insert("agent-1", "b.md", "v1", "human", "{}", "").unwrap();
1028        store.agent_native_memory_version_insert("agent-2", "a.md", "v1", "human", "{}", "").unwrap();
1029
1030        let mut files = store.agent_native_memory_version_list_distinct_files().unwrap();
1031        files.sort();
1032        assert_eq!(
1033            files,
1034            vec![
1035                ("agent-1".to_string(), "a.md".to_string()),
1036                ("agent-1".to_string(), "b.md".to_string()),
1037                ("agent-2".to_string(), "a.md".to_string()),
1038            ]
1039        );
1040    }
1041
1042    #[test]
1043    fn list_distinct_files_is_empty_when_no_versions_exist() {
1044        let store = shared_store();
1045        assert!(store.agent_native_memory_version_list_distinct_files().unwrap().is_empty());
1046    }
1047}
agentmux_srv/backend/storage/agent_native_memory_versions.rs

agentmux_srv/backend/storage/
agent_native_memory_versions.rs
