lean_ctx/core/agents/
shared.rs

1//! Knowledge agents share with the other agents of ONE project.
2//!
3//! Facts are partitioned by the project of the sharing request: each project
4//! has its own file under `agents/shared_knowledge/`, written atomically while
5//! the registry lock is held, and an agent receives only the facts of the
6//! project it is working in. The text is admitted before it is stored and
7//! re-authorized for the reader on every delivery, exactly like a bus message.
8//!
9//! Releases before 3.11.5 kept every shared fact in one installation-wide
10//! `shared_knowledge.json` without a project, so a fact shared in one project
11//! reached agents of every other project. Those facts carry no provable
12//! scope: the file is moved into quarantine unread and never served.
13
14use super::AgentRegistry;
15use super::persistence::{agents_dir, create_agents_dir};
16use crate::core::context_admission::stored::{self, Lineage, StoredBinding};
17use chrono::{DateTime, Utc};
18use serde::{Deserialize, Serialize};
19use std::path::{Path, PathBuf};
20
21const SHARED_SCHEMA_VERSION: u32 = 1;
22const MAX_FACTS_PER_PROJECT: usize = 500;
23
24#[derive(Debug, Clone, Serialize, Deserialize)]
25pub struct SharedFact {
26    pub id: String,
27    pub from_agent: String,
28    pub category: String,
29    /// The admitted `key=value` text.
30    pub text: String,
31    pub timestamp: DateTime<Utc>,
32    #[serde(default)]
33    pub received_by: Vec<String>,
34    #[serde(default)]
35    binding: Option<StoredBinding>,
36}
37
38#[derive(Debug, Serialize, Deserialize)]
39struct SharedFile {
40    schema_version: u32,
41    project_root: String,
42    facts: Vec<SharedFact>,
43}
44
45impl AgentRegistry {
46    /// Share `facts` with the other agents of `project_root`. Every fact is
47    /// admitted first; one withheld fact refuses the whole share and nothing
48    /// is written.
49    pub fn share_knowledge(
50        &mut self,
51        project_root: &str,
52        from: &str,
53        category: &str,
54        facts: &[(String, String)],
55    ) -> Result<usize, String> {
56        self.share(project_root, from, category, facts, None)
57    }
58
59    /// Adopt one legacy fact (`lean-ctx migrate legacy-context`) under the
60    /// stable id `id`: a rerun after an interrupted adoption finds the copy
61    /// and writes nothing twice.
62    pub fn adopt_shared_fact(
63        &mut self,
64        project_root: &str,
65        id: &str,
66        from: &str,
67        category: &str,
68        fact: (String, String),
69    ) -> Result<(), String> {
70        self.share(project_root, from, category, &[fact], Some(id))
71            .map(|_| ())
72    }
73
74    fn share(
75        &mut self,
76        project_root: &str,
77        from: &str,
78        category: &str,
79        facts: &[(String, String)],
80        adoption: Option<&str>,
81    ) -> Result<usize, String> {
82        let mut admitted = Vec::with_capacity(facts.len());
83        for (key, value) in facts {
84            let (text, binding) = stored::admit_and_bind(
85                &format!("{key}={value}"),
86                &format!("agent:{from}"),
87                Some(project_root),
88                Lineage::CallerSupplied,
89            )
90            .map_err(|withheld| withheld.message("shared knowledge"))?;
91            admitted.push(SharedFact {
92                id: adoption.map_or_else(super::persistence::generate_short_id, str::to_owned),
93                from_agent: from.to_string(),
94                category: category.to_string(),
95                text,
96                timestamp: Utc::now(),
97                received_by: Vec::new(),
98                binding: Some(binding),
99            });
100        }
101        let mut file = load_project_file(project_root)?;
102        if adoption.is_some_and(|id| file.facts.iter().any(|fact| fact.id == id)) {
103            return Ok(0);
104        }
105        file.facts.extend(admitted);
106        if file.facts.len() > MAX_FACTS_PER_PROJECT {
107            let excess = file.facts.len() - MAX_FACTS_PER_PROJECT;
108            file.facts.drain(..excess);
109        }
110        save_project_file(&file)?;
111        self.updated_at = Utc::now();
112        Ok(facts.len())
113    }
114
115    /// The facts of `project_root` that `agent_id` has not received yet, each
116    /// re-authorized for this reader now; a withheld fact is delivered as the
117    /// content-free gateway notice.
118    pub fn receive_shared_knowledge(
119        &mut self,
120        project_root: &str,
121        agent_id: &str,
122    ) -> Result<Vec<SharedFact>, String> {
123        let mut file = load_project_file(project_root)?;
124        let mut delivered = Vec::new();
125        for fact in &mut file.facts {
126            if fact.from_agent != agent_id && !fact.received_by.iter().any(|id| id == agent_id) {
127                fact.received_by.push(agent_id.to_string());
128                let mut served = fact.clone();
129                served.text = match stored::serve_stored(
130                    &fact.text,
131                    fact.binding.as_ref(),
132                    Some(project_root),
133                ) {
134                    Ok(text) => {
135                        stored::publish_served(fact.binding.as_ref());
136                        text
137                    }
138                    Err(withheld) => withheld.notice(),
139                };
140                delivered.push(served);
141            }
142        }
143        if !delivered.is_empty() {
144            save_project_file(&file)?;
145        }
146        Ok(delivered)
147    }
148}
149
150fn shared_dir() -> Result<PathBuf, String> {
151    Ok(agents_dir()?.join("shared_knowledge"))
152}
153
154fn project_file_path(project_root: &str) -> Result<PathBuf, String> {
155    Ok(shared_dir()?.join(format!(
156        "{}.json",
157        crate::core::project_hash::hash_project_root(project_root)
158    )))
159}
160
161fn same_root(stored: &str, requested: &str) -> bool {
162    use crate::core::pathutil::{normalize_tool_path, safe_canonicalize_bounded};
163    let canonical = |root: &str| safe_canonicalize_bounded(Path::new(root), 2000);
164    normalize_tool_path(stored).trim_end_matches('/')
165        == normalize_tool_path(requested).trim_end_matches('/')
166        || canonical(stored) == canonical(requested)
167}
168
169fn load_project_file(project_root: &str) -> Result<SharedFile, String> {
170    quarantine_legacy_file()?;
171    let path = project_file_path(project_root)?;
172    let file = match std::fs::read_to_string(&path) {
173        Ok(content) => serde_json::from_str::<SharedFile>(&content).map_err(|error| {
174            format!("shared knowledge is corrupt at {}: {error}", path.display())
175        })?,
176        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
177            if let Some(file) = adopt_drifted_file(project_root)? {
178                return Ok(file);
179            }
180            return Ok(SharedFile {
181                schema_version: SHARED_SCHEMA_VERSION,
182                project_root: project_root.to_string(),
183                facts: Vec::new(),
184            });
185        }
186        Err(error) => return Err(format!("read shared knowledge {}: {error}", path.display())),
187    };
188    if file.schema_version != SHARED_SCHEMA_VERSION {
189        return Err(format!(
190            "shared knowledge at {} has unsupported schema {}",
191            path.display(),
192            file.schema_version
193        ));
194    }
195    // The file name is a hash; the stored root is the authority.
196    if !same_root(&file.project_root, project_root) {
197        return Err("shared knowledge belongs to another project".into());
198    }
199    Ok(file)
200}
201
202/// The project's identity (remote, package name) changed and moved its
203/// hash: a file of an earlier hash whose recorded root is this project's is
204/// its own (proof by the stored root). It is copied, never moved, into the
205/// current partition and journaled once (`legacy_adoption`). Bounded scan.
206fn adopt_drifted_file(project_root: &str) -> Result<Option<SharedFile>, String> {
207    const SCAN_LIMIT: usize = 2048;
208    let current = project_file_path(project_root)?;
209    let Ok(entries) = std::fs::read_dir(shared_dir()?) else {
210        return Ok(None);
211    };
212    let mut candidates: Vec<std::path::PathBuf> = entries
213        .flatten()
214        .take(SCAN_LIMIT)
215        .map(|entry| entry.path())
216        .filter(|path| *path != current && path.extension().is_some_and(|ext| ext == "json"))
217        .collect();
218    candidates.sort();
219    for path in candidates {
220        let Some(file) = std::fs::read_to_string(&path)
221            .ok()
222            .and_then(|content| serde_json::from_str::<SharedFile>(&content).ok())
223        else {
224            continue;
225        };
226        if file.schema_version != SHARED_SCHEMA_VERSION
227            || !same_root(&file.project_root, project_root)
228        {
229            continue;
230        }
231        save_project_file(&file)?;
232        let partition = crate::core::project_hash::hash_project_root(project_root);
233        let stem = path
234            .file_stem()
235            .map(|s| s.to_string_lossy().into_owned())
236            .unwrap_or_default();
237        let _ = crate::core::legacy_adoption::record(
238            "shared-knowledge",
239            &format!("shared-knowledge:{stem}"),
240            &partition,
241            true,
242        );
243        return Ok(Some(file));
244    }
245    Ok(None)
246}
247
248fn save_project_file(file: &SharedFile) -> Result<(), String> {
249    let dir = shared_dir()?;
250    create_agents_dir(&dir)?;
251    let path = project_file_path(&file.project_root)?;
252    let json = serde_json::to_string_pretty(file).map_err(|error| error.to_string())?;
253    crate::config_io::write_atomic(&path, &json)
254        .map_err(|error| format!("persist shared knowledge {}: {error}", path.display()))
255}
256
257fn legacy_file() -> Result<PathBuf, String> {
258    Ok(crate::core::paths::data_dir()?.join("shared_knowledge.json"))
259}
260
261/// Move the installation-wide file of earlier releases into quarantine,
262/// unread. The name carries the content digest, so a second run (or an older
263/// client recreating the file) never overwrites an earlier quarantined copy.
264fn quarantine_legacy_file() -> Result<(), String> {
265    let legacy = legacy_file()?;
266    let bytes = match std::fs::read(&legacy) {
267        Ok(bytes) => bytes,
268        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
269        Err(error) => {
270            return Err(format!(
271                "read legacy shared knowledge {}: {error}",
272                legacy.display()
273            ));
274        }
275    };
276    let dir = shared_dir()?.join("quarantine");
277    create_agents_dir(&dir)?;
278    let digest = blake3::hash(&bytes).to_hex();
279    let target = dir.join(format!("shared_knowledge-{}.json", &digest[..16]));
280    move_into_quarantine(&legacy, &target)
281}
282
283fn move_into_quarantine(legacy: &Path, target: &Path) -> Result<(), String> {
284    #[cfg(test)]
285    if QUARANTINE_FAILS.with(|armed| armed.replace(false)) {
286        return Err("injected crash before the quarantine move".into());
287    }
288    if target.exists() {
289        // The same content is already quarantined.
290        return std::fs::remove_file(legacy)
291            .map_err(|error| format!("remove legacy shared knowledge: {error}"));
292    }
293    std::fs::rename(legacy, target).map_err(|error| {
294        format!(
295            "quarantine legacy shared knowledge {}: {error}",
296            legacy.display()
297        )
298    })
299}
300
301#[cfg(test)]
302thread_local! {
303    /// Test-only: the next quarantine move on this thread fails before it starts.
304    static QUARANTINE_FAILS: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
305}
306
307#[cfg(test)]
308#[path = "shared_tests.rs"]
309mod tests;
lean_ctx/core/agents/shared.rs

lean_ctx/core/agents/
shared.rs
