agentstatedeveloper_core/
ledger_integrity.rs

1//! Ledger integrity across the three places an entry lives.
2//!
3//! - **cache** — `asd_ledger_cache` in SQLite. Written by every
4//!   `append_entry` and never deleted from, so it holds every entry ever
5//!   acknowledged. Readers serve from it first.
6//! - **ASG** — `/asd/v1/ledger/<symbol>/<entry>`, the authoritative store
7//!   and the only thing `asd sync` exports.
8//! - **sidecar** — `.asd/v1/ledger/<symbol>/<entry>.json`, what `asd sync`
9//!   wrote and what `asd hydrate` restores from.
10//!
11//! Before ASG compare-and-swapped its refs, concurrent writers and `asd
12//! index` speculation commits discarded ledger commits from ASG after the
13//! cache had recorded them (1,684 of 12,085 on one store). Every surface kept
14//! working because reads come from the cache, so the loss was invisible
15//! until a hydrate came back short. This module makes the three counts
16//! visible and restores cache-only entries into ASG.
17
18use std::collections::BTreeSet;
19use std::path::Path;
20
21use agentstategraph::{CommitOptions, Repository};
22use agentstategraph_core::IntentCategory;
23use serde::Serialize;
24use serde_json::Value;
25
26use crate::engine::Engine;
27use crate::error::{AsdError, Result};
28use crate::paths;
29use crate::search_fts::SearchFtsDb;
30
31/// Ledger entry counts in each store, plus the gaps between them.
32#[derive(Debug, Clone, Serialize)]
33pub struct LedgerCounts {
34    /// Rows in `asd_ledger_cache` for the ref. `None` when no cache is open.
35    pub cache: Option<usize>,
36    /// Entries in the ASG ledger tree.
37    pub asg: usize,
38    /// Entry files under `.asd/v1/ledger/`. `None` when there is no sidecar.
39    pub sidecar: Option<usize>,
40    /// Cached entries absent from ASG — lost writes. `asd repair --fix`
41    /// restores them.
42    pub missing_from_asg: usize,
43    /// ASG entries absent from the sidecar — `asd sync` has not run since.
44    pub missing_from_sidecar: usize,
45}
46
47impl LedgerCounts {
48    /// True when nothing the cache acknowledged is missing downstream.
49    pub fn is_consistent(&self) -> bool {
50        self.missing_from_asg == 0 && self.missing_from_sidecar == 0
51    }
52
53    /// One-line warning for a mismatch, or `None` when consistent.
54    pub fn warning(&self) -> Option<String> {
55        if self.is_consistent() {
56            return None;
57        }
58        let mut parts = Vec::new();
59        if self.missing_from_asg > 0 {
60            parts.push(format!(
61                "{} ledger entr{} in the cache but missing from the store — \
62                 run `asd repair` to review and `asd repair --fix` to restore",
63                self.missing_from_asg,
64                if self.missing_from_asg == 1 {
65                    "y is"
66                } else {
67                    "ies are"
68                }
69            ));
70        }
71        if self.missing_from_sidecar > 0 {
72            parts.push(format!(
73                "{} ledger entr{} not in .asd/ — run `asd sync`",
74                self.missing_from_sidecar,
75                if self.missing_from_sidecar == 1 {
76                    "y is"
77                } else {
78                    "ies are"
79                }
80            ));
81        }
82        Some(parts.join("; "))
83    }
84}
85
86/// Every entry id in the ASG ledger tree, across all symbols.
87///
88/// Reads keys only: capped one level below the ledger root, each symbol's
89/// bucket comes back as its key list and no entry is ever loaded. `asd
90/// status` calls this, so it has to stay cheap on a store with tens of
91/// thousands of entries.
92pub fn asg_ledger_entry_ids(repo: &Repository, ref_name: &str) -> BTreeSet<String> {
93    let mut ids = BTreeSet::new();
94    // A store with no ledger yet has no `/asd/v1/ledger` path at all.
95    if let Ok(Value::Object(by_symbol)) = repo.get_json_capped(ref_name, &paths::ledger_root(), 1) {
96        for bucket in by_symbol.values() {
97            match bucket.get("_keys") {
98                Some(Value::Array(keys)) => {
99                    ids.extend(keys.iter().filter_map(|k| k.as_str().map(str::to_string)));
100                }
101                // Not truncated (e.g. an empty bucket): read the keys directly.
102                _ => {
103                    if let Value::Object(entries) = bucket {
104                        ids.extend(entries.keys().cloned());
105                    }
106                }
107            }
108        }
109    }
110    ids
111}
112
113/// Every entry id under `<root>/.asd/v1/ledger/`, or `None` when the sidecar
114/// has no ledger directory.
115pub fn sidecar_ledger_entry_ids(root: &Path) -> Option<BTreeSet<String>> {
116    let ledger = root.join(".asd").join("v1").join("ledger");
117    let symbols = std::fs::read_dir(&ledger).ok()?;
118    let mut ids = BTreeSet::new();
119    for sym in symbols.flatten() {
120        let Ok(files) = std::fs::read_dir(sym.path()) else {
121            continue;
122        };
123        for f in files.flatten() {
124            let path = f.path();
125            if path.extension().and_then(|e| e.to_str()) == Some("json")
126                && let Some(stem) = path.file_stem().and_then(|s| s.to_str())
127            {
128                ids.insert(stem.to_string());
129            }
130        }
131    }
132    Some(ids)
133}
134
135/// Count ledger entries in the cache, ASG and (when `sidecar_root` is given
136/// and has one) the sidecar, and the gaps between them.
137pub fn ledger_counts(
138    repo: &Repository,
139    ref_name: &str,
140    fts: Option<&SearchFtsDb>,
141    sidecar_root: Option<&Path>,
142) -> Result<LedgerCounts> {
143    let asg = asg_ledger_entry_ids(repo, ref_name);
144
145    let (cache, missing_from_asg) = match fts {
146        Some(fts) => {
147            let ids = fts
148                .ledger_cache_entry_ids(ref_name)
149                .map_err(|e| AsdError::Other(format!("read ledger cache: {e}")))?;
150            let missing = ids.iter().filter(|id| !asg.contains(*id)).count();
151            (Some(ids.len()), missing)
152        }
153        None => (None, 0),
154    };
155
156    let sidecar_ids = sidecar_root.and_then(sidecar_ledger_entry_ids);
157    let missing_from_sidecar = sidecar_ids
158        .as_ref()
159        .map(|s| asg.difference(s).count())
160        .unwrap_or(0);
161
162    Ok(LedgerCounts {
163        cache,
164        asg: asg.len(),
165        sidecar: sidecar_ids.map(|s| s.len()),
166        missing_from_asg,
167        missing_from_sidecar,
168    })
169}
170
171/// Outcome of [`restore_missing_ledger_entries`].
172#[derive(Debug, Clone, Default, Serialize)]
173pub struct LedgerRestoreReport {
174    /// Cached entries that were missing from ASG before the restore.
175    pub missing: usize,
176    /// Entries written back into ASG.
177    pub restored: usize,
178    /// `(entry_id, error)` for entries that could not be written.
179    pub failed: Vec<(String, String)>,
180    /// Cache rows missing from ASG whose body no longer parses, so there is
181    /// nothing to restore them from.
182    pub unparseable: Vec<String>,
183}
184
185/// Cached ledger entries whose `entry_id` appears nowhere in the ASG ledger
186/// tree. Matching by id rather than path means an entry moved by `ledger
187/// rebind` (whose cache row follows it to the new symbol) is never
188/// mistaken for a lost one.
189///
190/// Returns the restorable entries and the ids of missing rows whose cached
191/// body no longer parses.
192pub fn missing_ledger_entries(
193    engine: &Engine,
194) -> Result<(Vec<crate::schema::LedgerEntry>, Vec<String>)> {
195    let Some(fts) = engine.fts.as_ref() else {
196        return Ok((Vec::new(), Vec::new()));
197    };
198    let asg = asg_ledger_entry_ids(&engine.repo, &engine.ref_name);
199    let (entries, unparseable) = fts
200        .all_ledger_entries(&engine.ref_name)
201        .map_err(|e| AsdError::Other(format!("read ledger cache: {e}")))?;
202    Ok((
203        entries
204            .into_iter()
205            .filter(|e| !asg.contains(&e.entry_id))
206            .collect(),
207        unparseable
208            .into_iter()
209            .filter(|id| !asg.contains(id))
210            .collect(),
211    ))
212}
213
214/// Write every cache-only ledger entry back into ASG, with its reverse-index
215/// record, exactly as the cache holds it. Each restore is its own commit so
216/// the audit trail says what was recovered and from where.
217pub fn restore_missing_ledger_entries(
218    engine: &Engine,
219    agent_id: &str,
220) -> Result<LedgerRestoreReport> {
221    let (missing, unparseable) = missing_ledger_entries(engine)?;
222    let mut report = LedgerRestoreReport {
223        missing: missing.len() + unparseable.len(),
224        unparseable,
225        ..Default::default()
226    };
227    if missing.is_empty() {
228        return Ok(report);
229    }
230
231    // One commit for the whole restore. A commit per entry — two, with its
232    // reverse-index record — stored a fresh copy of the ledger and
233    // ledger-idx maps each time: restoring 1,684 entries grew a 1.5 GB store
234    // to 5.5 GB. Both subtrees are rebuilt in memory on top of the state the
235    // speculation forked from and written once each, so the commit adds
236    // exactly the restored entries and keeps anything written meanwhile.
237    let repo = &engine.repo;
238    let (spec, fork) =
239        crate::subtree::speculate_at_head(repo, &engine.ref_name, "asd-repair-ledger")?;
240    let staged = (|| -> Result<()> {
241        let mut ledger = crate::subtree::read_seed(repo, &fork, &paths::ledger_root())?;
242        let mut index = crate::subtree::read_seed(repo, &fork, &paths::ledger_index_root())?;
243        for entry in &missing {
244            let by_id = ledger
245                .entry(entry.symbol_id.clone())
246                .or_insert_with(|| Value::Object(Default::default()));
247            let Value::Object(by_id) = by_id else {
248                return Err(AsdError::Other(format!(
249                    "ledger node for {} is not a map",
250                    entry.symbol_id
251                )));
252            };
253            by_id.insert(entry.entry_id.clone(), serde_json::to_value(entry)?);
254            index.insert(
255                entry.entry_id.clone(),
256                Value::String(entry.symbol_id.clone()),
257            );
258        }
259        repo.spec_set_json(spec, &paths::ledger_root(), &Value::Object(ledger))
260            .map_err(|e| AsdError::Other(e.to_string()))?;
261        repo.spec_set_json(spec, &paths::ledger_index_root(), &Value::Object(index))
262            .map_err(|e| AsdError::Other(e.to_string()))?;
263        Ok(())
264    })();
265    let committed = staged.and_then(|()| {
266        // The audit trail: what was recovered, by kind, with every id.
267        let mut kinds: std::collections::BTreeMap<&str, usize> = Default::default();
268        for entry in &missing {
269            *kinds.entry(entry.kind.as_str()).or_default() += 1;
270        }
271        let kinds = kinds
272            .iter()
273            .map(|(kind, n)| format!("{n} {kind}"))
274            .collect::<Vec<_>>()
275            .join(", ");
276        let ids = missing
277            .iter()
278            .map(|e| format!("{} ({})", e.entry_id, e.symbol_id))
279            .collect::<Vec<_>>()
280            .join("\n");
281        let opts = CommitOptions::new(
282            agent_id,
283            IntentCategory::Refine,
284            format!(
285                "restore {} ledger entries lost from the store, recovered from the ledger cache ({kinds})",
286                missing.len()
287            ),
288        )
289        .with_reasoning(ids);
290        repo.commit_speculation(spec, opts)
291            .map(|_| ())
292            .map_err(|e| AsdError::Other(e.to_string()))
293    });
294    match committed {
295        Ok(()) => report.restored = missing.len(),
296        Err(e) => {
297            let _ = repo.discard_speculation(spec);
298            let e = e.to_string();
299            report.failed = missing
300                .iter()
301                .map(|entry| (entry.entry_id.clone(), e.clone()))
302                .collect();
303        }
304    }
305    Ok(report)
306}
agentstatedeveloper_core/ledger_integrity.rs

agentstatedeveloper_core/
ledger_integrity.rs
