Skip to main content

borg/
memory_export.rs

1//! `borg memory`: the evidence behind each lesson, for an outside store to take over long-term knowledge.
2//!
3//! - `borg memory export [--json] [--dir DIR] [--absorbed] [--since YYYY-MM-DD]`: read-only, 0 tokens.
4//!   One evidence record per lesson of the collective DIR (default: the current directory) routes to;
5//!   an off directory prints nothing. `--absorbed` lists the absorbed lessons only, `--since` the lessons
6//!   whose last change (`Evidence::last_change_ms`) is on or after that UTC day.
7//! - `borg memory absorb <id>... [--note TEXT]` and `borg memory release <id>...`: the lesson is now
8//!   held outside Borg, or no longer (absorb.rs, which also says what an absorb does to delivery).
9//!
10//! **The record** (`--json`: `{"schema": SCHEMA, "collective", "lessons": [...]}`, `document`): the
11//! lesson as stored (id, version, repo, question, situations, consequence, area, scope, reach `team` or
12//! `repo`, overrides, supersedes and the computed `superseded_by`, source, origin, support, engineers
13//! under this machine's aliases, history: the old versions' wording); its memory exactly as the
14//! conscience weighs it (memory.rs `derive` over outcome.rs `read_team`, as conscience.rs `load`): the
15//! Beta belief's α, β, mean and 95% equal-tailed credible interval, the per-session result counts,
16//! activation, consolidated and dormant; its breadth: the distinct sessions that changed their work on
17//! it and the distinct repos of those changes (team lines carry hashed repos, so counts, never names);
18//! the raw counts of its placebo and tail-sample judgements from this machine's outcome log (`Judged`),
19//! no estimate and no threshold; whether it is absorbed (`{note, ms, by, version}`, else null) and,
20//! absorbed, its recurrence: the episodes merged into it since the absorbed version.
21//!
22//! Every text field but the ids (`id`, `supersedes`, `superseded_by`) and `reach` goes through notes.rs
23//! `redact`, the engineers and an absorb's `by` too. A personal lesson's tombstone (reach
24//! `personal`) is left out, and the store's `personal.jsonl` is never read. A field the logs do not hold is omitted, never guessed:
25//! `version`, `support` and `history` for a lesson line without them (an older or hand-written line),
26//! `placebo` and `tail` when the log holds no such judgement of the lesson, `recurrence` when the
27//! absorb named no version (the lesson was not in the store yet) or the store no longer holds that
28//! version's line.
29
30use std::collections::{BTreeSet, HashMap, HashSet};
31use std::path::{Path, PathBuf};
32
33use serde::Serialize;
34use serde_json::{Value, json};
35
36use crate::absorb::{self, Record};
37use crate::conscience::{self, Challenge};
38use crate::notes::redact;
39use crate::outcome::{self, Outcome};
40use crate::{collective, learn, reach};
41
42/// Protocol: the export's schema version. A change to a field's meaning or a removed field raises it;
43/// an added field does not.
44pub const SCHEMA: u32 = 1;
45
46/// Which lessons `evidence` lists.
47#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
48pub struct Filter {
49    /// Only absorbed lessons.
50    pub absorbed: bool,
51    /// Only lessons whose last change is at or after this epoch ms.
52    pub since_ms: Option<i64>,
53}
54
55/// One lesson's evidence record (module doc).
56#[derive(Debug, Clone, Serialize, PartialEq)]
57pub struct Evidence {
58    pub id: String,
59    #[serde(skip_serializing_if = "Option::is_none")]
60    pub version: Option<u32>,
61    pub repo: String,
62    pub question: String,
63    pub situations: Vec<String>,
64    pub consequence: String,
65    pub area: String,
66    pub scope: Vec<String>,
67    /// `team` or `repo` (reach.rs).
68    pub reach: String,
69    pub overrides: String,
70    pub supersedes: Vec<String>,
71    /// The lessons that list this one in their `supersedes`, sorted.
72    pub superseded_by: Vec<String>,
73    pub source: String,
74    pub origin: String,
75    #[serde(skip_serializing_if = "Option::is_none")]
76    pub support: Option<u64>,
77    pub engineers: Vec<String>,
78    #[serde(skip_serializing_if = "Option::is_none")]
79    pub history: Option<Vec<Value>>,
80    pub memory: MemoryView,
81    pub breadth: Breadth,
82    #[serde(skip_serializing_if = "Option::is_none")]
83    pub placebo: Option<Judged>,
84    #[serde(skip_serializing_if = "Option::is_none")]
85    pub tail: Option<Judged>,
86    pub absorbed: Option<Absorbed>,
87    #[serde(skip_serializing_if = "Option::is_none")]
88    pub recurrence: Option<usize>,
89    /// The lesson's last change, epoch ms: the latest of its first correction, each later engineer
90    /// joining, each of its episodes this machine holds, each outcome naming it, and its last absorb or
91    /// release.
92    pub last_change_ms: i64,
93}
94
95/// The memory the conscience weighs the lesson by (memory.rs `Memory`), rounded to 4 decimals.
96#[derive(Debug, Clone, Serialize, PartialEq)]
97pub struct MemoryView {
98    pub alpha: f64,
99    pub beta: f64,
100    pub mean: f64,
101    /// The Beta(α, β) belief's 2.5% and 97.5% quantiles.
102    pub ci95: [f64; 2],
103    pub changed: u32,
104    pub ignored: u32,
105    pub corrected: u32,
106    pub vetoed: u32,
107    pub regretted: f64,
108    pub tested: u32,
109    pub activation: f64,
110    pub consolidated: bool,
111    pub dormant: bool,
112}
113
114/// The distinct sessions that changed their work on the lesson, and the distinct repos of those
115/// changes: what memory.rs counts to consolidate it.
116#[derive(Debug, Clone, Copy, Default, Serialize, PartialEq, Eq)]
117pub struct Breadth {
118    pub sessions: usize,
119    pub repos: usize,
120}
121
122/// Raw judgement counts. Placebo: the judge's verdicts on the lesson where it was routed nowhere
123/// (outcome.rs `PLACEBO`), and of them the ones that found the drone changed its work on its point
124/// anyway. Tail: the graded checks whose tail sample showed the lesson past the pool budget (outcome.rs
125/// `Outcome::tail`), and of them the ones whose model graded it as applying.
126#[derive(Debug, Clone, Copy, Default, Serialize, PartialEq, Eq)]
127pub struct Judged {
128    pub count: u32,
129    pub applied: u32,
130}
131
132/// The lesson's absorb (absorb.rs `Record`).
133#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
134pub struct Absorbed {
135    pub note: Option<String>,
136    pub ms: i64,
137    pub by: String,
138    pub version: Option<u32>,
139}
140
141/// Every line of the lessons file per id, in file order: the last is the lesson as the conscience
142/// reads it (conscience.rs `load_unweighted`).
143fn lines_by_id(lessons: &Path) -> HashMap<String, Vec<Value>> {
144    let mut by: HashMap<String, Vec<Value>> = HashMap::new();
145    for v in std::fs::read_to_string(lessons).unwrap_or_default().lines().filter_map(|l| serde_json::from_str::<Value>(l).ok()) {
146        if let Some(id) = v["id"].as_str().filter(|i| !i.is_empty()) {
147            by.entry(id.to_string()).or_default().push(v);
148        }
149    }
150    by
151}
152
153/// The lesson's current version (its last line's `version`), None for an id the lines lack or a line
154/// without one.
155fn current_version(lines: &HashMap<String, Vec<Value>>, id: &str) -> Option<u32> {
156    lines.get(id).and_then(|v| v.last()).and_then(|l| l["version"].as_u64()).and_then(|n| u32::try_from(n).ok())
157}
158
159fn episodes(line: &Value) -> impl Iterator<Item = &str> {
160    line["episodes"].as_array().into_iter().flatten().filter_map(Value::as_str)
161}
162
163/// The episodes merged into the lesson since its absorb: the latest line's episodes beyond those of the
164/// last line at the absorbed version. None when the absorb named no version or no line has it.
165fn recurrence(lines: &[Value], r: &Record) -> Option<usize> {
166    let v = u64::from(r.version?);
167    let at = lines.iter().rev().find(|l| l["version"].as_u64() == Some(v))?;
168    let then: HashSet<&str> = episodes(at).collect();
169    Some(lines.last().map_or(0, |l| episodes(l).filter(|e| !then.contains(e)).count()))
170}
171
172/// The ms of this machine's episodes (`episodes.jsonl`), by id.
173fn episode_ms(store: &Path) -> HashMap<String, i64> {
174    let text = std::fs::read_to_string(store.join("episodes.jsonl")).unwrap_or_default();
175    text.lines().filter_map(|l| serde_json::from_str::<Value>(l).ok()).filter_map(|v| Some((v["id"].as_str()?.to_string(), v["ms"].as_i64()?))).collect()
176}
177
178fn redact_value(v: &Value) -> Value {
179    match v {
180        Value::String(s) => Value::String(redact(s)),
181        Value::Array(a) => Value::Array(a.iter().map(redact_value).collect()),
182        Value::Object(o) => Value::Object(o.iter().map(|(k, v)| (k.clone(), redact_value(v))).collect()),
183        other => other.clone(),
184    }
185}
186
187/// The placebo and tail judgements of every lesson in this machine's outcome log (`Judged`).
188fn judgements(local: &[Outcome]) -> (HashMap<&str, Judged>, HashMap<&str, Judged>) {
189    let (mut placebo, mut tail): (HashMap<&str, Judged>, HashMap<&str, Judged>) = Default::default();
190    for o in local {
191        if o.channel == outcome::PLACEBO && (o.kind == outcome::CHANGED || o.kind == outcome::IGNORED) {
192            for l in &o.lessons {
193                let j = placebo.entry(l.as_str()).or_default();
194                j.count += 1;
195                j.applied += u32::from(o.kind == outcome::CHANGED);
196            }
197        }
198        if o.kind == outcome::GRADED {
199            for l in &o.tail {
200                let j = tail.entry(l.as_str()).or_default();
201                j.count += 1;
202                j.applied += u32::from(o.lessons.contains(l));
203            }
204        }
205    }
206    (placebo, tail)
207}
208
209/// The evidence records of the store `store` (the directory of its `lessons.jsonl`) that `filter`
210/// lets through, by first correction then id. Never fails: missing logs are no evidence.
211pub fn evidence(store: &Path, filter: &Filter) -> Vec<Evidence> {
212    let path = store.join("lessons.jsonl");
213    let lessons: Vec<Challenge> = conscience::load_unweighted(&path, None);
214    let outcomes = outcome::read_team(&path);
215    let memory = crate::memory::derive(&outcomes, &lessons, None);
216    let local = outcome::read(&outcome::path_beside(&path));
217    let (placebo, tail) = judgements(&local);
218    let lines = lines_by_id(&path);
219    let log = absorb::latest(&absorb::path(store), None);
220    let episode_ms = episode_ms(store);
221    // Per lesson: the sessions and repos of its changes, and its outcomes' latest ms.
222    let mut changes: HashMap<&str, (HashSet<&str>, HashSet<&str>)> = HashMap::new();
223    let mut last_outcome: HashMap<&str, i64> = HashMap::new();
224    for o in &outcomes {
225        for l in &o.lessons {
226            let m = last_outcome.entry(l.as_str()).or_insert(o.ms);
227            *m = (*m).max(o.ms);
228            if o.kind == outcome::CHANGED && outcome::is_result(o) {
229                let (s, r) = changes.entry(l.as_str()).or_default();
230                s.insert(o.session.as_str());
231                r.insert(o.repo.as_str());
232            }
233        }
234    }
235    let live: Vec<&Challenge> = lessons.iter().filter(|c| c.reach != reach::PERSONAL).collect();
236    let mut by: HashMap<&str, BTreeSet<String>> = HashMap::new();
237    for c in &live {
238        for s in &c.supersedes {
239            by.entry(s.as_str()).or_default().insert(c.id.clone());
240        }
241    }
242    let mut out: Vec<(i64, Evidence)> = live
243        .iter()
244        .filter_map(|c| {
245            let rec = log.get(&c.id);
246            let absorbed = rec.filter(|r| r.absorbed());
247            if filter.absorbed && absorbed.is_none() {
248                return None;
249            }
250            let all = lines.get(&c.id).map(Vec::as_slice).unwrap_or_default();
251            let line = all.last().cloned().unwrap_or(Value::Null);
252            let mut last = c.origin_ms.max(c.joined.values().copied().max().unwrap_or(0));
253            last = last.max(episodes(&line).filter_map(|e| episode_ms.get(e).copied()).max().unwrap_or(0));
254            last = last.max(last_outcome.get(c.id.as_str()).copied().unwrap_or(0)).max(rec.map_or(0, |r| r.ms));
255            if filter.since_ms.is_some_and(|s| last < s) {
256                return None;
257            }
258            let m = &memory[&c.id];
259            let (alpha, beta) = (m.alpha(), m.beta());
260            let (lo, hi) = interval(alpha, beta, 0.95);
261            let breadth = changes.get(c.id.as_str()).map_or_else(Breadth::default, |(s, r)| Breadth { sessions: s.len(), repos: r.len() });
262            Some((
263                c.origin_ms,
264                Evidence {
265                    id: c.id.clone(),
266                    version: line["version"].as_u64().and_then(|n| u32::try_from(n).ok()),
267                    repo: redact(&c.repo),
268                    question: redact(&c.question),
269                    situations: c.situations.iter().map(String::as_str).map(redact).collect(),
270                    consequence: redact(&c.consequence),
271                    area: redact(&c.area),
272                    scope: c.scope.iter().map(String::as_str).map(redact).collect(),
273                    reach: if c.reach == reach::REPO { reach::REPO.into() } else { "team".into() },
274                    overrides: redact(&c.overrides),
275                    supersedes: c.supersedes.clone(),
276                    superseded_by: by.get(c.id.as_str()).map(|s| s.iter().cloned().collect()).unwrap_or_default(),
277                    source: redact(&c.source),
278                    origin: redact(line["origin"].as_str().unwrap_or_default()),
279                    support: line["support"].as_u64(),
280                    engineers: c.engineers.iter().map(String::as_str).map(redact).collect(),
281                    history: line["history"].as_array().map(|h| h.iter().map(redact_value).collect()),
282                    memory: MemoryView {
283                        alpha: outcome::round_p(alpha),
284                        beta: outcome::round_p(beta),
285                        mean: outcome::round_p(m.mean()),
286                        ci95: [outcome::round_p(lo), outcome::round_p(hi)],
287                        changed: m.changed,
288                        ignored: m.ignored,
289                        corrected: m.corrected,
290                        vetoed: m.vetoed,
291                        regretted: outcome::round_p(m.regretted),
292                        tested: m.tested,
293                        activation: outcome::round_p(m.activation),
294                        consolidated: m.consolidated,
295                        dormant: m.dormant,
296                    },
297                    breadth,
298                    placebo: placebo.get(c.id.as_str()).copied(),
299                    tail: tail.get(c.id.as_str()).copied(),
300                    absorbed: absorbed.map(|r| Absorbed { note: r.note.as_deref().map(redact), ms: r.ms, by: redact(&r.by), version: r.version }),
301                    recurrence: absorbed.and_then(|r| recurrence(all, r)),
302                    last_change_ms: last,
303                },
304            ))
305        })
306        .collect();
307    out.sort_by(|a, b| (a.0, &a.1.id).cmp(&(b.0, &b.1.id)));
308    out.into_iter().map(|(_, e)| e).collect()
309}
310
311/// The export of the store `store` as one JSON document, `{"schema": SCHEMA, "collective", "lessons"}`:
312/// what `borg memory export --json` prints and borg-web's `/api/memory` serves.
313pub fn document(collective: &str, store: &Path, filter: &Filter) -> Value {
314    json!({"schema": SCHEMA, "collective": collective, "lessons": evidence(store, filter)})
315}
316
317/// `borg status`'s line of a store: its absorbed lessons, and how many of them recurred (a correction
318/// merged into one since its absorb, `recurrence`). An absorbed id the store does not hold yet is not
319/// counted.
320pub fn summary(store: &Path) -> (usize, usize) {
321    let lines = lines_by_id(&store.join("lessons.jsonl"));
322    let state = absorb::state(&absorb::path(store), None);
323    let held: Vec<(&Record, &[Value])> = state
324        .values()
325        .filter_map(|r| lines.get(&r.id).map(|l| (r, l.as_slice())))
326        .filter(|(_, l)| l.last().is_some_and(|x| x["reach"].as_str() != Some(reach::PERSONAL)))
327        .collect();
328    (held.len(), held.iter().filter(|(r, l)| recurrence(l, r).is_some_and(|n| n > 0)).count())
329}
330
331// ---------------------------------------------------------------- the Beta interval
332
333/// ln Γ(x) for x > 0 (Lanczos, g = 7, nine terms: about 15 significant digits).
334fn ln_gamma(x: f64) -> f64 {
335    const G: [f64; 9] = [
336        0.999_999_999_999_809_9,
337        676.520_368_121_885_1,
338        -1_259.139_216_722_402_8,
339        771.323_428_777_653_1,
340        -176.615_029_162_140_6,
341        12.507_343_278_686_905,
342        -0.138_571_095_265_720_12,
343        9.984_369_578_019_572e-6,
344        1.505_632_735_149_311_6e-7,
345    ];
346    if x < 0.5 {
347        // Reflection: Γ(x) Γ(1 − x) = π / sin(πx).
348        return (std::f64::consts::PI / (std::f64::consts::PI * x).sin()).ln() - ln_gamma(1.0 - x);
349    }
350    let x = x - 1.0;
351    let a = G[1..].iter().enumerate().fold(G[0], |s, (i, g)| s + g / (x + i as f64 + 1.0));
352    let t = x + 7.5;
353    0.5 * (2.0 * std::f64::consts::PI).ln() + (x + 0.5) * t.ln() - t + a.ln()
354}
355
356/// The continued fraction of the incomplete beta function (modified Lentz, as in Numerical Recipes'
357/// `betacf`).
358fn beta_cf(a: f64, b: f64, x: f64) -> f64 {
359    // Method: TINY keeps a denominator off zero; EPS is double precision; the fraction needs about
360    // √max(a, b) terms, so MAX_TERMS covers beliefs of millions of outcomes and is never the stop.
361    const TINY: f64 = 1e-300;
362    const EPS: f64 = 1e-15;
363    const MAX_TERMS: u32 = 10_000;
364    let (qab, qap, qam) = (a + b, a + 1.0, a - 1.0);
365    let mut c = 1.0;
366    let mut d = 1.0 - qab * x / qap;
367    d = if d.abs() < TINY { TINY } else { d };
368    d = 1.0 / d;
369    let mut h = d;
370    for m in 1..=MAX_TERMS {
371        let m = f64::from(m);
372        let m2 = 2.0 * m;
373        for aa in [m * (b - m) * x / ((qam + m2) * (a + m2)), -(a + m) * (qab + m) * x / ((a + m2) * (qap + m2))] {
374            d = 1.0 + aa * d;
375            d = if d.abs() < TINY { TINY } else { d };
376            c = 1.0 + aa / c;
377            c = if c.abs() < TINY { TINY } else { c };
378            d = 1.0 / d;
379            h *= d * c;
380        }
381        if (d * c - 1.0).abs() < EPS {
382            break;
383        }
384    }
385    h
386}
387
388/// The regularised incomplete beta function I_x(a, b): the Beta(a, b) CDF at x.
389fn beta_cdf(a: f64, b: f64, x: f64) -> f64 {
390    if x <= 0.0 {
391        return 0.0;
392    }
393    if x >= 1.0 {
394        return 1.0;
395    }
396    let front = (ln_gamma(a + b) - ln_gamma(a) - ln_gamma(b) + a * x.ln() + b * (1.0 - x).ln()).exp();
397    if x < (a + 1.0) / (a + b + 2.0) { front * beta_cf(a, b, x) / a } else { 1.0 - front * beta_cf(b, a, 1.0 - x) / b }
398}
399
400/// The Beta(a, b) quantile at `p`, by bisection on the CDF. Method: the CDF is monotone, so 60 halvings
401/// place it within 2⁻⁶⁰ of the root, finer than the 4 decimals the export shows.
402fn beta_quantile(a: f64, b: f64, p: f64) -> f64 {
403    let (mut lo, mut hi) = (0.0, 1.0);
404    for _ in 0..60 {
405        let mid = 0.5 * (lo + hi);
406        if beta_cdf(a, b, mid) < p {
407            lo = mid;
408        } else {
409            hi = mid;
410        }
411    }
412    0.5 * (lo + hi)
413}
414
415/// The equal-tailed `level` credible interval of Beta(a, b).
416pub fn interval(a: f64, b: f64, level: f64) -> (f64, f64) {
417    let tail = (1.0 - level) / 2.0;
418    (beta_quantile(a, b, tail), beta_quantile(a, b, 1.0 - tail))
419}
420
421// ---------------------------------------------------------------- the command
422
423/// `borg memory`'s subcommands and flags (completions.rs).
424pub const SUBS: &[&str] = &["export", "absorb", "release"];
425pub const FLAGS: &[&str] = &["--json", "--dir", "--absorbed", "--since", "--note"];
426
427const USAGE: &str = "usage: borg memory export [--json] [--dir DIR] [--absorbed] [--since YYYY-MM-DD]
428       borg memory absorb <id>... [--note TEXT] [--dir DIR] [--json]
429       borg memory release <id>... [--dir DIR] [--json]
430  export: each lesson's evidence (0 tokens); absorb: a lesson is now held outside Borg, which stops
431  delivering it; release: Borg delivers it again";
432
433/// The parsed arguments after the subcommand.
434#[derive(Debug, Default, PartialEq)]
435struct Args<'a> {
436    ids: Vec<&'a str>,
437    dir: Option<&'a str>,
438    note: Option<&'a str>,
439    since: Option<&'a str>,
440    json: bool,
441    absorbed: bool,
442}
443
444fn parse<'a>(args: &[&'a str]) -> Option<Args<'a>> {
445    let mut a = Args::default();
446    let mut it = args.iter();
447    while let Some(w) = it.next() {
448        match *w {
449            "--json" => a.json = true,
450            "--absorbed" => a.absorbed = true,
451            "--dir" => a.dir = Some(it.next()?),
452            "--note" => a.note = Some(it.next()?),
453            "--since" => a.since = Some(it.next()?),
454            w if w.starts_with("--") => return None,
455            w => a.ids.push(w),
456        }
457    }
458    Some(a)
459}
460
461fn usage() -> i32 {
462    eprintln!("{USAGE}");
463    2
464}
465
466/// `borg memory <export|absorb|release> ...`; the exit code.
467pub fn main(args: &[&str]) -> i32 {
468    let Some((&sub, rest)) = args.split_first() else { return usage() };
469    let Some(a) = parse(rest) else { return usage() };
470    let valid = match sub {
471        "export" => a.ids.is_empty() && a.note.is_none(),
472        "absorb" => !a.ids.is_empty() && !a.absorbed && a.since.is_none(),
473        "release" => !a.ids.is_empty() && !a.absorbed && a.since.is_none() && a.note.is_none(),
474        _ => false,
475    };
476    if !valid {
477        return usage();
478    }
479    // Routing compares path text, so a relative --dir is made absolute first, as hooks' cwd is.
480    let dir = a.dir.map_or_else(|| std::env::current_dir().unwrap_or_default(), PathBuf::from);
481    let cwd = std::path::absolute(&dir).unwrap_or(dir).display().to_string();
482    let route = match collective::load() {
483        Ok(cfg) => collective::route(cfg.as_ref(), &cwd, &learn::home()),
484        Err(e) => {
485            eprintln!("borg memory: invalid collectives config ({e}); everything is off");
486            return 1;
487        }
488    };
489    let collective::Route::Collective { name, home, .. } = &route else {
490        if sub == "export" {
491            return 0;
492        }
493        eprintln!("off: {cwd} belongs to no collective; borg records nothing here");
494        return 1;
495    };
496    match sub {
497        "export" => export_cmd(name, home, &a),
498        _ => absorb_cmd(sub, home, &cwd, &a),
499    }
500}
501
502fn export_cmd(name: &str, home: &Path, a: &Args) -> i32 {
503    let since_ms = match a.since.map(|d| (d.len() == 10).then(|| borg_trace::parse_ts(&format!("{d}T00:00:00Z"))).filter(|ms| *ms > 0)) {
504        Some(None) => {
505            eprintln!("borg memory export: --since wants YYYY-MM-DD\n{USAGE}");
506            return 2;
507        }
508        s => s.flatten(),
509    };
510    let filter = Filter { absorbed: a.absorbed, since_ms };
511    if a.json {
512        println!("{}", serde_json::to_string_pretty(&document(name, home, &filter)).unwrap_or_else(|_| "null".into()));
513        return 0;
514    }
515    print!("{}", render(name, &evidence(home, &filter)));
516    0
517}
518
519/// The compact listing: a header, then per lesson its belief, breadth and absorb, and its question.
520fn render(name: &str, records: &[Evidence]) -> String {
521    let absorbed = records.iter().filter(|e| e.absorbed.is_some()).count();
522    let mut s = format!("collective {name}: {} lessons, {absorbed} absorbed (borg memory export --json: every field)\n", records.len());
523    for e in records {
524        let version = e.version.map_or_else(String::new, |v| format!(" v{v}"));
525        let m = &e.memory;
526        let held = match &e.absorbed {
527            None => String::new(),
528            Some(x) => {
529                let note = x.note.as_deref().map_or_else(String::new, |n| format!(": {}", crate::stats::excerpt(n, 60)));
530                let recurred = e.recurrence.map_or_else(String::new, |n| format!(", recurred {n}"));
531                format!("  absorbed{note}{recurred}")
532            }
533        };
534        s.push_str(&format!(
535            "{}{version}  {}  mean {:.2} ({:.2}-{:.2})  changed {} ({} sessions, {} repos)  corrected {}{held}\n",
536            e.id, e.reach, m.mean, m.ci95[0], m.ci95[1], m.changed, e.breadth.sessions, e.breadth.repos, m.corrected
537        ));
538        s.push_str(&format!("    {}\n", crate::stats::excerpt(&e.question, 160)));
539    }
540    s
541}
542
543fn absorb_cmd(sub: &str, home: &Path, cwd: &str, a: &Args) -> i32 {
544    let note = match absorb::note(a.note) {
545        Ok(n) => n,
546        Err(e) => {
547            eprintln!("borg memory absorb: {e}");
548            return 2;
549        }
550    };
551    let kind = if sub == "absorb" { absorb::ABSORB } else { absorb::RELEASE };
552    let by = learn::engineer(cwd);
553    // Under learn's lock on the lessons: no merge lands between reading a lesson's version and the
554    // absorb that names it, and two commands here stamp in the order they write.
555    let (lines, before, written) = crate::live::locked(&home.join("lessons.jsonl"), || {
556        let lines = lines_by_id(&home.join("lessons.jsonl"));
557        let seen = absorb::latest(&absorb::path(home), None);
558        let before: HashSet<String> = seen.values().filter(|r| r.absorbed()).map(|r| r.id.clone()).collect();
559        // Stamped after what it sees: a peer's clock a little ahead never makes this command lose to it.
560        let now = absorb::stamp(&seen, &a.ids, learn::now_ms());
561        let written = absorb::write(home, &a.ids, kind, note.as_deref(), &by, now, &|id| current_version(&lines, id));
562        (lines, before, written)
563    });
564    let written = match written {
565        Ok(w) => w,
566        Err(e) => {
567            eprintln!("borg memory {sub}: cannot write {}: {e}", absorb::path(home).display());
568            return 1;
569        }
570    };
571    for r in &written {
572        if !lines.contains_key(&r.id) {
573            eprintln!("borg memory {sub}: {} is not a lesson of this collective (yet): recorded, it applies when the lesson arrives", r.id);
574        } else if kind == absorb::RELEASE && !before.contains(r.id.as_str()) {
575            eprintln!("borg memory release: {} was not absorbed here (yet): recorded", r.id);
576        }
577    }
578    if a.json {
579        let out: Vec<Value> = written
580            .iter()
581            .map(|r| {
582                let mut v = serde_json::to_value(r).unwrap_or_default();
583                v["known"] = json!(lines.contains_key(&r.id));
584                v
585            })
586            .collect();
587        println!("{}", Value::Array(out));
588        return 0;
589    }
590    for r in &written {
591        match (kind, &r.note) {
592            (absorb::RELEASE, _) => println!("released {}: borg delivers it again", r.id),
593            (_, Some(n)) => {
594                println!("absorbed {} (version {}): held in {n}; borg no longer delivers it", r.id, r.version.map_or("unknown".into(), |v| v.to_string()))
595            }
596            (_, None) => println!("absorbed {} (version {}): borg no longer delivers it", r.id, r.version.map_or("unknown".into(), |v| v.to_string())),
597        }
598    }
599    0
600}
601
602#[cfg(test)]
603mod tests {
604    use super::*;
605
606    fn temp(name: &str) -> PathBuf {
607        let d = std::env::temp_dir().join(format!("borg-memory-export-{name}-{}", std::process::id()));
608        let _ = std::fs::remove_dir_all(&d);
609        std::fs::create_dir_all(&d).unwrap();
610        d
611    }
612
613    fn close(a: f64, b: f64) -> bool {
614        (a - b).abs() < 1e-9
615    }
616
617    #[test]
618    fn the_beta_interval_matches_closed_forms() {
619        assert!(close(ln_gamma(5.0), 24f64.ln()) && close(ln_gamma(0.5), std::f64::consts::PI.sqrt().ln()));
620        // Beta(1, 1) is uniform; Beta(2, 1) has CDF x²; Beta(1, 2) has CDF 1 − (1 − x)².
621        let (lo, hi) = interval(1.0, 1.0, 0.95);
622        assert!(close(lo, 0.025) && close(hi, 0.975), "{lo} {hi}");
623        let (lo, hi) = interval(2.0, 1.0, 0.95);
624        assert!(close(lo, 0.025f64.sqrt()) && close(hi, 0.975f64.sqrt()), "{lo} {hi}");
625        let (lo, hi) = interval(1.0, 2.0, 0.95);
626        assert!(close(lo, 1.0 - 0.975f64.sqrt()) && close(hi, 1.0 - 0.025f64.sqrt()), "{lo} {hi}");
627        // Symmetric beliefs give symmetric intervals; more evidence narrows them; fractional counts work.
628        let (lo, hi) = interval(6.0, 6.0, 0.95);
629        assert!(close(lo, 1.0 - hi));
630        assert!(interval(60.0, 60.0, 0.95).1 - interval(60.0, 60.0, 0.95).0 < hi - lo);
631        let (lo, hi) = interval(1.5, 3.0, 0.95);
632        assert!(close(beta_cdf(1.5, 3.0, lo), 0.025) && close(beta_cdf(1.5, 3.0, hi), 0.975));
633    }
634
635    fn write_lines(path: &Path, lines: &[Value]) {
636        std::fs::write(path, lines.iter().map(|l| format!("{l}\n")).collect::<String>()).unwrap();
637    }
638
639    fn o(kind: &str, session: &str, lessons: &[&str], ms: i64, repo: &str, channel: &str) -> Value {
640        json!({"kind": kind, "session": session, "lessons": lessons, "ms": ms, "repo": repo, "channel": channel})
641    }
642
643    #[test]
644    fn the_export_holds_the_lesson_its_memory_breadth_judgements_and_absorb() {
645        let store = temp("record");
646        write_lines(
647            &store.join("lessons.jsonl"),
648            &[
649                json!({"id": "La", "origin_ms": 10, "repo": "app", "question": "Is the lock held across the await?", "situations": ["send under a lock"],
650                    "consequence": "a slow client stalls senders", "support": 1, "episodes": ["e1"], "history": [], "version": 1, "engineers": ["a@x"]}),
651                json!({"id": "La", "origin_ms": 10, "repo": "app", "question": "Is the send lock held across any await?", "situations": ["send under a lock"],
652                    "consequence": "a slow client stalls senders", "support": 2, "episodes": ["e1", "e2"],
653                    "history": [{"question": "Is the lock held across the await? token ghp_0123456789abcdefghijABCDEFGHIJ012345", "consequence": "c", "version": 1}],
654                    "version": 2, "engineers": ["a@x"], "area": "concurrency"}),
655                json!({"id": "Lb", "origin_ms": 20, "repo": "app", "question": "Is the queue drained by one writer?", "supersedes": ["La"], "reach": "repo"}),
656                json!({"id": "Lgone", "origin_ms": 30, "question": "(withdrawn)", "reach": "personal"}),
657            ],
658        );
659        write_lines(
660            &store.join("outcomes.jsonl"),
661            &[
662                o("asked", "s1", &["La"], 100, "app", "stop"),
663                o("asked_then_changed", "s1", &["La"], 110, "app", "stop"),
664                o("asked_then_changed", "s2", &["La"], 120, "web", "stop"),
665                o("asked_then_corrected", "s3", &["La"], 130, "app", "stop"),
666                // A placebo's verdicts are no result: counted raw, never in the belief.
667                o("asked_then_changed", "s4", &["La"], 140, "app", "placebo"),
668                o("asked_ignored", "s5", &["La"], 150, "app", "placebo"),
669                json!({"kind": "graded", "session": "s6", "lessons": ["La"], "tail": ["La", "Lb"], "ms": 160, "repo": "app"}),
670            ],
671        );
672        absorb::write(&store, &["La"], absorb::ABSORB, Some("skills/review@2026-10-30"), "lead@x", 200, &|_| Some(1)).unwrap();
673        let all = evidence(&store, &Filter::default());
674        assert_eq!(all.iter().map(|e| e.id.as_str()).collect::<Vec<_>>(), ["La", "Lb"], "a personal tombstone is left out");
675        let a = &all[0];
676        assert_eq!((a.version, a.support, a.area.as_str(), a.reach.as_str()), (Some(2), Some(2), "concurrency", "team"));
677        assert_eq!(a.superseded_by, ["Lb"]);
678        assert_eq!((a.memory.changed, a.memory.corrected), (2, 1));
679        assert_eq!((a.memory.alpha, a.memory.beta), (3.0, 3.0), "Beta(1 + 2 changed, 1 + 2 × 1 corrected)");
680        assert!(a.memory.ci95[0] < a.memory.mean && a.memory.mean < a.memory.ci95[1]);
681        assert_eq!(a.breadth, Breadth { sessions: 2, repos: 2 });
682        assert_eq!(a.placebo, Some(Judged { count: 2, applied: 1 }));
683        assert_eq!(a.tail, Some(Judged { count: 1, applied: 1 }));
684        assert_eq!(all[1].tail, Some(Judged { count: 1, applied: 0 }));
685        assert_eq!(all[1].placebo, None, "no placebo judgement: omitted");
686        let held = a.absorbed.as_ref().unwrap();
687        assert_eq!((held.note.as_deref(), held.by.as_str(), held.version, held.ms), (Some("skills/review@2026-10-30"), "lead@x", Some(1), 200));
688        assert_eq!(a.recurrence, Some(1), "e2 merged after version 1");
689        assert_eq!(a.last_change_ms, 200);
690        assert!(!serde_json::to_string(&a.history).unwrap().contains("ghp_0123"), "history is redacted");
691        // Lb: no version, support or history on its line: omitted, not guessed; not absorbed: null.
692        let b = serde_json::to_value(&all[1]).unwrap();
693        assert!(b.get("version").is_none() && b.get("support").is_none() && b.get("history").is_none() && b.get("recurrence").is_none(), "{b}");
694        assert_eq!((b["absorbed"].clone(), b["reach"].as_str()), (Value::Null, Some("repo")));
695        // The filters, and the document.
696        let only = evidence(&store, &Filter { absorbed: true, ..Filter::default() });
697        assert_eq!(only.iter().map(|e| e.id.as_str()).collect::<Vec<_>>(), ["La"]);
698        assert_eq!(evidence(&store, &Filter { since_ms: Some(170), ..Filter::default() }).len(), 1, "Lb last changed at its first correction, 20");
699        let doc = document("corp", &store, &Filter::default());
700        assert_eq!((doc["schema"].as_u64(), doc["collective"].as_str(), doc["lessons"].as_array().map(Vec::len)), (Some(1), Some("corp"), Some(2)));
701        assert_eq!(summary(&store), (1, 1));
702        // Released: no longer absorbed, nothing recurred.
703        absorb::write(&store, &["La"], absorb::RELEASE, None, "lead@x", 300, &|_| None).unwrap();
704        assert!(evidence(&store, &Filter::default())[0].absorbed.is_none());
705        assert_eq!(summary(&store), (0, 0));
706        let _ = std::fs::remove_dir_all(store);
707    }
708
709    #[test]
710    fn recurrence_needs_the_absorbed_version() {
711        let lines = vec![json!({"id": "L", "version": 1, "episodes": ["e1"]}), json!({"id": "L", "version": 3, "episodes": ["e1", "e2", "e3"]})];
712        let r = |v: Option<u32>| Record { id: "L".into(), kind: absorb::ABSORB.into(), ms: 1, by: String::new(), note: None, version: v };
713        assert_eq!(recurrence(&lines, &r(Some(1))), Some(2));
714        assert_eq!(recurrence(&lines, &r(Some(3))), Some(0));
715        assert_eq!(recurrence(&lines, &r(Some(2))), None, "no line at that version: not guessed");
716        assert_eq!(recurrence(&lines, &r(None)), None, "absorbed before the lesson arrived");
717    }
718
719    #[test]
720    fn a_correction_learned_after_the_absorb_merges_into_it_and_recurs() {
721        let store = temp("learn");
722        let lesson = json!({"id": "La", "origin_ms": 1, "repo": "app", "question": "Is the send lock held across any await?", "situations": ["send under a lock"],
723            "consequence": "a slow client stalls senders", "support": 1, "episodes": ["e1"], "history": [], "version": 1});
724        write_lines(&store.join("lessons.jsonl"), &[lesson]);
725        absorb::write(&store, &["La"], absorb::ABSORB, Some("skills/review@1"), "lead@x", 5, &|id| {
726            current_version(&lines_by_id(&store.join("lessons.jsonl")), id)
727        })
728        .unwrap();
729        let e = learn::Episode {
730            id: "e9".into(),
731            ms: 10,
732            repo: "app".into(),
733            session: "s9".into(),
734            task: "send frames".into(),
735            agent: "I hold the send lock across the await".into(),
736            correction: "no, the send lock must never be held across an await".into(),
737            engineer: "b@x".into(),
738            ..Default::default()
739        };
740        learn::record_episode_in(&store, &e).unwrap();
741        // Learning still sees the absorbed lesson: it is offered for merging.
742        let ask = |p: &str| {
743            assert!(p.contains("La"), "the absorbed lesson is a merge candidate: {p}");
744            Some(json!({"is_lesson": true, "question": "Is any lock held across an await?", "situations": ["send under a lock"],
745                "consequence": "a slow client stalls senders", "same_as": "La", "area": "concurrency", "supersedes": [], "reach": "team"}))
746        };
747        assert_eq!(learn::learn_in(&store, None, &ask, false, 20), (0, 1, 0));
748        let a = evidence(&store, &Filter::default()).remove(0);
749        assert_eq!((a.version, a.recurrence), (Some(2), Some(1)));
750        assert_eq!(summary(&store), (1, 1));
751        let _ = std::fs::remove_dir_all(store);
752    }
753
754    #[test]
755    fn the_arguments_parse() {
756        assert_eq!(
757            parse(&["La", "Lb", "--note", "skills/x", "--dir", "/w", "--json"]),
758            Some(Args { ids: vec!["La", "Lb"], dir: Some("/w"), note: Some("skills/x"), json: true, ..Args::default() })
759        );
760        assert_eq!(parse(&["--since"]), None);
761        assert_eq!(parse(&["--frobnicate"]), None);
762    }
763}