lean_ctx/core/knowledge/
journal.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Change journal of a project's knowledge.
3//!
4//! Every save of the store appends one row per fact it added, changed or
5//! removed, numbered by a log sequence number (LSN) that only grows. A
6//! reader asks "what changed since LSN n" instead of reloading and diffing
7//! the whole store, and a session can be shown the facts it touched.
8//!
9//! Rows carry the fact's identity (category, key with secrets redacted,
10//! creation time), the names of the fields that changed and the writing
11//! session, never a value: the store holds the values, and a row is shown
12//! with its key only while that fact is visible in the reader's admitted
13//! view (`ctx_knowledge changes`). Local only; nothing is sent anywhere.
14//!
15//! The journal sits beside `knowledge.json` and is written under the same
16//! project lock as the store, so its order is the order of the saves. It
17//! keeps the newest rows within `MAX_JOURNAL_BYTES`; LSNs continue across
18//! a trim.
19
20use std::io::{Read, Seek, SeekFrom, Write};
21use std::path::Path;
22
23use serde::{Deserialize, Serialize};
24
25use super::types::KnowledgeFact;
26
27pub(crate) const JOURNAL_FILE: &str = "journal.jsonl";
28/// Size at which the oldest half of the rows is dropped.
29const MAX_JOURNAL_BYTES: u64 = 2 * 1024 * 1024;
30/// Bytes read from the end of the journal to find the newest LSN.
31const TAIL_BYTES: u64 = 4096;
32/// Fields whose change is a change of the fact. Read counters, feedback
33/// tallies and confirmation stamps change on every use and are not journaled.
34const TRACKED_FIELDS: [&str; 8] = [
35    "value",
36    "status",
37    "validity",
38    "anchors",
39    "review",
40    "valid_until",
41    "supersedes",
42    "privacy",
43];
44
45/// One journaled change of one fact.
46#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
47pub struct JournalRow {
48    pub lsn: u64,
49    pub at: String,
50    /// `added`, `changed` or `removed`.
51    pub op: String,
52    pub category: String,
53    pub key: String,
54    /// The fact's creation time: with category and key, its identity (a
55    /// superseded fact and its successor share category and key).
56    pub created: String,
57    #[serde(default, skip_serializing_if = "Vec::is_empty")]
58    pub fields: Vec<String>,
59    pub session: String,
60}
61
62/// A change before it is numbered.
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub struct Change {
65    pub op: &'static str,
66    pub category: String,
67    pub key: String,
68    pub created: String,
69    pub fields: Vec<String>,
70    pub session: String,
71}
72
73fn identity(fact: &KnowledgeFact) -> (String, String, String) {
74    (
75        fact.category.clone(),
76        fact.key.clone(),
77        fact.created_at.to_rfc3339(),
78    )
79}
80
81fn tracked(fact: &KnowledgeFact) -> Vec<serde_json::Value> {
82    let value = serde_json::to_value(fact).unwrap_or_default();
83    TRACKED_FIELDS
84        .iter()
85        .map(|field| value.get(field).cloned().unwrap_or_default())
86        .collect()
87}
88
89/// The changes from `before` to `after`, in identity order.
90pub fn diff(before: &[KnowledgeFact], after: &[KnowledgeFact]) -> Vec<Change> {
91    use std::collections::BTreeMap;
92    let index = |facts: &[KnowledgeFact]| -> BTreeMap<(String, String, String), usize> {
93        facts
94            .iter()
95            .enumerate()
96            .map(|(i, fact)| (identity(fact), i))
97            .collect()
98    };
99    let (old, new) = (index(before), index(after));
100    let change = |op, fact: &KnowledgeFact, fields: Vec<String>| Change {
101        op,
102        category: fact.category.clone(),
103        key: crate::core::redaction::redact_text(&fact.key),
104        created: fact.created_at.to_rfc3339(),
105        fields,
106        session: fact.source_session.clone(),
107    };
108    let mut changes = Vec::new();
109    for (id, &i) in &new {
110        let fact = &after[i];
111        match old.get(id) {
112            None => changes.push(change("added", fact, Vec::new())),
113            Some(&j) => {
114                let fields: Vec<String> = tracked(&before[j])
115                    .iter()
116                    .zip(tracked(fact))
117                    .zip(TRACKED_FIELDS)
118                    .filter(|((was, now), _)| *was != now)
119                    .map(|(_, name)| name.to_owned())
120                    .collect();
121                if !fields.is_empty() {
122                    changes.push(change("changed", fact, fields));
123                }
124            }
125        }
126    }
127    for (id, &j) in &old {
128        if !new.contains_key(id) {
129            changes.push(change("removed", &before[j], Vec::new()));
130        }
131    }
132    changes
133        .sort_by(|a, b| (&a.category, &a.key, &a.created).cmp(&(&b.category, &b.key, &b.created)));
134    changes
135}
136
137/// Append `changes` under the next LSNs. The caller holds the project's
138/// knowledge lock. Best effort: a journal that cannot be written never fails
139/// the save it describes.
140pub(crate) fn append(dir: &Path, changes: &[Change], at: &str) {
141    if changes.is_empty() {
142        return;
143    }
144    let path = dir.join(JOURNAL_FILE);
145    let first = head_lsn(dir) + 1;
146    let mut text = String::new();
147    for (lsn, change) in (first..).zip(changes) {
148        let row = JournalRow {
149            lsn,
150            at: at.to_owned(),
151            op: change.op.to_owned(),
152            category: change.category.clone(),
153            key: change.key.clone(),
154            created: change.created.clone(),
155            fields: change.fields.clone(),
156            session: change.session.clone(),
157        };
158        if let Ok(line) = serde_json::to_string(&row) {
159            text.push_str(&line);
160            text.push('\n');
161        }
162    }
163    let appended = crate::core::policy::diagnostics::open_append(&path)
164        .and_then(|mut file| file.write_all(text.as_bytes()));
165    if appended.is_err() {
166        tracing::debug!("[knowledge] journal append failed");
167        return;
168    }
169    trim(&path);
170}
171
172/// The newest LSN, 0 for an empty or missing journal.
173pub fn head_lsn(dir: &Path) -> u64 {
174    let Ok(mut file) = std::fs::File::open(dir.join(JOURNAL_FILE)) else {
175        return 0;
176    };
177    let len = file.metadata().map_or(0, |m| m.len());
178    let start = len.saturating_sub(TAIL_BYTES);
179    if file.seek(SeekFrom::Start(start)).is_err() {
180        return 0;
181    }
182    let mut tail = String::new();
183    if file.read_to_string(&mut tail).is_err() {
184        return 0;
185    }
186    tail.lines()
187        .rev()
188        .find_map(|line| serde_json::from_str::<JournalRow>(line).ok())
189        .map_or(0, |row| row.lsn)
190}
191
192/// The rows after `since`, oldest first, at most `limit`.
193pub fn rows_since(dir: &Path, since: u64, limit: usize) -> Vec<JournalRow> {
194    let Ok(text) = std::fs::read_to_string(dir.join(JOURNAL_FILE)) else {
195        return Vec::new();
196    };
197    text.lines()
198        .filter_map(|line| serde_json::from_str::<JournalRow>(line).ok())
199        .filter(|row| row.lsn > since)
200        .take(limit)
201        .collect()
202}
203
204/// Keep the journal within [`MAX_JOURNAL_BYTES`] by dropping its older half.
205fn trim(path: &Path) {
206    if std::fs::metadata(path).map_or(0, |m| m.len()) <= MAX_JOURNAL_BYTES {
207        return;
208    }
209    let Ok(text) = std::fs::read_to_string(path) else {
210        return;
211    };
212    let lines: Vec<&str> = text.lines().collect();
213    let kept = lines[lines.len() / 2..].join("\n") + "\n";
214    let _ = crate::core::atomic_fs::try_atomic_write(path, kept.as_bytes(), None);
215}
216
217#[cfg(test)]
218mod tests {
219    use super::*;
220
221    fn fact(category: &str, key: &str, value: &str, created: &str) -> KnowledgeFact {
222        serde_json::from_value(serde_json::json!({
223            "category": category, "key": key, "value": value,
224            "source_session": "s1", "confidence": 0.9,
225            "created_at": created, "last_confirmed": created,
226        }))
227        .unwrap()
228    }
229
230    const T0: &str = "2026-10-31T10:00:00Z";
231    const T1: &str = "2026-10-31T11:00:00Z";
232
233    #[test]
234    fn a_diff_names_added_changed_and_removed_facts_by_identity() {
235        let before = vec![
236            fact("decision", "db", "postgres", T0),
237            fact("decision", "queue", "redis", T0),
238            fact("convention", "naming", "snake", T0),
239        ];
240        let mut changed = fact("decision", "db", "sqlite", T0);
241        changed.retrieval_count = 9; // a read counter is not a change
242        let mut read_only = fact("convention", "naming", "snake", T0);
243        read_only.retrieval_count = 3;
244        let after = vec![changed, read_only, fact("decision", "db", "duckdb", T1)];
245        let changes = diff(&before, &after);
246        let summary: Vec<(&str, &str, &[String])> = changes
247            .iter()
248            .map(|c| (c.op, c.key.as_str(), c.fields.as_slice()))
249            .collect();
250        assert_eq!(
251            summary,
252            vec![
253                ("changed", "db", &["value".to_owned()][..]),
254                ("added", "db", &[][..]),
255                ("removed", "queue", &[][..]),
256            ]
257        );
258    }
259
260    #[test]
261    fn rows_are_numbered_on_and_read_back_after_an_lsn() {
262        let dir = tempfile::tempdir().unwrap();
263        assert_eq!(head_lsn(dir.path()), 0);
264        let changes = diff(
265            &[],
266            &[
267                fact("decision", "db", "pg", T0),
268                fact("decision", "q", "r", T0),
269            ],
270        );
271        append(dir.path(), &changes, T0);
272        append(
273            dir.path(),
274            &diff(&[fact("decision", "db", "pg", T0)], &[]),
275            T1,
276        );
277        assert_eq!(head_lsn(dir.path()), 3);
278        let rows = rows_since(dir.path(), 1, 10);
279        assert_eq!(rows.iter().map(|r| r.lsn).collect::<Vec<_>>(), vec![2, 3]);
280        assert_eq!(rows[1].op, "removed");
281        assert_eq!(rows_since(dir.path(), 0, 1).len(), 1);
282        // No row carries a value.
283        let raw = std::fs::read_to_string(dir.path().join(JOURNAL_FILE)).unwrap();
284        assert!(!raw.contains("\"pg\""));
285    }
286
287    #[test]
288    fn a_secret_in_a_key_is_redacted_in_the_journal() {
289        let key = concat!("deploy-", "gh", "p_", "Zq8LmW3vXc9TbN2rKp7YdF4sHj6Ga1Ue");
290        let changes = diff(&[], &[fact("decision", key, "v", T0)]);
291        assert!(
292            !changes[0].key.contains("Zq8LmW3vXc9TbN2rKp7Y"),
293            "{}",
294            changes[0].key
295        );
296    }
297
298    #[test]
299    fn a_trimmed_journal_keeps_its_newest_rows_and_numbering() {
300        let dir = tempfile::tempdir().unwrap();
301        let path = dir.path().join(JOURNAL_FILE);
302        let row = |lsn: u64| {
303            serde_json::to_string(&JournalRow {
304                lsn,
305                at: T0.into(),
306                op: "added".into(),
307                category: "c".into(),
308                key: "k".repeat(200),
309                created: T0.into(),
310                fields: Vec::new(),
311                session: "s".into(),
312            })
313            .unwrap()
314        };
315        let lines: Vec<String> = (1..=10_000).map(row).collect();
316        std::fs::write(&path, lines.join("\n") + "\n").unwrap();
317        trim(&path);
318        assert!(std::fs::metadata(&path).unwrap().len() <= MAX_JOURNAL_BYTES);
319        assert_eq!(head_lsn(dir.path()), 10_000);
320        assert_eq!(rows_since(dir.path(), 0, 1)[0].lsn, 5_001);
321    }
322}
lean_ctx/core/knowledge/journal.rs

lean_ctx/core/knowledge/
journal.rs
