agentmux_srv/backend/
memory_history_import.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The one-time import of an agent's earlier memory history into its record
5//! (SPEC_MEMORY_FOLLOWS_THE_AGENT_2026_09_24.md §2.1.1, phase M3b).
6//!
7//! Before the record, each channel kept its own history of an agent's memory
8//! files (`db_agent_native_memory_versions`, in that channel's identity
9//! store), so an agent's past was split across every channel it ran in. At
10//! the agent's first reconcile that proves its folder exclusive, the rows
11//! from every channel store on this machine are imported into the record,
12//! deduplicated by content.
13//!
14//! Rows from `agent_inferred` (the m0024 backfill) and `external_fs_write`
15//! (drift) were attributed to agents by a registry lookup that could name
16//! the wrong agent, so they are **unproven**: imported only when the file
17//! of that name in the agent's own proven folder holds that exact content.
18//! The rest wait for the adoption list (M3c).
19//!
20//! History only: nothing here changes a head or touches the folder.
21
22use std::path::PathBuf;
23use std::time::Instant;
24
25use crate::backend::memory_record::{self as record, ImportedVersion};
26use crate::backend::storage::agent_native_memory_versions::NativeMemoryVersion;
27use crate::backend::storage::error::StoreError;
28use crate::backend::storage::filestore::FileStore;
29use crate::backend::storage::store::Store;
30
31/// Sources written by a path that knew which agent it was writing for.
32const PROVEN_SOURCES: &[&str] = &["agent", "armory-ui", "human", "revert", "jekt"];
33
34/// Every store on this machine that may hold per-channel memory history:
35/// each channel's and dev instance's identity store, and the stable
36/// channel's (the shared store).
37fn history_store_paths() -> Vec<PathBuf> {
38    let Some(shared) = crate::registry::resolve_global_shared_root() else {
39        return Vec::new();
40    };
41    history_store_paths_under(&shared)
42}
43
44fn history_store_paths_under(shared: &std::path::Path) -> Vec<PathBuf> {
45    let mut paths = vec![shared.join("store.db")];
46    if let Some(home) = shared.parent() {
47        let children = |dir: PathBuf| -> Vec<PathBuf> {
48            std::fs::read_dir(dir).map(|d| d.flatten().map(|e| e.path()).collect()).unwrap_or_default()
49        };
50        for channel in children(home.join("channels")) {
51            paths.push(channel.join("identity-store.db"));
52        }
53        // A dev instance's store sits at `dev/<branch>/<clone>/`, or at
54        // `dev/<branch>/` itself in the older layout without a clone id.
55        for branch in children(home.join("dev")) {
56            paths.push(branch.join("identity-store.db"));
57            for instance in children(branch) {
58                paths.push(instance.join("identity-store.db"));
59            }
60        }
61    }
62    paths.retain(|p| p.is_file());
63    paths
64}
65
66/// `uid`'s rows from every history store, or `None` if `deadline` passed
67/// before they were all read (the import then waits for a later pass). A
68/// store that can't be opened or read is skipped.
69fn gather(uid: &str, stores: &[PathBuf], deadline: Instant) -> Option<Vec<NativeMemoryVersion>> {
70    let mut rows = Vec::new();
71    for path in stores {
72        if Instant::now() > deadline {
73            return None;
74        }
75        let Ok(store) = Store::open_read_only(path) else { continue };
76        match store.agent_native_memory_versions_for_agent(uid) {
77            Ok(r) => rows.extend(r),
78            Err(e) => tracing::debug!(store = %path.display(), error = %e, "memory import: history unreadable; skipped"),
79        }
80    }
81    Some(rows)
82}
83
84/// The rows to import: proven ones, and unproven ones whose content is what
85/// the agent's own folder holds under that name (`disk_sha`).
86fn select(rows: Vec<NativeMemoryVersion>, disk_sha: impl Fn(&str) -> Option<String>) -> Vec<ImportedVersion> {
87    rows.into_iter()
88        .filter(|r| crate::server::native_memory_handlers::validate_filename(&r.filename).is_ok())
89        .filter(|r| {
90            PROVEN_SOURCES.contains(&r.source.as_str())
91                || disk_sha(&r.filename).as_deref() == Some(record::sha256_hex(r.content.as_bytes()).as_str())
92        })
93        .map(|r| ImportedVersion {
94            file: r.filename,
95            body: r.content.into_bytes(),
96            source_detail: if r.source_detail.is_empty() {
97                format!("imported {}", r.id)
98            } else {
99                format!("imported {}; {}", r.id, r.source_detail)
100            },
101            source: r.source,
102            created_at_ms: r.created_at,
103        })
104        .collect()
105}
106
107/// Import `uid`'s earlier history, once, from `stores` (every history store
108/// on the machine when `None`). `disk_sha` gives the content hash of a file
109/// in the agent's proven folder. Returns how many versions were imported,
110/// or `None` when there was nothing to do (already imported, or out of
111/// time — retried next pass).
112pub(crate) fn import_once(
113    fs: &FileStore,
114    uid: &str,
115    stores: Option<&[PathBuf]>,
116    disk_sha: impl Fn(&str) -> Option<String>,
117    deadline: Instant,
118) -> Result<Option<usize>, StoreError> {
119    if record::heads(fs, uid)?.imported_at_ms.is_some() {
120        return Ok(None);
121    }
122    let found;
123    let stores = match stores {
124        Some(s) => s,
125        None => {
126            found = history_store_paths();
127            &found
128        }
129    };
130    let Some(rows) = gather(uid, stores, deadline) else {
131        return Ok(None);
132    };
133    record::import_history(fs, uid, &select(rows, disk_sha))
134}
135
136#[cfg(test)]
137mod tests;
agentmux_srv/backend/memory_history_import.rs

agentmux_srv/backend/
memory_history_import.rs
