Skip to main content

borg/
learn.rs

1//! The conscience's live learning loop.
2//!
3//! **Capture** (hooks, deterministic, 0 tokens): an engineer correction is a prompt that pushes
4//! back on what the drone just proposed ("no…", "double check…", "why do you…"), or a rejected
5//! tool call / plan whose rejection carries the engineer's words. Each becomes a raw *episode*:
6//! the task, the drone's proposal, the correction.
7//!
8//! **Generalise** (`borg learn`, at session end, one model call per episode): the episode becomes a
9//! lesson (question + situations → consequence) or is merged into an existing lesson that
10//! expresses the same concern — the old wording is kept in `history` and `support` grows, so the
11//! library consolidates instead of piling up near-duplicates. Each lesson lists the engineers
12//! (git user.email) whose corrections it was learned from, its area (the kind of concern, for trust per
13//! engineer × area, trust.rs) and, when the correction contradicts a lesson it was shown, which
14//! lessons it `supersedes`; all from the same model call.
15//!
16//! **Outcomes** (`borg learn`, 0 tokens): a merge into a lesson the conscience asked earlier in the
17//! same session, or a new lesson superseding one it asked, means asking did not prevent the
18//! correction; a correction siding with one side of a rendered disagreement corrects the other
19//! sides instead; see outcome.rs.
20//!
21//! **Vetoes** (`borg no`, veto.rs): an engineer's "no" is an episode of kind `veto`, asked with its own
22//! prompt (`veto::OPENING`), whose answer may also name an invariant. It is learned at the end of any
23//! session, not only its own (a terminal veto may name an ended session). A veto merged into a lesson
24//! an earlier veto made counts as `veto_repeat`; an invariant of a lesson placed in this store becomes
25//! a proposed constraint in the queen's queue (`veto::propose`), never a constraint.
26
27use std::collections::{HashMap, HashSet};
28use std::io::Write;
29use std::path::{Path, PathBuf};
30use std::process::Command;
31use std::sync::LazyLock;
32
33use regex::Regex;
34use serde::{Deserialize, Serialize};
35use serde_json::{Value, json};
36
37use crate::conscience::{self, Challenge};
38use crate::outcome;
39// After `MAX_FAILURES` failed model calls in a row `borg learn` stops (runner.rs: the model is
40// unavailable, and asking about every remaining episode would only burn time and quota).
41use crate::runner::MAX_FAILURES;
42use crate::trust;
43
44pub mod regret;
45
46static CORRECTION: LazyLock<Regex> = LazyLock::new(|| {
47    Regex::new(r"(?i)^\s*(no\b|nope|don'?t|do not|stop|wrong|that'?s not|not what|why did you|why do you|i said|instead|you should|never\b|always\b|please don'?t|double check|carefully check|check against|are you sure|hmm|but\b|just a second|still\b|can'?t we|don'?t we|isn'?t)").unwrap()
48});
49static REJECTION: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"(?is)want to proceed with this tool use.*?the user said:\s*(.*)").unwrap());
50/// A tool call the engineer refused without saying why: omp's approval prompt ("Tool call denied by
51/// user: bash"), omp over ACP ("Tool call rejected by user (bash)"), Claude's plain rejection.
52static DENIAL: LazyLock<Regex> = LazyLock::new(|| {
53    Regex::new(r"(?i)^\s*(tool call denied by user: |tool call rejected by user \(|the user doesn'?t want to proceed with this tool use)").unwrap()
54});
55
56/// The longest excerpt an episode keeps of a correction, a rejected tool call, and the task as the
57/// learn prompt shows it. Guard: bounds what one episode adds to the store and to its prompt (a pasted
58/// log would otherwise go in whole); a regret's excerpts share it (regret.rs `CAP`).
59pub const EXCERPT_CHARS: usize = 1500;
60/// The session's task an episode keeps (hook.rs `capture_prompt`): the prompt it started from. Guard:
61/// bounds the `.task` state file and each episode's `task` (a pasted spec would otherwise go in whole);
62/// the learn prompt shows its first `EXCERPT_CHARS`.
63pub const TASK_CHARS: usize = 3000;
64/// The end of the drone's last proposal an episode keeps (hook.rs `proposal`). Guard: bounds one
65/// episode's `agent` text.
66pub const PROPOSAL_CHARS: usize = 2000;
67
68/// A prompt that pushes back on the drone's last proposal (same heuristic as the offline miner).
69pub fn is_correction(prompt: &str) -> bool {
70    let t = prompt.trim();
71    t.len() > 3 && !t.starts_with('<') && CORRECTION.is_match(t)
72}
73
74/// The engineer's words from a rejected tool call or plan, if any.
75pub fn rejection(error: &str) -> Option<String> {
76    let said = REJECTION.captures(error)?.get(1)?.as_str();
77    let said = said.split("Note: The user").next().unwrap_or(said).trim();
78    (said.len() > 3).then(|| said.chars().take(EXCERPT_CHARS).collect())
79}
80
81/// A refusal that carries no words (see `DENIAL`): the engineer's reason is their next prompt.
82pub fn denied(error: &str) -> bool {
83    rejection(error).is_none() && DENIAL.is_match(error)
84}
85
86#[derive(Debug, Clone, Default, Serialize, Deserialize)]
87pub struct Episode {
88    pub id: String,
89    pub ms: i64,
90    pub repo: String,
91    pub session: String,
92    pub task: String,
93    pub agent: String,
94    pub rejected_tool: Option<String>,
95    pub correction: String,
96    /// Who corrected: `git config user.email` in the episode's cwd, empty if unknown.
97    #[serde(default)]
98    pub engineer: String,
99    /// Empty for a correction; `regret::KIND` for a decision a later session fixed.
100    #[serde(default, skip_serializing_if = "String::is_empty")]
101    pub kind: String,
102    #[serde(default, skip_serializing_if = "Option::is_none")]
103    pub regret: Option<regret::Regret>,
104    /// How the engineer said it, where that is not a prompt or a rejected tool call: [`ANSWER`] for an
105    /// answer to the agent's own question (the questions and their options are the `agent` text).
106    #[serde(default, skip_serializing_if = "String::is_empty")]
107    pub via: String,
108    /// The harness whose transcript the episode was read from (`Harness::name`); empty for one the hook
109    /// captured live.
110    #[serde(default, skip_serializing_if = "String::is_empty")]
111    pub harness: String,
112    /// That harness's version as its transcript names it (Claude Code's `version`, omp's
113    /// `runtime.version`, opencode 1.x's session `version`, Codex's `cli_version`); empty where it names none.
114    #[serde(default, skip_serializing_if = "String::is_empty")]
115    pub harness_version: String,
116}
117
118/// `Episode::via` of an answer to the agent's own question, and its `why` in the usage statistics.
119pub const ANSWER: &str = "answer";
120/// `Episode::via` of a correction the engineer recorded in a collective by name (`borg correct
121/// --collective`): placed there explicitly, so `borg learn` never holds it for its repo's route
122/// ([`held`]).
123pub const CORRECT: &str = "correct";
124
125/// Two prompt episodes of one session this close together are the same correction. Method: a dedupe
126/// window; the live hook's capture time and the transcript's message time of one prompt differ by a
127/// second or two, two corrections of a session by far more. Episodes of a tool call are matched by
128/// their id and words instead ([`Known::holds`]).
129pub const SAME_MS: i64 = 5_000;
130
131/// The id of an episode read from a transcript: `<session8>-T<call>` for one about a tool call (the
132/// harness's own id for it: Claude's `tool_use_id`, omp's `toolCallId`, opencode's part id, Codex's
133/// `call_id`), else `<session8>-<ms>` as the hook names a prompt's. Content, never a position: a
134/// transcript still being written cannot shift it.
135pub fn transcript_id(session: &str, call: &str, ms: i64) -> String {
136    let s8: String = session.chars().take(8).collect();
137    if call.is_empty() { format!("{s8}-{ms}") } else { format!("{s8}-T{call}") }
138}
139
140/// A correction's words with its whitespace collapsed: the hook and a transcript may wrap or pad the
141/// same words differently.
142fn words(t: &str) -> String {
143    t.split_whitespace().collect::<Vec<_>>().join(" ")
144}
145
146/// What a store already holds, so an episode read again (the transcript sweep, session end,
147/// `borg assimilate`) is never appended twice. Read once, then grown with what is appended.
148#[derive(Debug, Default)]
149pub struct Known {
150    ids: HashSet<String>,
151    /// Per session: each episode's time, whether it was a prompt's, whether a transcript named it by its
152    /// tool call ([`from_call`]), and its words ([`words`]).
153    sessions: HashMap<String, Vec<Held>>,
154}
155
156impl Known {
157    /// The episodes of `store`'s `episodes.jsonl`.
158    pub fn read(store: &Path) -> Self {
159        let mut k = Known::default();
160        for e in read_jsonl::<Episode>(&store.join("episodes.jsonl")) {
161            k.add(&e);
162        }
163        k
164    }
165
166    /// Whether `e` is already held: an episode with its id; one of its session with the same words that no
167    /// transcript named by its call (the hook stamps capture time, a transcript message time, so their ids
168    /// differ; two calls the transcript named are two events, whatever their words); or, for a prompt's
169    /// episode, a prompt's of its session within [`SAME_MS`] (the hook and the transcript cut a prompt
170    /// alike, but an older store's wording may differ). An answer is matched by its id alone: only a
171    /// transcript records answers, and the same pick to two questions is two answers.
172    pub fn holds(&self, e: &Episode) -> bool {
173        if self.ids.contains(&e.id) {
174            return true;
175        }
176        let said = words(&e.correction);
177        let (prompt, answer) = (prompt_episode(e), e.via == ANSWER);
178        self.sessions
179            .get(&e.session)
180            .is_some_and(|seen| seen.iter().any(|h| (!answer && !h.from_call && h.words == said) || (prompt && h.prompt && (h.ms - e.ms).abs() < SAME_MS)))
181    }
182
183    /// Remember `e` as held.
184    pub fn add(&mut self, e: &Episode) {
185        self.ids.insert(e.id.clone());
186        let held = Held { ms: e.ms, prompt: prompt_episode(e), from_call: from_call(e), words: words(&e.correction) };
187        self.sessions.entry(e.session.clone()).or_default().push(held);
188    }
189}
190
191/// One held episode, as [`Known::holds`] compares it.
192#[derive(Debug)]
193struct Held {
194    ms: i64,
195    prompt: bool,
196    from_call: bool,
197    words: String,
198}
199
200/// Whether a transcript named `e` by its tool call ([`transcript_id`]'s `<session8>-T<call>`); the hook
201/// names its episodes `<session8>-<ms>`.
202fn from_call(e: &Episode) -> bool {
203    let s8: String = e.session.chars().take(8).collect();
204    e.id.strip_prefix(&s8).is_some_and(|rest| rest.starts_with("-T"))
205}
206
207/// An episode said in a prompt: a pushback or a veto, not a rejected tool call or an answer.
208fn prompt_episode(e: &Episode) -> bool {
209    e.rejected_tool.is_none() && e.via.is_empty() && e.kind != regret::KIND
210}
211
212impl AsMut<Episode> for Episode {
213    fn as_mut(&mut self) -> &mut Episode {
214        self
215    }
216}
217
218/// Appends to `store` each of `eps` it does not hold yet ([`Known`]), each counted as
219/// `record_episode_in` counts it, and returns those appended, in order. `ready` completes an episode
220/// (its repo, its engineer) just before it is appended, so one already held costs nothing. The one
221/// entry for transcript capture (the transcript sweep, `borg assimilate --apply`).
222///
223/// What the store holds is read under the lock of its `episodes.jsonl`, which every caller of this
224/// takes: a writer that read the store earlier (an assimilate that learned another collective
225/// meanwhile, a turn-end sweep of the same session) never appends an episode another one appended
226/// since.
227///
228/// # Errors
229/// When an episode cannot be appended; none after it is.
230pub fn record_new_in<T: AsMut<Episode>>(store: &Path, eps: Vec<T>, mut ready: impl FnMut(&mut T)) -> std::io::Result<Vec<T>> {
231    if eps.is_empty() {
232        return Ok(Vec::new());
233    }
234    std::fs::create_dir_all(store)?;
235    crate::live::locked(&store.join("episodes.jsonl"), || {
236        let mut known = Known::read(store);
237        let mut appended = Vec::new();
238        for mut x in eps {
239            if known.holds(x.as_mut()) {
240                continue;
241            }
242            ready(&mut x);
243            record_episode_in(store, x.as_mut())?;
244            known.add(x.as_mut());
245            appended.push(x);
246        }
247        Ok(appended)
248    })
249}
250
251/// A lesson as stored: the conscience's `Challenge` plus provenance and consolidation state.
252/// Append-only: a merge writes a new version with the same id; readers keep the latest.
253#[derive(Debug, Clone, Default, Serialize, Deserialize)]
254pub struct Lesson {
255    pub id: String,
256    pub origin_ms: i64,
257    pub repo: String,
258    pub question: String,
259    pub situations: Vec<String>,
260    pub consequence: String,
261    pub support: u32,
262    pub episodes: Vec<String>,
263    pub history: Vec<Value>,
264    pub version: u32,
265    /// Engineers whose corrections this lesson was learned from, first contributor first.
266    #[serde(default)]
267    pub engineers: Vec<String>,
268    /// When each later engineer joined (ms of their first episode merged in): trust.rs counts a
269    /// lesson's outcomes for an engineer only from then on. The first contributor is absent.
270    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
271    pub joined: std::collections::BTreeMap<String, i64>,
272    /// Kind of concern (trust.rs `AREAS`), set by the generalising model call; "other" if absent.
273    #[serde(default = "other_area")]
274    pub area: String,
275    /// Existing lessons this one replaces or contradicts; the conscience asks this one instead.
276    #[serde(default, skip_serializing_if = "Vec::is_empty")]
277    pub supersedes: Vec<String>,
278    /// "queen" for a queen answer (queen.rs), "import" for team knowledge read from a directory
279    /// (assimilate_dir.rs), kept across merges; empty for learned lessons.
280    #[serde(default, skip_serializing_if = "String::is_empty")]
281    pub source: String,
282    /// Where the lesson applies, as names (systems, repos, services: "Orion", "harbor_orion"). Empty:
283    /// wherever it matches. A scoped lesson is asked only when the drone's context names one of them
284    /// (conscience.rs `in_scope`), so a team rule for one system never reaches another.
285    #[serde(default, skip_serializing_if = "Vec::is_empty")]
286    pub scope: Vec<String>,
287    /// The outside rule or idiom this lesson replaces, and why ("upstream petri-net-design seeds a budget
288    /// with a reset arc; Orion drains instead because ..."). Lessons replace lessons via `supersedes`;
289    /// this names what is not a lesson (a vendored skill, upstream docs). Rendered with the question:
290    /// an override that does not name what it replaces loses to it (seen in an earlier eval).
291    #[serde(default, skip_serializing_if = "String::is_empty")]
292    pub overrides: String,
293    /// Source location of an imported lesson (`path:line` in the source repo); empty for learned ones.
294    #[serde(default, skip_serializing_if = "String::is_empty")]
295    pub origin: String,
296    /// Alternative phrasings of the lesson for search (facets.rs): the concept without names, the symptom,
297    /// the action, the consequence, German, the identifiers as aliases. Each is its own BM25 document
298    /// (conscience.rs `rank`); never shown to a drone. Empty for lessons learned before facets existed.
299    #[serde(default, skip_serializing_if = "Vec::is_empty")]
300    pub facets: Vec<String>,
301    /// Who should hear the lesson (reach.rs): empty for the team, `repo` for its repo only, `personal`
302    /// for a lesson in the store's local-only `personal.jsonl`, and for the tombstone that withdrew it
303    /// from `lessons.jsonl`.
304    #[serde(default, skip_serializing_if = "String::is_empty")]
305    pub reach: String,
306}
307
308fn other_area() -> String {
309    trust::OTHER.into()
310}
311
312pub fn home() -> PathBuf {
313    std::env::var("BORG_HOME").map(PathBuf::from).unwrap_or_else(|_| PathBuf::from(std::env::var("HOME").unwrap_or_default()).join(".local/share/borg"))
314}
315
316pub(crate) fn append_jsonl<T: Serialize>(path: &Path, v: &T) -> std::io::Result<()> {
317    append_line(path, &serde_json::to_string(v)?)
318}
319
320/// Appends `line` and its newline in one write. Under O_APPEND one write lands whole; `writeln!` makes
321/// two (the record, then "\n"), so two writers at once (two drones on one store) can leave
322/// `{A}{B}\n\n`, and every reader drops both unparseable records without a word.
323pub(crate) fn append_line(path: &Path, line: &str) -> std::io::Result<()> {
324    if let Some(d) = path.parent() {
325        std::fs::create_dir_all(d)?;
326    }
327    let mut f = std::fs::OpenOptions::new().create(true).append(true).open(path)?;
328    f.write_all(format!("{line}\n").as_bytes())
329}
330
331fn read_jsonl<T: for<'de> Deserialize<'de>>(path: &Path) -> Vec<T> {
332    std::fs::read_to_string(path).unwrap_or_default().lines().filter_map(|l| serde_json::from_str(l).ok()).collect()
333}
334
335/// The episodes of the `episodes.jsonl` at `path` by id, the first line of each; read line by line.
336pub fn episodes_of(path: &Path) -> HashMap<String, Episode> {
337    use std::io::BufRead;
338    let mut out = HashMap::new();
339    let Ok(f) = std::fs::File::open(path) else { return out };
340    for e in std::io::BufReader::new(f).lines().map_while(Result::ok).filter_map(|l| serde_json::from_str::<Episode>(&l).ok()) {
341        out.entry(e.id.clone()).or_insert(e);
342    }
343    out
344}
345
346/// False when BORG_LEARN=off (eval drones): nothing is captured or recorded.
347pub fn learning() -> bool {
348    std::env::var("BORG_LEARN").as_deref() != Ok("off")
349}
350
351/// The engineer working in `cwd`: their git user.email, empty when unknown.
352pub fn engineer(cwd: &str) -> String {
353    if cwd.is_empty() {
354        return String::new();
355    }
356    Command::new("git")
357        .args(["-C", cwd, "config", "user.email"])
358        .stdin(std::process::Stdio::null())
359        .stderr(std::process::Stdio::null())
360        .output()
361        .ok()
362        .filter(|o| o.status.success())
363        .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
364        .unwrap_or_default()
365}
366
367/// Records the hook's episode `e` in this session's store (`record_episode_locked`).
368///
369/// # Errors
370/// When the episode cannot be appended.
371pub fn record_episode(e: &Episode) -> std::io::Result<()> {
372    record_episode_locked(&home(), e)
373}
374
375/// `record_episode_in` under the store's `episodes.jsonl` lock (live.rs `locked`): a `borg move` of a
376/// repo's evidence rewrites that file under the lock, and an append landing between its read and its
377/// rename would be lost. Contended only by those moves and the captures that also lock it; uncontended,
378/// one lock taken on a path that writes anyway.
379///
380/// # Errors
381/// When the episode cannot be appended.
382pub fn record_episode_locked(store: &Path, e: &Episode) -> std::io::Result<()> {
383    std::fs::create_dir_all(store)?;
384    crate::live::locked(&store.join("episodes.jsonl"), || record_episode_in(store, e))
385}
386
387/// `record_episode` into `store` (the end-of-session job writes to the route's store).
388///
389/// # Errors
390/// When the episode cannot be appended.
391pub fn record_episode_in(store: &Path, e: &Episode) -> std::io::Result<()> {
392    append_jsonl(&store.join("episodes.jsonl"), e)?;
393    count_correction(store, e);
394    Ok(())
395}
396
397/// A correction in the usage statistics: a rejected tool call, a pushback, an answer to the agent's question
398/// (`why` [`ANSWER`], rolled up apart from the corrections, stats.rs `fold`) or a regret, with a short excerpt.
399pub fn count_correction(store: &Path, e: &Episode) {
400    // A veto counts itself (`stats::VETO`, veto.rs).
401    if e.kind == crate::veto::KIND {
402        return;
403    }
404    let snippet = crate::stats::excerpt(&e.correction, crate::stats::FEED_CHARS);
405    let (why, text) = match &e.rejected_tool {
406        _ if e.kind == regret::KIND => ("regret", snippet),
407        _ if e.via == ANSWER => (ANSWER, snippet),
408        Some(tool) => ("rejected_tool", format!("{tool}: {snippet}")),
409        None => ("pushback", snippet),
410    };
411    crate::stats::count(store, crate::stats::CORRECTION, &e.session, |ev| {
412        ev.ms = e.ms;
413        ev.engineer.clone_from(&e.engineer);
414        ev.repo.clone_from(&e.repo);
415        ev.why = why.into();
416        ev.text = text;
417        ev.source = crate::stats::source(&e.harness, &e.harness_version);
418    });
419}
420
421/// The lessons of the store `home` as this engineer has them: `lessons.jsonl` with the store's personal
422/// lessons (reach.rs `PERSONAL_FILE`) over it, a personal version winning over the team one of its id
423/// (the tombstone that withdrew it). For counts and "learned" (`learned_episodes`); `borg status`.
424pub fn store_lessons(home: &Path) -> Vec<Lesson> {
425    let mine = crate::reach::personal_lessons(home);
426    let mut all: Vec<Lesson> = lessons(&home.join("lessons.jsonl")).into_iter().filter(|l| !mine.iter().any(|m| m.id == l.id)).collect();
427    all.extend(mine);
428    all.sort_by_key(|l| l.origin_ms);
429    all
430}
431
432/// The episodes the store `home` has folded into a lesson, team or personal (tombstones included: they
433/// keep their episodes).
434pub fn learned_episodes(home: &Path) -> HashSet<String> {
435    let mine = crate::reach::personal_all(home);
436    lessons(&home.join("lessons.jsonl")).into_iter().chain(mine).flat_map(|l| l.episodes).collect()
437}
438
439/// The latest version of every lesson.
440pub fn lessons(path: &Path) -> Vec<Lesson> {
441    let mut by: HashMap<String, Lesson> = HashMap::new();
442    for l in read_jsonl::<Lesson>(path) {
443        if by.get(&l.id).is_none_or(|o| o.version <= l.version) {
444            by.insert(l.id.clone(), l);
445        }
446    }
447    let mut v: Vec<Lesson> = by.into_values().collect();
448    v.sort_by_key(|l| l.origin_ms);
449    v
450}
451
452/// The correction prompt's opening.
453const OPENING: &str = "An engineer corrected a coding agent. Decide whether the correction expresses a reusable engineering \
454lesson: a critical concern about quality, design, correctness, reuse, spec conformance or root cause that this engineer would \
455plausibly raise again about similar future work. Operational remarks (\"still working?\", \"stop\", \"push\", \"restart\") are NOT lessons.";
456
457/// Who should hear a lesson: shared by the correction and the regret prompt.
458const REACH_RULES: &str = "Then decide its reach: who should hear it. Ask: would a teammate working on the same code get this correction too?
459- personal: how this one engineer likes the agent to work with them, not a property of the code: commit message style \
460or length, AI attribution or co-author trailers, whether and when to commit, push, branch or open a PR, what to stage, \
461how to report or ask, language and tone, their own tooling habits.
462- repo: a convention of the repo it happened in that other repos do not share (its branch or deploy workflow, a tool \
463or command it mandates).
464- team: everything else: an engineering concern about the code, design, correctness or verification.";
465
466const FIELDS: &str = "If it is a lesson, write:
467- question: ONE question the agent should ask itself next time, general enough to transfer, keeping the technical substance;
468- situations: 1-3 lines, each `<action> on/in <kind of state or component> [under <condition>]`, without project/class names;
469- consequence: one line, what goes wrong or what the engineer requires instead;";
470
471/// The last field of either prompt.
472const AREA: &str = "- area: the one kind of concern: concurrency, spec-conformance, reuse-existing, architecture-boundary, verification, minimality, \
473protocol-correctness, data-source, security, dependency-version, scope, or other if none fits.";
474
475/// Merging and superseding: shared by the correction and the regret prompt.
476const MERGE_RULES: &str = "Existing lessons that may express the same concern are listed. If one of them IS the same concern (the same kind of mistake) \
477and agrees with this correction, set same_as to its id and write the merged, improved wording in question/situations/consequence. \
478Otherwise same_as is empty.
479
480If this correction contradicts or replaces a listed lesson (in the same situation it requires something different or the \
481opposite, e.g. the engineer reversed an earlier correction), list that lesson's id in supersedes; never the same_as lesson. \
482Otherwise supersedes is empty.";
483
484/// The prompt for one episode: a correction's, a veto's (veto.rs) or a regret's (regret.rs).
485fn episode_prompt(e: &Episode, listed: &[String]) -> String {
486    let existing = if listed.is_empty() { "(none)".into() } else { listed.join("\n") };
487    if e.kind == crate::veto::KIND {
488        let agent = if e.agent.trim().is_empty() { "(not available)" } else { e.agent.as_str() };
489        return format!(
490            "{}\n\n{REACH_RULES}\n\n{FIELDS}\n{AREA}\n{}\n\n{MERGE_RULES}\n\n## Task the agent had\n{}\n\n## What the agent proposed (excerpt)\n{agent}\n\n## Engineer's veto\n{}\n\n## Existing lessons\n{existing}\n",
491            crate::veto::OPENING,
492            crate::veto::INVARIANT,
493            e.task.chars().take(EXCERPT_CHARS).collect::<String>(),
494            e.correction,
495        );
496    }
497    if let Some(r) = e.regret.as_ref().filter(|_| e.kind == regret::KIND) {
498        let (head, sections) = (regret::head(r), regret::sections(r, &e.task));
499        let (reach, merge) = (regret::neutral(REACH_RULES), regret::neutral(MERGE_RULES));
500        return format!("{head}\n\n{reach}\n\n{}\n{AREA}\n\n{merge}\n\n{sections}## Existing lessons\n{existing}\n", regret::FIELDS);
501    }
502    format!(
503        "{OPENING}\n\n{REACH_RULES}\n\n{FIELDS}\n{AREA}\n\n{MERGE_RULES}\n\n## Task the agent had\n{}\n\n## What the agent proposed (excerpt)\n{}\n{}\n\n## Engineer's correction\n{}\n\n## Existing lessons\n{existing}\n",
504        e.task.chars().take(EXCERPT_CHARS).collect::<String>(),
505        e.agent,
506        if e.via == ANSWER { ANSWERED } else { e.rejected_tool.as_deref().unwrap_or("") },
507        e.correction,
508    )
509}
510
511/// Under an answer's proposal (the agent's own questions): what makes an answer a lesson.
512const ANSWERED: &str = "(The agent asked these questions; the correction below is the engineer's answer. An answer that only picks \
513what fits this one task is NOT a lesson; one that states a rule, a preference or a reason that holds beyond it is.)";
514
515fn schema() -> Value {
516    json!({"type": "object", "additionalProperties": false, "properties": {
517        "is_lesson": {"type": "boolean"}, "question": {"type": "string"},
518        "situations": {"type": "array", "items": {"type": "string"}}, "consequence": {"type": "string"},
519        "same_as": {"type": "string"}, "area": {"type": "string", "enum": trust::AREAS},
520        "supersedes": {"type": "array", "items": {"type": "string"}},
521        "reach": {"type": "string", "enum": ["team", "repo", "personal"]}},
522        "required": ["is_lesson", "question", "situations", "consequence", "same_as", "area", "supersedes", "reach"]})
523}
524
525/// `schema` plus the invariant a veto's reason may name (veto.rs).
526fn veto_schema() -> Value {
527    let mut s = schema();
528    s["properties"]["invariant"] = json!({"type": "string"});
529    if let Some(r) = s["required"].as_array_mut() {
530        r.push(json!("invariant"));
531    }
532    s
533}
534
535/// The answer schema for `prompt`: a veto's prompt (it starts with `veto::OPENING`) gets the invariant
536/// field; correction and regret calls stay as they were.
537pub fn schema_for(prompt: &str) -> Value {
538    if prompt.starts_with(crate::veto::OPENING) { veto_schema() } else { schema() }
539}
540
541/// The model call failed (quota, rate limit, not logged in, network, malformed answer): unlike a
542/// verdict of "not a lesson", the episode stays waiting and a later `borg learn` retries it.
543#[derive(Debug, Clone, Copy, PartialEq, Eq)]
544pub struct AskFailed;
545
546/// The model's verdict on an episode that is a lesson.
547struct Answer {
548    question: String,
549    situations: Vec<String>,
550    consequence: String,
551    area: String,
552    reach: String,
553    same_as: String,
554    supersedes: Vec<String>,
555    /// A veto's invariant (veto.rs); empty for every other episode, and when the model named none.
556    invariant: String,
557}
558
559impl Answer {
560    /// `Ok(None)` not a lesson; `Err` no usable verdict.
561    fn parse(v: &Value) -> Result<Option<Answer>, AskFailed> {
562        match v["is_lesson"].as_bool() {
563            Some(true) => {}
564            Some(false) => return Ok(None),
565            None => return Err(AskFailed),
566        }
567        let strs = |k: &str| v[k].as_array().into_iter().flatten().filter_map(|x| x.as_str().map(str::to_string)).collect::<Vec<_>>();
568        let text = |k: &str| v[k].as_str().unwrap_or_default().to_string();
569        let question = v["question"].as_str().filter(|q| !q.trim().is_empty()).ok_or(AskFailed)?.to_string();
570        let reach = v["reach"].as_str().filter(|r| [crate::reach::REPO, crate::reach::PERSONAL].contains(r)).unwrap_or_default();
571        Ok(Some(Answer {
572            question,
573            situations: strs("situations"),
574            consequence: text("consequence"),
575            area: trust::area(v["area"].as_str().unwrap_or_default()).to_string(),
576            reach: reach.to_string(),
577            same_as: text("same_as"),
578            supersedes: strs("supersedes"),
579            invariant: text("invariant"),
580        }))
581    }
582}
583
584/// The lessons most like an episode that the learn prompt offers for merging (`similar`). Baseline:
585/// the count since 0.1.0, never evaluated, the reference a later value is compared against. A duplicate
586/// ranked below it is not offered, so its episode can become a second lesson instead of a merge.
587const MERGE_CANDIDATES: usize = 5;
588
589/// The library's lessons most like the episode (at most `MERGE_CANDIDATES`), offered for merging.
590fn similar<'a>(e: &Episode, library: &'a [Lesson]) -> Vec<&'a Lesson> {
591    let as_challenges: Vec<Challenge> = library
592        .iter()
593        .map(|l| Challenge {
594            id: l.id.clone(),
595            origin_ms: l.origin_ms,
596            question: format!("{} {}", l.question, l.situations.join(" ")),
597            situations: l.situations.clone(),
598            consequence: l.consequence.clone(),
599            // A superseded lesson is not offered for merging; its live successor is, in its place. A
600            // re-reversal then contradicts the successor and becomes a new lesson superseding it.
601            supersedes: l.supersedes.clone(),
602            engineers: l.engineers.clone(),
603            source: l.source.clone(),
604            ..Default::default() // similarity for merging is about meaning, not usefulness or trust
605        })
606        .collect();
607    let query = format!("{} {} {}", e.task, e.agent, e.correction);
608    conscience::select(&as_challenges, &query, MERGE_CANDIDATES).iter().filter_map(|c| library.iter().find(|l| l.id == c.id)).collect()
609}
610
611/// `old` with the episode merged in: the new wording, the old one kept in `history`.
612fn merged(old: &Lesson, e: &Episode, a: Answer, mut supersedes: Vec<String>) -> Lesson {
613    let mut history = old.history.clone();
614    history.push(json!({"question": old.question, "consequence": old.consequence, "version": old.version}));
615    let mut episodes = old.episodes.clone();
616    episodes.push(e.id.clone());
617    let (mut engineers, mut joined) = (old.engineers.clone(), old.joined.clone());
618    // Emails compare case-insensitively; an alias (collective.rs "Engineers") is stored as
619    // its raw email and mapped when read, so the store never depends on this machine's config.
620    let same = |x: &String| x.trim().to_lowercase() == e.engineer.trim().to_lowercase();
621    if !e.engineer.trim().is_empty() && !engineers.iter().any(same) {
622        engineers.push(e.engineer.clone());
623        joined.insert(e.engineer.clone(), e.ms);
624    }
625    for id in old.supersedes.iter().rev() {
626        if !supersedes.contains(id) {
627            supersedes.insert(0, id.clone());
628        }
629    }
630    Lesson {
631        question: a.question,
632        situations: a.situations,
633        consequence: a.consequence,
634        support: old.support + 1,
635        episodes,
636        history,
637        version: old.version + 1,
638        engineers,
639        joined,
640        area: a.area,
641        supersedes,
642        // New wording: the old facets describe the old wording; `borg learn` generates new ones.
643        facets: Vec::new(),
644        reach: scoped(&old.repo, crate::reach::narrower(&old.reach, &a.reach)),
645        ..old.clone()
646    }
647}
648
649/// A new lesson from the episode alone.
650fn fresh(e: &Episode, a: Answer, supersedes: Vec<String>) -> Lesson {
651    Lesson {
652        id: format!("L{}", e.id),
653        origin_ms: e.ms,
654        repo: e.repo.clone(),
655        question: a.question,
656        situations: a.situations,
657        consequence: a.consequence,
658        support: 1,
659        episodes: vec![e.id.clone()],
660        version: 1,
661        engineers: Some(e.engineer.clone()).filter(|x| !x.is_empty()).into_iter().collect(),
662        area: a.area,
663        supersedes,
664        reach: scoped(&e.repo, &a.reach),
665        ..Lesson::default()
666    }
667}
668
669/// `reach` for a lesson of `repo`: one repo's reach needs a repo to be delivered in (reach.rs
670/// `deliverable`), so a lesson of none (a `borg correct` without `--repo`) is the team's instead of
671/// heard nowhere.
672fn scoped(repo: &str, reach: &str) -> String {
673    if reach == crate::reach::REPO && repo.trim().is_empty() { String::new() } else { reach.to_string() }
674}
675
676/// Generalise one episode against the existing library (the merge targets: team and personal lessons, no
677/// tombstone): `Ok(Some)` a new lesson or a merged version of an existing one, `Ok(None)` not a lesson,
678/// `Err` no verdict. `ask` performs the model call
679/// (injectable for tests). A regret (regret.rs) gets its own prompt, the same verdict.
680///
681/// # Errors
682/// `AskFailed` when the model gave no usable verdict.
683pub fn generalise(e: &Episode, library: &[Lesson], ask: &dyn Fn(&str) -> Option<Value>) -> Result<Option<Lesson>, AskFailed> {
684    generalise_with(e, library, ask).map(|o| o.map(|(l, _)| l))
685}
686
687/// `generalise`, with the invariant a veto's answer named (empty otherwise).
688///
689/// # Errors
690/// `AskFailed` when the model gave no usable verdict.
691pub fn generalise_with(e: &Episode, library: &[Lesson], ask: &dyn Fn(&str) -> Option<Value>) -> Result<Option<(Lesson, String)>, AskFailed> {
692    let similar = similar(e, library);
693    let listed: Vec<String> = similar.iter().map(|l| format!("[{}] {} -> {}", l.id, l.question, l.consequence)).collect();
694    let Some(a) = Answer::parse(&ask(&episode_prompt(e, &listed)).ok_or(AskFailed)?)? else { return Ok(None) };
695    let same_as = similar.iter().find(|l| a.same_as == l.id);
696    // Only lessons the model was shown can be superseded, and never the one it merges into.
697    let supersedes: Vec<String> =
698        a.supersedes.iter().filter(|id| similar.iter().any(|l| &l.id == *id) && same_as.is_none_or(|m| &m.id != *id)).cloned().collect();
699    let invariant = a.invariant.trim().to_string();
700    Ok(Some((
701        match same_as {
702            Some(old) => merged(old, e, a, supersedes),
703            None => fresh(e, a, supersedes),
704        },
705        invariant,
706    )))
707}
708
709/// One structured call to the drone's model (runner.rs picks claude, omp or opencode).
710pub fn ask_claude(prompt: &str) -> Option<Value> {
711    ask_model(prompt, &schema_for(prompt)).map(|(v, _)| v)
712}
713
714/// The model call behind `ask_claude`, for any JSON schema: the structured answer and its cost in USD
715/// (None when the CLI reported none). `BORG_LEARN_MODEL` picks the model, `runner::choose` the CLI.
716pub fn ask_model(prompt: &str, schema: &Value) -> Option<(Value, Option<f64>)> {
717    ask_model_for(crate::stats::purpose(), prompt, schema)
718}
719
720/// `ask_model`, its `MODEL_CALL` event counted under `purpose` instead of the process's (a learn run's
721/// cues, cues.rs, are `cues`).
722pub fn ask_model_for(purpose: &str, prompt: &str, schema: &Value) -> Option<(Value, Option<f64>)> {
723    let started = std::time::Instant::now();
724    let answer = call_model(prompt, schema);
725    crate::stats::count(&home(), crate::stats::MODEL_CALL, "", |ev| {
726        ev.why = purpose.into();
727        ev.usd = answer.as_ref().and_then(|(_, c)| *c).unwrap_or(0.0);
728        ev.dur_ms = started.elapsed().as_millis() as u64;
729        if answer.is_none() {
730            ev.text = "no answer".into();
731        }
732    });
733    answer
734}
735
736fn call_model(prompt: &str, schema: &Value) -> Option<(Value, Option<f64>)> {
737    let candidates = crate::runner::choose(std::env::var("BORG_LEARN_MODEL").ok().as_deref());
738    crate::runner::structured(&candidates, prompt, schema, learn_timeout())
739}
740
741/// How long one learn-side model call may take (`BORG_LEARN_TIMEOUT_S`, default 300 s): it runs in
742/// the background, but a hung runner must not keep `borg learn` alive forever.
743pub(crate) fn learn_timeout() -> std::time::Duration {
744    std::time::Duration::from_secs(std::env::var("BORG_LEARN_TIMEOUT_S").ok().and_then(|v| v.parse().ok()).unwrap_or(300))
745}
746
747pub fn now_ms() -> i64 {
748    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as i64).unwrap_or(0)
749}
750
751/// `borg learn [--session ID]`: generalise every episode not yet folded into a lesson.
752/// `--session` is given when that session ended: its asked lessons and the placebos the judge never
753/// reached are then settled. Any run also settles sessions idle for a day, so a missing `end` hook does
754/// not bias the outcomes.
755/// Returns (new lessons, merges, episodes still waiting because the model gave no verdict).
756pub fn learn_pending(session: Option<&str>, ask: &dyn Fn(&str) -> Option<Value>) -> (usize, usize, usize) {
757    learn_in(&home(), session, ask, learning(), now_ms())
758}
759
760/// `learn_pending` on the store in `home`. With `track`, outcomes of asked lessons are recorded
761/// (outcome.rs): a merge into a lesson asked earlier in the episode's session, and a lesson asked
762/// earlier that the new one supersedes, are `asked_then_corrected` (`outcome::contradicted`); at the end of `session`, and in every session idle for
763/// `outcome::IDLE_MS` at `now`, the other asked lessons are `asked_not_corrected`. So any run
764/// settles, not only the `end` hook's; a session it skipped episodes of is not settled as idle.
765/// Only a "not a lesson" verdict is recorded as skipped: an episode whose model call failed stays
766/// waiting (and its session unsettled) for the next run, and after `MAX_FAILURES` failures in a row
767/// the run stops asking. Returns (new lessons, merges, episodes still waiting for a verdict).
768pub fn learn_in(home: &Path, session: Option<&str>, ask: &dyn Fn(&str) -> Option<Value>, track: bool, now: i64) -> (usize, usize, usize) {
769    learn_in_reporting(home, session, ask, track, now, &mut |_| {})
770}
771
772/// A progress line on stderr: `[12/104 · 18 s · ~28 min left] new lesson L…: <question>`.
773pub fn print_step(st: Step) {
774    let left = (st.total - st.done) as f64 * st.secs / 60.0;
775    let mut v: String = st.verdict.chars().take(110).collect();
776    if v.len() < st.verdict.len() {
777        v.push('…');
778    }
779    eprintln!("[{}/{} · {:.0} s · ~{:.0} min left] {v}", st.done, st.total, st.secs, left);
780}
781
782/// One model verdict during a learn run, for a progress line: the episode's place among those this
783/// run asks about (`done` of `total`), and what came of it.
784pub struct Step<'a> {
785    pub done: usize,
786    pub total: usize,
787    pub verdict: &'a str,
788    pub secs: f64,
789}
790
791/// Whether a veto's lesson, placed as `placed`, proposes its `invariant` as a constraint: only a team
792/// lesson; a personal one never does, a preference being no invariant.
793fn proposes(placed: &crate::reach::Placed, invariant: &str) -> bool {
794    matches!(placed, crate::reach::Placed::Here(_)) && !invariant.trim().is_empty()
795}
796
797/// False when BORG_REGRET_CREDIT=off: a regret no longer counts against the lessons asked before the
798/// decision (`credit_regrets`).
799pub fn regret_credit() -> bool {
800    std::env::var("BORG_REGRET_CREDIT").as_deref() != Ok("off")
801}
802
803/// Crediting a session's own green test runs on (`outcome::tested`): learning is, and BORG_TEST_CREDIT is
804/// not `off` (an eval's frozen arm).
805pub fn test_credit() -> bool {
806    learning() && std::env::var("BORG_TEST_CREDIT").as_deref() != Ok("off")
807}
808
809/// Each regret episode generalised into a lesson of this store (`library`, as stored: a lesson placed in
810/// a private store is not in it) blames the lessons asked in its deciding session before the decision
811/// (`outcome::regretted`). Runs over every such episode on every run, so regrets learned before this
812/// existed are credited once too; the (cause, lesson) key keeps it idempotent. A regret the model judged
813/// "not a lesson", one still waiting, and one without `decision_ms` credit nothing. For a review regret
814/// the deciding session is the episode's own: a lesson `contradicted` already marked corrected there is
815/// not blamed twice.
816///
817/// Indexed once per run, so a run costs one pass over the log and the library plus each deciding
818/// session's own records: `regretted` reads only the deciding session's records, and a regret's credits
819/// carry that session too.
820///
821/// A regret is also a correction of the deciding session's work: the lesson it was folded into, and the
822/// lessons that one supersedes, if a graded check of that session found them applying before the
823/// decision and the session never asked them, were questions needed and not asked
824/// (`applied_then_corrected`, `outcome::missed`, at the decision's time and with the regret as `cause`),
825/// as for a correction merged into a lesson. These misses are recorded with `credit` off too: they blame
826/// no lesson asked.
827fn credit_regrets(regrets: &[Episode], library: &[Lesson], outcomes: &mut Vec<outcome::Outcome>, path: &Path, credit: bool) {
828    let mut holder: HashMap<&str, &Lesson> = HashMap::new();
829    for l in library {
830        for id in &l.episodes {
831            holder.entry(id.as_str()).or_insert(l);
832        }
833    }
834    let deciding: HashSet<&str> = regrets.iter().filter_map(|e| e.regret.as_ref()).map(|r| r.decision_session.as_str()).collect();
835    let mut by_session: HashMap<String, Vec<outcome::Outcome>> = HashMap::new();
836    for o in outcomes.iter().filter(|o| deciding.contains(o.session.as_str())) {
837        by_session.entry(o.session.clone()).or_default().push(o.clone());
838    }
839    for e in regrets {
840        let Some(r) = &e.regret else { continue };
841        let Some(holder) = holder.get(e.id.as_str()) else { continue };
842        let target: Vec<String> = std::iter::once(holder.id.clone()).chain(holder.supersedes.iter().cloned()).collect();
843        let records = by_session.entry(r.decision_session.clone()).or_default();
844        let blamed = match credit {
845            true => outcome::regretted(records, &r.decision_session, r.decision_ms, &e.id, &target, regret::strength(r), e.ms, &e.repo),
846            false => Vec::new(),
847        };
848        let missed = match r.decision_ms > 0 && !r.decision_session.is_empty() {
849            true => outcome::missed(records, &r.decision_session, r.decision_ms, Some(&holder.id), &holder.supersedes),
850            false => Vec::new(),
851        };
852        let missed = missed.into_iter().map(|l| outcome::Outcome {
853            kind: outcome::MISSED.into(),
854            session: r.decision_session.clone(),
855            lessons: vec![l],
856            ms: r.decision_ms,
857            repo: e.repo.clone(),
858            cause: e.id.clone(),
859            ..Default::default()
860        });
861        for o in blamed.into_iter().chain(missed.collect::<Vec<_>>()) {
862            let _ = outcome::append(path, &o);
863            records.push(o.clone());
864            outcomes.push(o);
865        }
866    }
867}
868
869/// Whether the session whose state file is `ctx` is still open at `now`: the file changed within
870/// `outcome::IDLE_MS` (every prompt and touch appends to it). No file: not open.
871fn open(ctx: &Path, now: i64) -> bool {
872    let ms = std::fs::metadata(ctx).and_then(|m| m.modified()).ok().and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()).map(|d| d.as_millis() as i64);
873    ms.is_some_and(|ms| ms > now - outcome::IDLE_MS)
874}
875
876/// The sessions besides `session` whose waiting episodes (not in `done` nor `skipped`) a run with
877/// `session` learns too: quiet ones, whose newest episode is `outcome::IDLE_MS` old at `now` and which
878/// `is_open` (its state file, `open`) says are not open. A session left open never sends its end, and
879/// the transcript sweep (sweep.rs `idle`) adds its last corrections after it went quiet. Each such
880/// session costs one stat.
881fn quiet(
882    episodes: &[Episode],
883    session: Option<&str>,
884    done: &HashSet<String>,
885    skipped: &HashSet<String>,
886    now: i64,
887    is_open: impl Fn(&str) -> bool,
888) -> HashSet<String> {
889    let Some(session) = session else { return HashSet::new() };
890    let mut newest: HashMap<&str, i64> = HashMap::new();
891    for e in episodes {
892        let n = newest.entry(e.session.as_str()).or_insert(e.ms);
893        *n = (*n).max(e.ms);
894    }
895    let waiting: HashSet<&str> =
896        episodes.iter().filter(|e| e.session != session && !done.contains(&e.id) && !skipped.contains(&e.id)).map(|e| e.session.as_str()).collect();
897    waiting.into_iter().filter(|s| newest[s] <= now - outcome::IDLE_MS && !is_open(s)).map(str::to_string).collect()
898}
899
900/// Episode kinds the removed hive made (a group's reflection, a rework), skipped where an older store has them.
901const RETIRED: [&str; 2] = ["group", "rework"];
902
903/// Where the episodes of a store route now, one lookup per repo (collective.rs `route_repo`).
904struct Routes<'a> {
905    cfg: Option<&'a crate::collective::Config>,
906    home: PathBuf,
907    base: PathBuf,
908    by_repo: HashMap<String, Option<crate::collective::Route>>,
909}
910
911impl<'a> Routes<'a> {
912    /// For the store `home` under `cfg`. Holds nothing without a config, nor in a store that is no
913    /// collective's (the root store, a test's or an eval's directory).
914    fn new(cfg: Option<&'a crate::collective::Config>, home: &Path) -> Self {
915        let cfg = cfg.filter(|c| crate::collective::is_store(c, home));
916        Routes { cfg, home: home.to_path_buf(), base: crate::collective::root_of(home), by_repo: HashMap::new() }
917    }
918
919    /// Where `e` belongs when that is not this store: its repo routes to another collective's store, or
920    /// off, under the current config. `None`: it belongs here, or its route is unknown (never guessed),
921    /// or the engineer placed it here by name ([`CORRECT`]).
922    fn elsewhere(&mut self, e: &Episode) -> Option<crate::collective::Route> {
923        let cfg = self.cfg?;
924        if e.via == CORRECT {
925            return None;
926        }
927        let (base, home) = (&self.base, &self.home);
928        let r = self.by_repo.entry(e.repo.clone()).or_insert_with(|| crate::collective::route_repo(cfg, &e.repo, base)).clone()?;
929        match &r {
930            crate::collective::Route::Collective { home: h, .. } if h == home => None,
931            _ => Some(r),
932        }
933    }
934}
935
936/// The waiting episodes of one repo that `borg learn` holds in a store because the repo routes elsewhere
937/// now (`held`).
938#[derive(Debug, Clone, PartialEq)]
939pub struct HeldRepo {
940    pub repo: String,
941    /// Where the repo routes now: another collective (its name and store) or `Off`.
942    pub route: crate::collective::Route,
943    /// The held episodes' ids, in log order.
944    pub episodes: Vec<String>,
945}
946
947/// The episodes of the store `home` that `borg learn` holds under `cfg`, per repo: not learned, not
948/// skipped, not recorded here by name ([`CORRECT`]), of a repo that routes to another store (or off)
949/// now. They wait, never learned here, until they are moved where they belong (`borg move <repo>`).
950/// Reads this store and the config only.
951pub fn held(home: &Path, cfg: Option<&crate::collective::Config>) -> Vec<HeldRepo> {
952    let mut routes = Routes::new(cfg, home);
953    if routes.cfg.is_none() {
954        return Vec::new();
955    }
956    let done = learned_episodes(home);
957    let skipped: HashSet<String> = std::fs::read_to_string(home.join("episodes-not-lessons.txt")).unwrap_or_default().lines().map(str::to_string).collect();
958    let mut out: Vec<HeldRepo> = Vec::new();
959    for e in read_jsonl::<Episode>(&home.join("episodes.jsonl")) {
960        if done.contains(&e.id) || skipped.contains(&e.id) || RETIRED.contains(&e.kind.as_str()) {
961            continue;
962        }
963        let Some(route) = routes.elsewhere(&e) else { continue };
964        match out.iter_mut().find(|h| h.repo == e.repo) {
965            Some(h) if !h.episodes.contains(&e.id) => h.episodes.push(e.id),
966            Some(_) => {}
967            None => out.push(HeldRepo { repo: e.repo, route, episodes: vec![e.id] }),
968        }
969    }
970    out
971}
972
973/// `learn_in`, calling `report` after every model call (the CLI prints it; the hooks pass a no-op).
974///
975/// One run per store at a time (an exclusive lock on `lessons.jsonl.lock`): session-end runs of several
976/// drones share the veto episodes and the regret credits, and two runs reading the store before either
977/// writes would learn a veto twice and count its credits twice. A waiting run then reads what the first
978/// one wrote and has nothing left to do for them.
979pub fn learn_in_reporting(
980    home: &Path,
981    session: Option<&str>,
982    ask: &dyn Fn(&str) -> Option<Value>,
983    track: bool,
984    now: i64,
985    report: &mut dyn FnMut(Step),
986) -> (usize, usize, usize) {
987    let cfg = crate::collective::load().ok().flatten();
988    learn_in_config(home, cfg.as_ref(), session, ask, track, now, report)
989}
990
991/// `learn_in_reporting` under an explicit collectives config: an episode whose repo routes to another
992/// store under `cfg` is held, never learned here ([`held`]).
993pub fn learn_in_config(
994    home: &Path,
995    cfg: Option<&crate::collective::Config>,
996    session: Option<&str>,
997    ask: &dyn Fn(&str) -> Option<Value>,
998    track: bool,
999    now: i64,
1000    report: &mut dyn FnMut(Step),
1001) -> (usize, usize, usize) {
1002    crate::live::locked(&home.join("lessons.jsonl"), || learn_unlocked(home, cfg, session, ask, track, now, report))
1003}
1004
1005fn learn_unlocked(
1006    home: &Path,
1007    cfg: Option<&crate::collective::Config>,
1008    session: Option<&str>,
1009    ask: &dyn Fn(&str) -> Option<Value>,
1010    track: bool,
1011    now: i64,
1012    report: &mut dyn FnMut(Step),
1013) -> (usize, usize, usize) {
1014    let (ep_path, lessons_path) = (home.join("episodes.jsonl"), home.join("lessons.jsonl"));
1015    let outcomes_path = outcome::path_beside(&lessons_path);
1016    let mut outcomes = if track { outcome::read(&outcomes_path) } else { Vec::new() };
1017    // The merge targets: this engineer's personal lessons (reach.rs) and the team's, no tombstone. Done:
1018    // every episode either file holds, tombstones included.
1019    let team = lessons(&lessons_path);
1020    let personal_lines = crate::reach::personal_all(home);
1021    let done: HashSet<String> = team.iter().chain(&personal_lines).flat_map(|l| l.episodes.clone()).collect();
1022    // A lesson `borg move` took elsewhere is no merge target here (reach.rs `MOVED`).
1023    let personal: Vec<Lesson> = personal_lines.into_iter().filter(|l| l.question != crate::reach::MOVED).collect();
1024    let mut personal_ids: HashSet<String> = personal.iter().map(|l| l.id.clone()).collect();
1025    let mut library: Vec<Lesson> = team.into_iter().filter(|l| l.reach != crate::reach::PERSONAL && !personal_ids.contains(&l.id)).chain(personal).collect();
1026    let skipped_path = home.join("episodes-not-lessons.txt");
1027    let mut skipped: HashSet<String> = std::fs::read_to_string(&skipped_path).unwrap_or_default().lines().map(str::to_string).collect();
1028    let (mut new, mut merged, mut failed, mut failures_in_row) = (0, 0, 0, 0);
1029    // Newest episode per session, and sessions with episodes this run left pending (--session).
1030    let mut last_episode: std::collections::HashMap<String, i64> = std::collections::HashMap::new();
1031    let mut pending: std::collections::HashSet<String> = std::collections::HashSet::new();
1032    // `RETIRED` kinds an older store holds are never learned.
1033    let episodes: Vec<Episode> = read_jsonl::<Episode>(&ep_path).into_iter().filter(|e| !RETIRED.contains(&e.kind.as_str())).collect();
1034    // Held (`held`): an episode of a repo that routes elsewhere now waits, unlearned and unmarked, and
1035    // keeps no session pending. Treated as settled for this run only, never written as skipped.
1036    let mut routes = Routes::new(cfg, home);
1037    let held_here: HashSet<String> =
1038        episodes.iter().filter(|e| !done.contains(&e.id) && !skipped.contains(&e.id) && routes.elsewhere(e).is_some()).map(|e| e.id.clone()).collect();
1039    skipped.extend(held_here);
1040    let regrets: Vec<Episode> = episodes.iter().filter(|e| e.kind == regret::KIND && e.regret.is_some()).cloned().collect();
1041    let state = std::env::var_os("BORG_STATE").map_or_else(|| home.join("state"), std::path::PathBuf::from);
1042    let ctx = |s: &str| state.join(format!("{}.ctx", crate::notes::safe(s)));
1043    // A veto is learned at the end of any session (veto.rs): a terminal veto may name one already ended.
1044    // So is a quiet session's episode (`quiet`): one that never sends its end would keep it waiting.
1045    let quiet = quiet(&episodes, session, &done, &skipped, now, |s| open(&ctx(s), now));
1046    let mine = |e: &Episode| session.is_none_or(|s| s == e.session) || e.kind == crate::veto::KIND || quiet.contains(&e.session);
1047    let vetoes: HashSet<String> = episodes.iter().filter(|e| e.kind == crate::veto::KIND).map(|e| e.id.clone()).collect();
1048    let total = episodes.iter().filter(|e| !done.contains(&e.id) && !skipped.contains(&e.id) && mine(e)).count();
1049    let mut asked = 0;
1050    for e in episodes {
1051        let last = last_episode.entry(e.session.clone()).or_insert(e.ms);
1052        *last = (*last).max(e.ms);
1053        if done.contains(&e.id) || skipped.contains(&e.id) {
1054            continue;
1055        }
1056        if !mine(&e) {
1057            pending.insert(e.session.clone());
1058            continue;
1059        }
1060        if failures_in_row >= MAX_FAILURES {
1061            failed += 1;
1062            pending.insert(e.session.clone());
1063            continue;
1064        }
1065        let started = std::time::Instant::now();
1066        let verdict = generalise_with(&e, &library, ask);
1067        asked += 1;
1068        let what = match &verdict {
1069            Err(AskFailed) => "no answer from the model, left pending".to_string(),
1070            Ok(Some((l, _))) if l.version > 1 => format!("merged into {}: {}", l.id, l.question),
1071            Ok(Some((l, _))) => format!("new lesson {}: {}", l.id, l.question),
1072            Ok(None) => "not a lesson".to_string(),
1073        };
1074        report(Step { done: asked, total, verdict: &what, secs: started.elapsed().as_secs_f64() });
1075        failures_in_row = if verdict.is_err() { failures_in_row + 1 } else { 0 };
1076        match verdict {
1077            Err(AskFailed) => {
1078                failed += 1;
1079                pending.insert(e.session.clone());
1080            }
1081            Ok(Some((l, invariant))) => {
1082                // A personal lesson stays in this store, for this engineer only (reach.rs `place`).
1083                let team = library.iter().find(|x| x.id == l.id && !personal_ids.contains(&x.id));
1084                let placed = crate::reach::place(&l, &e, personal_ids.contains(&l.id), team);
1085                if placed.lesson().version > 1 {
1086                    merged += 1
1087                } else {
1088                    new += 1
1089                }
1090                // Where it came from, for the feed: a review finding, a later fix, or a correction.
1091                let source = match &e.regret {
1092                    Some(r) if regret::is_review(r) => "review",
1093                    Some(_) => "regret",
1094                    None if e.kind == crate::veto::KIND => "veto",
1095                    None => "session",
1096                };
1097                let _ = crate::reach::write(home, &placed);
1098                let written = placed.lesson();
1099                let kind = if written.version > 1 { crate::stats::LESSON_MERGED } else { crate::stats::LESSON_NEW };
1100                let why = if matches!(placed, crate::reach::Placed::Personal { .. }) { "personal" } else { source };
1101                crate::stats::count(home, kind, &e.session, |ev| {
1102                    ev.engineer = e.engineer.clone();
1103                    ev.repo = e.repo.clone();
1104                    ev.lessons = vec![written.id.clone()];
1105                    ev.why = why.into();
1106                });
1107                if e.kind == crate::veto::KIND {
1108                    // The same "no" said twice: merged into a lesson an earlier veto made.
1109                    let earlier = library.iter().find(|x| x.id == l.id);
1110                    if l.version > 1 && earlier.is_some_and(|o| o.episodes.iter().any(|id| id != &e.id && vetoes.contains(id))) {
1111                        // Dated when the engineer said it, the week of its own `veto`, not when it was
1112                        // learned: a weekly share of repeated vetoes pairs counts of the same no's.
1113                        crate::stats::count(home, crate::stats::VETO_REPEAT, &e.session, |ev| {
1114                            ev.ms = if e.ms > 0 { e.ms } else { ev.ms };
1115                            ev.engineer = e.engineer.clone();
1116                            ev.repo = e.repo.clone();
1117                            ev.lessons = vec![l.id.clone()];
1118                            ev.text = crate::stats::excerpt(&e.correction, crate::stats::FEED_CHARS);
1119                        });
1120                    }
1121                    if proposes(&placed, &invariant) {
1122                        let _ = crate::veto::propose(home, &l, &invariant, &e, now);
1123                    }
1124                }
1125                let target = (l.version > 1).then_some(l.id.as_str());
1126                // A veto credited the lessons shown before it already (veto.rs): it never doubles as a
1127                // correction, nor as a missed question (`applied_then_corrected`).
1128                let track_results = track && e.kind != crate::veto::KIND;
1129                let contradicted = match track_results {
1130                    true => outcome::contradicted(&outcomes, &e.session, e.ms, target, &l.supersedes),
1131                    false => Vec::new(),
1132                };
1133                for id in contradicted {
1134                    let o = outcome::Outcome {
1135                        kind: outcome::CORRECTED.into(),
1136                        session: e.session.clone(),
1137                        lessons: vec![id],
1138                        ms: e.ms,
1139                        repo: e.repo.clone(),
1140                        ..Default::default()
1141                    };
1142                    let _ = outcome::append(&outcomes_path, &o);
1143                    outcomes.push(o);
1144                }
1145                let missed = match track_results {
1146                    true => outcome::missed(&outcomes, &e.session, e.ms, target, &l.supersedes),
1147                    false => Vec::new(),
1148                };
1149                for id in missed {
1150                    let o = outcome::Outcome {
1151                        kind: outcome::MISSED.into(),
1152                        session: e.session.clone(),
1153                        lessons: vec![id],
1154                        ms: e.ms,
1155                        repo: e.repo.clone(),
1156                        ..Default::default()
1157                    };
1158                    let _ = outcome::append(&outcomes_path, &o);
1159                    outcomes.push(o);
1160                }
1161                let written = placed.lesson().clone();
1162                if matches!(placed, crate::reach::Placed::Personal { .. }) {
1163                    personal_ids.insert(written.id.clone());
1164                }
1165                library.retain(|x| x.id != written.id);
1166                library.push(written);
1167            }
1168            Ok(None) => {
1169                let _ = append_line(&skipped_path, &e.id);
1170                crate::stats::count(home, crate::stats::NOT_A_LESSON, &e.session, |ev| ev.engineer = e.engineer.clone());
1171            }
1172        }
1173    }
1174    if track {
1175        credit_regrets(&regrets, &store_lessons(home), &mut outcomes, &outcomes_path, regret_credit());
1176        // Settled only once ended: its `end` hook ran (`session`), or it is idle by the log and its state
1177        // file is as quiet (`open`): a session working without asks or episodes for a day is still open.
1178        let mut ended: Vec<String> = outcome::idle_sessions(&outcomes, &last_episode, now)
1179            .into_iter()
1180            .filter(|s| !pending.contains(*s) && !open(&ctx(s), now))
1181            .map(str::to_string)
1182            .collect();
1183        if let Some(s) = session.filter(|s| !ended.iter().any(|x| x == s) && !pending.contains(*s)) {
1184            ended.push(s.to_string());
1185        }
1186        for s in &ended {
1187            // The job files a detached check of the hooks' session left (its process gone, or older than
1188            // a check may take): no later stop comes to remove them (check_async.rs `sweep`).
1189            if home == self::home().as_path() && ctx(s) == crate::hook::state_file(s) {
1190                crate::check_async::sweep(s);
1191            }
1192            // A session whose lock another process holds is still being judged (a session-end pass of a
1193            // resumed session, hook.rs `judge_session`, or a stop): settled once it lets go,
1194            // never under it, where a verdict would come after its settle.
1195            if crate::live::lock_state(&ctx(s), now_ms(), crate::live::stale_ms(), crate::live::alive) == crate::live::Lock::Held {
1196                continue;
1197            }
1198            // Its settle's time: after a verdict the session-end judge gives here.
1199            let mut at = now;
1200            // An idle session's end hook never ran: the session-end judge (hook.rs `judge_session`) gives
1201            // its open deliveries their last verdict here, before its tests are credited and the rest is
1202            // settled, as `borg learn --session` does for the session that ended (main.rs `learn_cmd`).
1203            // It writes to the hooks' store and reads their state file, so only for those; what it
1204            // judged it marks (`END_JUDGED`), so never twice.
1205            if Some(s.as_str()) != session && home == self::home().as_path() && ctx(s) == crate::hook::state_file(s) && crate::hook::judge_session(s) > 0 {
1206                outcomes = outcome::read(&outcomes_path);
1207                at = at.max(now_ms());
1208            }
1209            // The session's own tests that went green after an ask (hook.rs `capture_test`) credit the
1210            // lessons asked, once each, as it settles.
1211            if test_credit() {
1212                let tests = std::fs::read_to_string(state.join(format!("{}.tests", crate::notes::safe(s)))).unwrap_or_default();
1213                for o in outcome::tested(&outcomes, s, &tests) {
1214                    let _ = outcome::append(&outcomes_path, &o);
1215                    outcomes.push(o);
1216                }
1217            }
1218            // Each lesson left without a verdict says why (hook.rs `unjudged`, from the state file):
1219            // censored, never an ignore (memory.rs).
1220            let why = crate::hook::unjudged(&std::fs::read_to_string(ctx(s)).unwrap_or_default());
1221            for o in outcome::settle_with(&outcomes, s, at, &|l| why.get(l).copied().unwrap_or(outcome::UNJUDGED_NO_CALL)) {
1222                let _ = outcome::append(&outcomes_path, &o);
1223                outcomes.push(o);
1224            }
1225        }
1226        // The hooks' belief snapshot folded afresh, once per learn: it read the log whole anyway, and
1227        // this catches an edit inside a log that no checkpoint sees (facts.rs).
1228        crate::facts::rebuild(&lessons_path);
1229    }
1230    (new, merged, failed)
1231}
1232
1233#[cfg(test)]
1234mod tests {
1235    use super::*;
1236
1237    #[test]
1238    fn concurrent_jsonl_appends_keep_every_line() {
1239        let dir = std::env::temp_dir().join(format!("borg-learn-append-{}", std::process::id()));
1240        let _ = std::fs::remove_dir_all(&dir);
1241        let p = dir.join("episodes.jsonl");
1242        let (writers, each) = (8, 2000);
1243        std::thread::scope(|scope| {
1244            for w in 0..writers {
1245                let p = &p;
1246                scope.spawn(move || (0..each).for_each(|i| append_jsonl(p, &serde_json::json!({"w": w, "i": i})).unwrap()));
1247            }
1248        });
1249        assert_eq!(read_jsonl::<Value>(&p).len(), writers * each);
1250        let _ = std::fs::remove_dir_all(dir);
1251    }
1252
1253    #[test]
1254    fn a_regret_is_asked_with_its_own_prompt_and_the_shared_rules() {
1255        let r = regret::Regret {
1256            decision_task: "read the flag in b".into(),
1257            fix_subject: "fix: take the lock".into(),
1258            signals: vec!["subject".into()],
1259            age_days: 3,
1260            ..regret::Regret::default()
1261        };
1262        let e = Episode {
1263            kind: regret::KIND.into(),
1264            regret: Some(r),
1265            task: "make the flag safe".into(),
1266            ..ep("Rdddd-ffff", "Fixed later (3d, subject): fix: take the lock")
1267        };
1268        let shown = std::cell::RefCell::new(String::new());
1269        let l = generalise(&e, &[], &|p| {
1270            *shown.borrow_mut() = p.to_string();
1271            Some(json!({"is_lesson": true, "question": "Before reading a shared flag without a lock, has a concurrent writer been considered?",
1272                "situations": ["reading a shared flag across threads"], "consequence": "a race", "same_as": "", "area": "concurrency", "supersedes": []}))
1273        })
1274        .unwrap()
1275        .unwrap();
1276        let p = shown.into_inner();
1277        assert!(p.starts_with("A coding agent made a decision in an earlier session") && p.contains("3 days later"), "{p}");
1278        for part in [
1279            regret::neutral(REACH_RULES).as_str(),
1280            regret::FIELDS,
1281            AREA,
1282            regret::neutral(MERGE_RULES).as_str(),
1283            "## The decision (task)\nread the flag in b",
1284            "## The fix (subject, task)\nfix: take the lock\nmake the flag safe",
1285        ] {
1286            assert!(p.contains(part), "{part}");
1287        }
1288        assert!(!p.contains("## Engineer's correction") && !p.contains(OPENING) && !p.contains("correction"), "{p}");
1289        assert_eq!((l.id.as_str(), l.area.as_str(), l.episodes.as_slice()), ("LRdddd-ffff", "concurrency", &["Rdddd-ffff".to_string()][..]));
1290        // A correction keeps its prompt; a kind without its regret is asked as a correction.
1291        assert!(episode_prompt(&ep("e1", "no, the lock"), &[]).starts_with(OPENING));
1292        assert!(episode_prompt(&Episode { kind: regret::KIND.into(), ..ep("e2", "no") }, &[]).starts_with(OPENING));
1293    }
1294
1295    /// The injected model for vetoes: like `fake_ask` on the veto's section, plus `invariant:<rest>`.
1296    fn fake_veto_ask(p: &str) -> Option<Value> {
1297        let reason = p.split("## Engineer's veto\n").nth(1)?.lines().next()?;
1298        let after = |k: &str| reason.split(k).nth(1).and_then(|r| r.split_whitespace().next()).unwrap_or("").to_string();
1299        let invariant = reason.split("invariant:").nth(1).unwrap_or("").trim().to_string();
1300        Some(json!({"is_lesson": true, "question": format!("Q {reason}"), "situations": [], "consequence": "c",
1301            "same_as": after("merge:"), "supersedes": [], "invariant": invariant}))
1302    }
1303
1304    fn veto(id: &str, session: &str, ms: i64, reason: &str) -> Episode {
1305        Episode { kind: crate::veto::KIND.into(), ..at(id, session, ms, "ann@x", reason) }
1306    }
1307
1308    #[test]
1309    fn a_veto_is_asked_with_its_own_prompt_and_schema() {
1310        let p = episode_prompt(&veto("no-s-1", "s", 1, "never read the commit log for merges"), &[]);
1311        assert!(p.starts_with(crate::veto::OPENING), "{p}");
1312        for part in [
1313            "## Engineer's veto\nnever read the commit log for merges",
1314            crate::veto::INVARIANT,
1315            REACH_RULES,
1316            FIELDS,
1317            MERGE_RULES,
1318            "## Task the agent had\nplan the relay",
1319        ] {
1320            assert!(p.contains(part), "{part}");
1321        }
1322        assert!(!p.contains(OPENING) && !p.contains("## Engineer's correction"));
1323        let without = episode_prompt(&Episode { agent: String::new(), ..veto("no-s-2", "s", 1, "no") }, &[]);
1324        assert!(without.contains("## What the agent proposed (excerpt)\n(not available)"), "{without}");
1325        assert!(schema_for(&p)["required"].as_array().unwrap().contains(&json!("invariant")));
1326        assert!(schema_for(&p)["properties"]["invariant"].is_object());
1327        let correction = episode_prompt(&ep("e1", "no, the lock"), &[]);
1328        assert_eq!(schema_for(&correction), schema(), "a correction's call is unchanged");
1329        // The invariant comes back with the lesson; a correction's answer has none.
1330        let (l, inv) =
1331            generalise_with(&veto("no-s-3", "s", 1, "merge lookups invariant: lookups go through the review API"), &[], &fake_veto_ask).unwrap().unwrap();
1332        assert_eq!((l.id.as_str(), inv.as_str()), ("Lno-s-3", "lookups go through the review API"));
1333        let (_, none) = generalise_with(&ep("e2", "no, the lock merge:x"), &[], &fake_ask).unwrap().unwrap();
1334        assert!(none.is_empty());
1335    }
1336
1337    #[test]
1338    fn a_veto_is_learned_at_any_sessions_end_and_never_counted_as_a_correction() {
1339        let home = temp_home("veto-any");
1340        crate::learn::record_episode_in(&home, &veto("no-s1-1", "s1", 1, "never read the commit log for merges")).unwrap();
1341        append_jsonl(&home.join("episodes.jsonl"), &at("e1", "s1", 2, "ann@x", "a correction of s1")).unwrap();
1342        assert!(!crate::stats::events(&home).iter().any(|e| e.kind == crate::stats::CORRECTION), "count_correction skips a veto");
1343        // s2 ends: the veto of s1 is learned, s1's correction waits for its own session.
1344        assert_eq!(learn_in(&home, Some("s2"), &fake_veto_ask, true, 3), (1, 0, 0));
1345        let ls = lessons(&home.join("lessons.jsonl"));
1346        assert_eq!(ls.iter().map(|l| l.id.as_str()).collect::<Vec<_>>(), ["Lno-s1-1"]);
1347        let news: Vec<crate::stats::Event> = crate::stats::events(&home).into_iter().filter(|e| e.kind == crate::stats::LESSON_NEW).collect();
1348        assert_eq!(news[0].why, "veto");
1349        let _ = std::fs::remove_dir_all(home);
1350    }
1351
1352    #[test]
1353    fn a_veto_merged_into_an_earlier_vetos_lesson_is_a_repeat() {
1354        let home = temp_home("veto-repeat");
1355        let repeats = |home: &Path| crate::stats::events(home).iter().filter(|e| e.kind == crate::stats::VETO_REPEAT).count();
1356        // A lesson from a correction, then a veto merged into it: no repeat (the first "no").
1357        append_jsonl(&home.join("episodes.jsonl"), &at("e0", "s0", 1, "ann@x", "use the lock on the flag")).unwrap();
1358        append_jsonl(&home.join("episodes.jsonl"), &veto("no-s1-2", "s1", 2, "the lock again merge:Le0")).unwrap();
1359        // Le0 was asked in s1 before the veto: the veto credits it (veto.rs), learning never adds a correction.
1360        let asked = outcome::Outcome { kind: outcome::ASKED.into(), session: "s1".into(), lessons: vec!["Le0".into()], ms: 1, ..Default::default() };
1361        outcome::append(&home.join("outcomes.jsonl"), &asked).unwrap();
1362        learn_in(&home, None, &|p| if p.starts_with(crate::veto::OPENING) { fake_veto_ask(p) } else { fake_ask(p) }, true, 3);
1363        assert_eq!(repeats(&home), 0);
1364        let results = outcome::read(&home.join("outcomes.jsonl"));
1365        assert!(!results.iter().any(|o| o.kind == outcome::CORRECTED || o.kind == outcome::MISSED), "{results:?}");
1366        assert!(!crate::stats::events(&home).iter().any(|e| e.kind == crate::stats::CORRECTED));
1367        // A second veto with the same reason: merged into the lesson a veto is part of.
1368        append_jsonl(&home.join("episodes.jsonl"), &veto("no-s2-4", "s2", 4, "the lock, third time merge:Le0")).unwrap();
1369        assert_eq!(learn_in(&home, Some("s2"), &fake_veto_ask, true, 5), (0, 1, 0));
1370        assert_eq!(repeats(&home), 1);
1371        let r = crate::stats::events(&home).into_iter().find(|e| e.kind == crate::stats::VETO_REPEAT).unwrap();
1372        assert_eq!((r.session.as_str(), r.lessons.clone()), ("s2", vec!["Le0".to_string()]));
1373        assert_eq!(r.ms, 4, "dated when the veto was said, the week of its own veto");
1374        let _ = std::fs::remove_dir_all(home);
1375    }
1376
1377    #[test]
1378    fn an_invariant_of_a_veto_becomes_a_proposed_constraint_never_a_constraint() {
1379        let home = temp_home("veto-invariant");
1380        append_jsonl(&home.join("episodes.jsonl"), &veto("no-s1-1", "s1", 1, "no commit log invariant: merge lookups use the review API")).unwrap();
1381        append_jsonl(&home.join("episodes.jsonl"), &veto("no-s1-2", "s1", 2, "not now, later")).unwrap();
1382        learn_in(&home, Some("s1"), &fake_veto_ask, true, 3);
1383        let queue = crate::queen::entries(&home);
1384        assert_eq!(queue.len(), 1, "only the veto that named an invariant");
1385        assert_eq!((queue[0].kind, queue[0].session.as_str()), (crate::queen::Kind::Constraint, "s1"));
1386        assert!(
1387            queue[0].evidence.contains(&"invariant: merge lookups use the review API".to_string())
1388                && queue[0].evidence.contains(&"lesson Lno-s1-1".to_string())
1389        );
1390        assert!(!crate::veto::constraints_path(&home).exists());
1391        // A personal lesson never proposes; nor does one without an invariant.
1392        let l = Lesson { id: "L".into(), ..Lesson::default() };
1393        let personal = crate::reach::Placed::Personal { lesson: Box::new(l.clone()), withdraw: None };
1394        assert!(proposes(&crate::reach::Placed::Here(Box::new(l)), "x") && !proposes(&personal, "x"));
1395        assert!(!proposes(&crate::reach::Placed::Here(Box::default()), " "));
1396        let _ = std::fs::remove_dir_all(home);
1397    }
1398
1399    #[test]
1400    fn detects_corrections_and_rejections() {
1401        assert!(is_correction("double check against spec document"));
1402        assert!(is_correction("No, let's not move this into the net"));
1403        assert!(!is_correction("let's pick up the gpt live plan"));
1404        let err = "The user doesn't want to proceed with this tool use. The tool use was rejected. the user said:\ncarefully check TOCTOU\nNote: The user's next message may contain a correction";
1405        assert_eq!(rejection(err).as_deref(), Some("carefully check TOCTOU"));
1406        assert_eq!(rejection("Exit code 1\nTraceback"), None);
1407        // omp's refusals carry no words; the reason follows as the next prompt.
1408        assert!(denied("Tool call denied by user: bash"));
1409        assert!(denied("Tool call rejected by user (edit)"));
1410        assert!(denied(
1411            "The user doesn't want to proceed with this tool use. The tool use was rejected (eg. if it was a file edit, the new_string was NOT written to the file). STOP what you are doing and wait for the user to tell you how to proceed."
1412        ));
1413        assert!(!denied(err), "a rejection with words is captured directly");
1414        assert!(!denied("Exit code 1\nTool call denied by user: bash"), "only the harness's own message, not tool output");
1415    }
1416
1417    #[test]
1418    fn known_holds_an_id_the_same_words_or_a_prompt_moments_apart() {
1419        let at = |id: &str, ms: i64, tool: Option<&str>, said: &str| Episode {
1420            id: id.into(),
1421            ms,
1422            session: "s1".into(),
1423            rejected_tool: tool.map(String::from),
1424            correction: said.into(),
1425            ..Episode::default()
1426        };
1427        let mut k = Known::default();
1428        // What the hook recorded: a rejection at capture time, a prompt.
1429        k.add(&at("s1-100000", 100_000, Some("Edit {}"), "use backoff,\n  not a fixed sleep"));
1430        k.add(&at("s1-200000", 200_000, None, "no, the other crate"));
1431        assert!(k.holds(&at("s1-Ttoolu_1", 70_000, Some("Edit {}"), "use backoff, not a fixed sleep")), "same words, whitespace aside");
1432        assert!(k.holds(&at("s1-100000", 1, None, "anything")), "same id");
1433        assert!(k.holds(&at("s1-201500", 201_500, None, "no, the other crate!")), "a prompt within SAME_MS");
1434        assert!(!k.holds(&at("s1-Ttoolu_2", 201_000, Some("Bash {}"), "keep the cache")), "a tool call's episode needs its id or its words");
1435        assert!(!k.holds(&Episode { via: ANSWER.into(), ..at("s1-Ttoolu_3", 200_500, None, "Exponential") }), "nor does an answer");
1436        assert!(!k.holds(&Episode { session: "s2".into(), ..at("s2-Tx", 100_000, Some("Edit {}"), "use backoff, not a fixed sleep") }));
1437        // Two calls the transcript named are two events, though their words match: a second rejection
1438        // with the same words, the same pick to another question.
1439        let mut t = Known::default();
1440        t.add(&at("s1-Ttoolu_4", 300_000, Some("Edit {}"), "keep it in the relay crate"));
1441        t.add(&Episode { via: ANSWER.into(), ..at("s1-Ttoolu_5", 400_000, None, "Create, no parent") });
1442        assert!(!t.holds(&at("s1-Ttoolu_6", 900_000, Some("Edit {}"), "keep it in the relay crate")), "another rejected call");
1443        assert!(!t.holds(&Episode { via: ANSWER.into(), ..at("s1-Ttoolu_7", 900_000, None, "Create, no parent") }), "another question");
1444        // An answer is never matched to a hook's episode by words.
1445        assert!(!k.holds(&Episode { via: ANSWER.into(), ..at("s1-Ttoolu_8", 900_000, None, "no, the other crate") }));
1446        assert_eq!(transcript_id("abcdefghij", "toolu_9", 5), "abcdefgh-Ttoolu_9");
1447        assert_eq!(transcript_id("abcdefghij", "", 5), "abcdefgh-5");
1448
1449        // record_new_in appends once, and once within one batch.
1450        let store = std::env::temp_dir().join(format!("borg-known-{}", std::process::id()));
1451        let _ = std::fs::remove_dir_all(&store);
1452        let e = at("s1-Tq", 5, None, "Exponential");
1453        let mut readied = 0;
1454        assert_eq!(record_new_in(&store, vec![e.clone(), e.clone()], |_| readied += 1).unwrap().len(), 1);
1455        assert!(record_new_in(&store, vec![e.clone()], |_| panic!("a held episode is not readied")).unwrap().is_empty());
1456        assert_eq!(readied, 1);
1457        assert!(Known::read(&store).holds(&e));
1458        assert_eq!(read_jsonl::<Episode>(&store.join("episodes.jsonl")).len(), 1);
1459        let _ = std::fs::remove_dir_all(&store);
1460    }
1461
1462    #[test]
1463    fn writers_at_once_append_an_episode_once() {
1464        let store = std::env::temp_dir().join(format!("borg-known-race-{}", std::process::id()));
1465        let _ = std::fs::remove_dir_all(&store);
1466        let e = Episode { id: "s1-Tx".into(), session: "s1".into(), correction: "keep the cache dir".into(), ..Episode::default() };
1467        let start = std::sync::Barrier::new(8);
1468        std::thread::scope(|s| {
1469            for _ in 0..8 {
1470                s.spawn(|| {
1471                    start.wait();
1472                    record_new_in(&store, vec![e.clone()], |_| std::thread::yield_now()).unwrap();
1473                });
1474            }
1475        });
1476        assert_eq!(read_jsonl::<Episode>(&store.join("episodes.jsonl")).len(), 1);
1477        let _ = std::fs::remove_dir_all(&store);
1478    }
1479
1480    #[test]
1481    fn an_answer_is_asked_as_an_answer_and_counted_as_one() {
1482        let answer = Episode { via: ANSWER.into(), agent: "Which backoff?\n  options: Exponential | Linear".into(), ..ep("a1", "Exponential") };
1483        let p = episode_prompt(&answer, &[]);
1484        assert!(p.contains(ANSWERED) && p.contains("## Engineer's correction\nExponential"), "{p}");
1485        assert!(!episode_prompt(&ep("e1", "no"), &[]).contains(ANSWERED));
1486        let store = std::env::temp_dir().join(format!("borg-answer-count-{}", std::process::id()));
1487        let _ = std::fs::remove_dir_all(&store);
1488        count_correction(&store, &answer);
1489        let ev = crate::stats::live(&store);
1490        assert_eq!((ev.len(), ev[0].why.as_str()), (1, ANSWER));
1491        let _ = std::fs::remove_dir_all(&store);
1492    }
1493
1494    fn ep(id: &str, correction: &str) -> Episode {
1495        Episode {
1496            id: id.into(),
1497            ms: 1,
1498            repo: "r".into(),
1499            session: "s".into(),
1500            task: "plan the relay".into(),
1501            agent: "one ReentrantLock guards delivery".into(),
1502            rejected_tool: None,
1503            correction: correction.into(),
1504            ..Episode::default()
1505        }
1506    }
1507
1508    /// The injected model: `same_as` is the lesson id after "merge:" in the correction, if any, and
1509    /// `supersedes` the one after "supersede:".
1510    fn fake_ask(p: &str) -> Option<Value> {
1511        let correction = p.split("## Engineer's correction\n").nth(1)?.lines().next()?;
1512        let after = |k: &str| correction.split(k).nth(1).and_then(|r| r.split_whitespace().next()).unwrap_or("").to_string();
1513        let supersedes: Vec<String> = Some(after("supersede:")).filter(|x| !x.is_empty()).into_iter().collect();
1514        Some(json!({"is_lesson": true, "question": format!("Q {correction}"), "situations": [], "consequence": "c",
1515            "same_as": after("merge:"), "supersedes": supersedes}))
1516    }
1517
1518    fn results(home: &Path) -> Vec<(String, String, String)> {
1519        outcome::read(&home.join("outcomes.jsonl")).into_iter().filter(|o| o.kind != outcome::ASKED).map(|o| (o.kind, o.session, o.lessons.join(","))).collect()
1520    }
1521
1522    #[test]
1523    fn every_model_call_is_reported_with_its_place_and_verdict() {
1524        let home = temp_home("progress");
1525        for e in [at("e0", "s0", 1, "ann@x", "use the lock on the flag"), at("e1", "s0", 2, "bob@x", "the lock again merge:Le0")] {
1526            append_jsonl(&home.join("episodes.jsonl"), &e).unwrap();
1527        }
1528        let mut steps = Vec::new();
1529        learn_in_reporting(&home, None, &fake_ask, false, 3, &mut |st| steps.push((st.done, st.total, st.verdict.split(':').next().unwrap().to_string())));
1530        assert_eq!(steps, [(1, 2, "new lesson Le0".to_string()), (2, 2, "merged into Le0".to_string())]);
1531        // A second run has nothing left to ask, so it reports nothing.
1532        steps.clear();
1533        learn_in_reporting(&home, None, &fake_ask, false, 4, &mut |st| steps.push((st.done, st.total, String::new())));
1534        assert!(steps.is_empty());
1535        let _ = std::fs::remove_dir_all(home);
1536    }
1537
1538    #[test]
1539    fn a_lesson_contradicted_after_asking_is_corrected_not_followed() {
1540        let home = temp_home("contradicted");
1541        append_jsonl(&home.join("episodes.jsonl"), &at("e0", "s0", 1, "ann@x", "use the lock on the flag")).unwrap();
1542        learn_in(&home, Some("s0"), &fake_ask, true, 2);
1543        // s1 asks Le0, then the engineer contradicts it: a new lesson superseding Le0.
1544        asked(&home, "s1", 10, &["Le0"]);
1545        append_jsonl(&home.join("episodes.jsonl"), &at("e1", "s1", 20, "bob@x", "no, CAS on the flag supersede:Le0")).unwrap();
1546        assert_eq!(learn_in(&home, Some("s1"), &fake_ask, true, 30), (1, 0, 0));
1547        assert_eq!(results(&home), [(outcome::CORRECTED.into(), "s1".into(), "Le0".into())]);
1548        let _ = std::fs::remove_dir_all(home);
1549    }
1550
1551    #[test]
1552    fn a_correction_of_a_lesson_that_applied_but_was_not_asked_is_a_miss() {
1553        let home = temp_home("missed");
1554        for e in [at("e0", "s0", 1, "ann@x", "use the lock on the flag"), at("e1", "s0", 2, "bob@x", "retry with jitter")] {
1555            append_jsonl(&home.join("episodes.jsonl"), &e).unwrap();
1556        }
1557        learn_in(&home, Some("s0"), &fake_ask, true, 3);
1558        // s1's check graded both as applying and chose to ask only Le1; the engineer then corrects the
1559        // drone on Le0's concern: Le0 was needed and not asked.
1560        let graded = outcome::Outcome {
1561            kind: outcome::GRADED.into(),
1562            session: "s1".into(),
1563            lessons: vec!["Le0".into(), "Le1".into()],
1564            ms: 10,
1565            repo: "r".into(),
1566            grades: vec![2, 3],
1567            ranks: vec![0, 1],
1568            applied: 2,
1569            chosen: vec!["Le1".into()],
1570            ..Default::default()
1571        };
1572        outcome::append(&home.join("outcomes.jsonl"), &graded).unwrap();
1573        asked(&home, "s1", 11, &["Le1"]);
1574        append_jsonl(&home.join("episodes.jsonl"), &at("e2", "s1", 20, "cy@x", "no, the lock on the flag merge:Le0")).unwrap();
1575        assert_eq!(learn_in(&home, Some("s1"), &fake_ask, true, 30), (0, 1, 0));
1576        let r = results(&home);
1577        assert!(r.contains(&(outcome::MISSED.into(), "s1".into(), "Le0".into())), "{r:?}");
1578        // The asked one is not a miss; a second run records nothing new.
1579        assert!(!r.iter().any(|(k, _, l)| k == outcome::MISSED && l == "Le1"), "{r:?}");
1580        learn_in(&home, Some("s1"), &fake_ask, true, 40);
1581        assert_eq!(results(&home).iter().filter(|(k, ..)| k == outcome::MISSED).count(), 1);
1582        let _ = std::fs::remove_dir_all(home);
1583    }
1584
1585    #[test]
1586    fn siding_with_one_side_of_a_disagreement_corrects_the_other() {
1587        let home = temp_home("sides");
1588        for e in [at("e0", "s0", 1, "ann@x", "use the lock on the flag"), at("e1", "s0", 2, "bob@x", "use CAS on the flag")] {
1589            append_jsonl(&home.join("episodes.jsonl"), &e).unwrap();
1590        }
1591        learn_in(&home, Some("s0"), &fake_ask, true, 3);
1592        // The conscience rendered Le0 and Le1 as one disagreement; the engineer sides with Le0.
1593        let o = outcome::Outcome {
1594            kind: outcome::ASKED.into(),
1595            session: "s1".into(),
1596            lessons: vec!["Le0".into(), "Le1".into()],
1597            ms: 10,
1598            repo: "r".into(),
1599            groups: vec![vec!["Le0".into(), "Le1".into()]],
1600            ..Default::default()
1601        };
1602        outcome::append(&home.join("outcomes.jsonl"), &o).unwrap();
1603        append_jsonl(&home.join("episodes.jsonl"), &at("e2", "s1", 20, "cy@x", "no, the lock on the flag merge:Le0")).unwrap();
1604        assert_eq!(learn_in(&home, Some("s1"), &fake_ask, true, 30), (0, 1, 0));
1605        assert_eq!(results(&home), [(outcome::CORRECTED.into(), "s1".into(), "Le1".into()), (outcome::NOT_CORRECTED.into(), "s1".into(), "Le0".into())]);
1606        // cy joined Le0 at the correction; ann's earlier evidence is not his.
1607        let l0 = lessons(&home.join("lessons.jsonl")).into_iter().find(|l| l.id == "Le0").unwrap();
1608        assert_eq!(l0.joined.get("cy@x"), Some(&20));
1609        let _ = std::fs::remove_dir_all(home);
1610    }
1611
1612    #[test]
1613    fn the_learn_prompt_offers_at_most_merge_candidates_lessons_for_merging() {
1614        let lib: Vec<Lesson> = (0..MERGE_CANDIDATES + 2)
1615            .map(|i| Lesson {
1616                id: format!("L{i}"),
1617                origin_ms: i as i64,
1618                question: format!("Is the ReentrantLock held while delivery {i} runs?"),
1619                consequence: "c".into(),
1620                version: 1,
1621                ..Lesson::default()
1622            })
1623            .collect();
1624        let e = ep("e1", "no, the ReentrantLock must guard delivery");
1625        let offered = similar(&e, &lib);
1626        assert_eq!(offered.len(), MERGE_CANDIDATES, "every lesson matches; the prompt offers the best few");
1627        let prompt = std::cell::RefCell::new(String::new());
1628        let _ = generalise(&e, &lib, &|p| {
1629            *prompt.borrow_mut() = p.to_string();
1630            None
1631        });
1632        let named = lib.iter().filter(|l| prompt.borrow().contains(&format!("[{}]", l.id))).count();
1633        assert_eq!(named, MERGE_CANDIDATES, "{}", prompt.borrow());
1634    }
1635
1636    #[test]
1637    fn a_superseded_lesson_is_not_a_merge_target_its_successor_is_shown() {
1638        let lesson = |id: &str, ms: i64, q: &str, sup: &[&str]| Lesson {
1639            id: id.into(),
1640            origin_ms: ms,
1641            repo: "r".into(),
1642            question: q.into(),
1643            situations: vec![],
1644            consequence: "c".into(),
1645            support: 1,
1646            episodes: vec![],
1647            history: vec![],
1648            version: 1,
1649            engineers: vec!["ann@x".into()],
1650            joined: Default::default(),
1651            area: "concurrency".into(),
1652            supersedes: sup.iter().map(|s| s.to_string()).collect(),
1653            source: String::new(),
1654            scope: vec![],
1655            overrides: String::new(),
1656            origin: String::new(),
1657            facets: vec![],
1658            reach: String::new(),
1659        };
1660        let lib = [lesson("L0", 1, "Is the closed flag guarded by the ReentrantLock?", &[]), lesson("L1", 2, "Would a CAS suffice?", &["L0"])];
1661        // ann reverts to the lock: the model only sees L1 (in L0's place), contradicts it, and the
1662        // restored view becomes a new lesson that supersedes L1 and is asked again.
1663        let shown = std::cell::RefCell::new(String::new());
1664        let back = generalise(&ep("e9", "no, guard the closed flag with the ReentrantLock"), &lib, &|p| {
1665            *shown.borrow_mut() = p.to_string();
1666            Some(json!({"is_lesson": true, "question": "Is the closed flag guarded by the ReentrantLock?", "situations": [],
1667                "consequence": "c", "same_as": "L0", "area": "concurrency", "supersedes": ["L1"]}))
1668        })
1669        .unwrap()
1670        .unwrap();
1671        let shown = shown.into_inner();
1672        assert!(shown.contains("[L1]") && !shown.contains("[L0]"), "{shown}");
1673        assert_eq!((back.version, back.supersedes.clone()), (1, vec!["L1".to_string()]), "same_as on a hidden lesson is no merge");
1674        let all: Vec<Challenge> = lib
1675            .iter()
1676            .chain([&back])
1677            .map(|l| Challenge {
1678                id: l.id.clone(),
1679                origin_ms: l.origin_ms,
1680                question: l.question.clone(),
1681                engineers: l.engineers.clone(),
1682                supersedes: l.supersedes.clone(),
1683                ..Default::default()
1684            })
1685            .collect();
1686        assert_eq!(trust::successors(&all).get("L0").map(String::as_str), Some(back.id.as_str()));
1687    }
1688
1689    /// `fake_ask`, judging a correction that says "mine only" personal.
1690    fn fake_personal(p: &str) -> Option<Value> {
1691        let mut v = fake_ask(p)?;
1692        if p.split("## Engineer's correction\n").nth(1)?.lines().next()?.contains("mine only") {
1693            v["reach"] = json!(crate::reach::PERSONAL);
1694        }
1695        Some(v)
1696    }
1697
1698    #[test]
1699    fn a_personal_lesson_stays_in_its_collective_is_learned_once_and_delivered_only_there() {
1700        let base = temp_home("personal");
1701        let (work, oss) = (crate::collective::store(&base, "work"), crate::collective::store(&base, "oss"));
1702        std::fs::create_dir_all(&oss).unwrap();
1703        append_jsonl(&work.join("episodes.jsonl"), &at("e0", "s0", 1, "ann@x", "no co-author trailer, mine only")).unwrap();
1704        let calls = std::cell::Cell::new(0);
1705        let ask = |p: &str| {
1706            calls.set(calls.get() + 1);
1707            fake_personal(p)
1708        };
1709        assert_eq!(learn_in(&work, None, &ask, false, 3), (1, 0, 0));
1710        let personal = crate::reach::personal_lessons(&work);
1711        assert_eq!(personal.iter().map(|l| (l.id.as_str(), l.reach.as_str())).collect::<Vec<_>>(), [("Le0", crate::reach::PERSONAL)]);
1712        assert!(lessons(&work.join("lessons.jsonl")).is_empty(), "nothing in the team's lessons");
1713        assert_eq!(std::fs::read_dir(base.join("collectives")).unwrap().count(), 2, "no other store written");
1714        assert!(!oss.join("lessons.jsonl").exists() && !crate::reach::personal_path(&oss).exists());
1715        // Learned once: the second run finds its episode done and asks nothing.
1716        assert_eq!(learn_in(&work, None, &ask, false, 4), (0, 0, 0));
1717        assert_eq!(calls.get(), 1);
1718        // A later correction merging into it stays personal, whatever its own verdict: a merge never widens.
1719        append_jsonl(&work.join("episodes.jsonl"), &at("e1", "s1", 5, "ann@x", "the trailer again merge:Le0")).unwrap();
1720        assert_eq!(learn_in(&work, None, &ask, false, 6), (0, 1, 0));
1721        assert_eq!(crate::reach::personal_lessons(&work)[0].version, 2);
1722        assert!(lessons(&work.join("lessons.jsonl")).is_empty());
1723        assert_eq!(learned_episodes(&work), ["e0".to_string(), "e1".to_string()].into_iter().collect());
1724        // Delivered in this collective's sessions, as a team lesson for this engineer; in another
1725        // collective's never, nor in the team's own view of the store.
1726        let aliases = crate::collective::Aliases::default();
1727        let here = crate::conscience::load_store_with(&work.join("lessons.jsonl"), None, &aliases);
1728        assert!(here.iter().any(|c| c.id == "Le0" && crate::reach::deliverable(c, &["r".into()])), "{here:?}");
1729        assert!(crate::conscience::load_store_with(&oss.join("lessons.jsonl"), None, &aliases).is_empty());
1730        assert!(crate::conscience::load_with(&work.join("lessons.jsonl"), None, &aliases).is_empty());
1731        // Asked beside a team lesson: the team's synced logs name the team lesson only, never the
1732        // personal one (its id, when, which repo).
1733        for id in ["Le0", "Lteam"] {
1734            let o =
1735                outcome::Outcome { kind: outcome::ASKED.into(), session: "s9".into(), lessons: vec![id.into()], ms: 9, repo: "r".into(), ..Default::default() };
1736            outcome::append(&work.join("outcomes.jsonl"), &o).unwrap();
1737        }
1738        crate::stats::count(&work, crate::stats::LESSON_NEW, "s9", |ev| ev.lessons = vec!["Lteam".into()]);
1739        assert!(outcome::publish(&work) > 0 && crate::stats::rollup(&work) > 0);
1740        for f in [outcome::TEAM_FILE, "stats.jsonl"] {
1741            let text = std::fs::read_to_string(work.join(f)).unwrap();
1742            assert!(text.contains("Lteam") && !text.contains("Le0"), "{f}: {text}");
1743        }
1744        let _ = std::fs::remove_dir_all(base);
1745    }
1746
1747    #[test]
1748    fn a_personal_lesson_replaces_its_tombstone_where_it_is_delivered() {
1749        let base = temp_home("tombstone");
1750        let work = crate::collective::store(&base, "work");
1751        let l = Lesson { id: "L1".into(), question: "short commit subjects?".into(), version: 2, engineers: vec!["ann@x".into()], ..Lesson::default() };
1752        append_jsonl(&work.join("lessons.jsonl"), &crate::reach::tombstone(&l)).unwrap();
1753        append_jsonl(&crate::reach::personal_path(&work), &Lesson { reach: crate::reach::PERSONAL.into(), ..l }).unwrap();
1754        let cs = crate::conscience::load_store_with(&work.join("lessons.jsonl"), None, &crate::collective::Aliases::default());
1755        assert_eq!(cs.len(), 1);
1756        assert!(cs[0].question == "short commit subjects?" && crate::reach::deliverable(&cs[0], &[]));
1757        assert_eq!(store_lessons(&work).len(), 1);
1758        let _ = std::fs::remove_dir_all(base);
1759    }
1760
1761    #[test]
1762    fn a_personal_lesson_moved_elsewhere_is_neither_delivered_nor_merged_into_here() {
1763        let base = temp_home("moved-personal");
1764        let work = crate::collective::store(&base, "work");
1765        let l = Lesson {
1766            id: "Le0".into(),
1767            question: "short commit subjects?".into(),
1768            version: 1,
1769            episodes: vec!["e0".into()],
1770            reach: crate::reach::PERSONAL.into(),
1771            engineers: vec!["ann@x".into()],
1772            ..Lesson::default()
1773        };
1774        append_jsonl(&crate::reach::personal_path(&work), &l).unwrap();
1775        // What `borg move` leaves behind (moves.rs): the tombstone keeps the episodes.
1776        let t = Lesson { version: 2, question: crate::reach::MOVED.into(), consequence: String::new(), ..l.clone() };
1777        append_jsonl(&crate::reach::personal_path(&work), &t).unwrap();
1778        append_jsonl(&work.join("episodes.jsonl"), &at("e0", "s0", 1, "ann@x", "short subjects, mine only")).unwrap();
1779        let cs = crate::conscience::load_store_with(&work.join("lessons.jsonl"), None, &crate::collective::Aliases::default());
1780        assert!(cs.is_empty(), "the tombstone is never delivered: {cs:?}");
1781        assert!(store_lessons(&work).is_empty() && crate::reach::personal_lessons(&work).is_empty());
1782        assert!(learned_episodes(&work).contains("e0"), "its episodes stay learned");
1783        // A later correction that would merge into it makes a lesson of its own: the moved one stays gone.
1784        append_jsonl(&work.join("episodes.jsonl"), &at("e1", "s1", 2, "ann@x", "subjects moved to another collective again merge:Le0 mine only")).unwrap();
1785        assert_eq!(learn_in(&work, None, &fake_personal, false, 3), (1, 0, 0), "e0 is done; e1 cannot merge into Le0");
1786        let latest = crate::learn::lessons(&crate::reach::personal_path(&work));
1787        assert_eq!(latest.iter().find(|l| l.id == "Le0").map(|l| l.question.as_str()), Some(crate::reach::MOVED));
1788        let _ = std::fs::remove_dir_all(base);
1789    }
1790
1791    #[test]
1792    fn the_hooks_episode_waits_for_a_move_rewriting_the_log() {
1793        let home = temp_home("record-locked");
1794        let (held_tx, held_rx) = std::sync::mpsc::channel();
1795        let (go_tx, go_rx) = std::sync::mpsc::channel::<()>();
1796        let log = home.join("episodes.jsonl");
1797        let mover = {
1798            let log = log.clone();
1799            std::thread::spawn(move || {
1800                crate::live::locked(&log, || {
1801                    held_tx.send(()).unwrap();
1802                    go_rx.recv().unwrap();
1803                    // A move's rewrite: the log taken whole while the lock is held.
1804                    std::fs::write(&log, "").unwrap();
1805                })
1806            })
1807        };
1808        held_rx.recv().unwrap();
1809        let writer = {
1810            let home = home.clone();
1811            std::thread::spawn(move || record_episode_locked(&home, &at("e1", "s1", 1, "ann@x", "no, use the lock")).unwrap())
1812        };
1813        std::thread::sleep(std::time::Duration::from_millis(150));
1814        assert!(std::fs::read_to_string(&log).unwrap_or_default().is_empty(), "the append waits for the lock");
1815        go_tx.send(()).unwrap();
1816        mover.join().unwrap();
1817        writer.join().unwrap();
1818        assert!(std::fs::read_to_string(&log).unwrap().contains("\"e1\""), "and lands after the rewrite, not in it");
1819        let _ = std::fs::remove_dir_all(home);
1820    }
1821
1822    #[test]
1823    fn a_repo_reach_without_a_repo_is_the_teams() {
1824        let reach = |repo: &str, same_as: &str, lib: &[Lesson]| {
1825            let e = Episode { repo: repo.into(), ..ep("e1", "in the library keep the upstream grammar") };
1826            let answer = json!({"is_lesson": true, "question": "Q", "situations": [], "consequence": "c", "same_as": same_as, "area": "", "supersedes": [], "reach": "repo"});
1827            generalise(&e, lib, &|_| Some(answer.clone())).unwrap().unwrap().reach
1828        };
1829        assert_eq!(reach("", "", &[]), "", "a borg correct without --repo: heard by the team, not nowhere");
1830        assert_eq!(reach("lib", "", &[]), crate::reach::REPO);
1831        let old = Lesson { id: "L0".into(), question: "Does the library keep the upstream grammar?".into(), version: 1, ..Lesson::default() };
1832        assert_eq!(reach("lib", "L0", std::slice::from_ref(&old)), "", "merged into a lesson of no repo");
1833    }
1834
1835    #[test]
1836    fn an_episode_of_a_repo_that_routes_elsewhere_now_is_held_and_listed() {
1837        let base = temp_home("held");
1838        let repos = base.join("repos");
1839        for d in ["work/app", "oss/lib"] {
1840            std::fs::create_dir_all(repos.join(d)).unwrap();
1841        }
1842        let config = |oss: &str| -> crate::collective::Config {
1843            serde_json::from_value(json!({"collectives": [
1844                {"name": "work", "paths": [repos.join("work")]},
1845                {"name": "oss", "paths": [repos.join(oss)]}]}))
1846            .unwrap()
1847        };
1848        let cfg = config("oss/lib");
1849        let (work, oss) = (crate::collective::store(&base, "work"), crate::collective::store(&base, "oss"));
1850        let on = |id: &str, repo: &str, via: &str| Episode { repo: repo.into(), via: via.into(), ..at(id, "s0", 1, "ann@x", &format!("check {id}")) };
1851        for e in [on("e0", "app", ""), on("e1", "lib", ""), on("e2", "lib", CORRECT), on("e3", "elsewhere", ""), on("e4", "lib", "")] {
1852            append_jsonl(&work.join("episodes.jsonl"), &e).unwrap();
1853        }
1854        let calls = std::cell::Cell::new(0);
1855        let ask = |p: &str| {
1856            calls.set(calls.get() + 1);
1857            fake_ask(p)
1858        };
1859        // Learned here: its repo routes here, it was recorded here by name, or its route is unknown.
1860        assert_eq!(learn_in_config(&work, Some(&cfg), None, &ask, false, 3, &mut |_| {}), (3, 0, 0));
1861        let mut learned: Vec<String> = lessons(&work.join("lessons.jsonl")).into_iter().map(|l| l.id).collect();
1862        learned.sort();
1863        assert_eq!(learned, ["Le0", "Le2", "Le3"]);
1864        let route = crate::collective::Route::Collective { name: "oss".into(), home: oss.clone(), share: crate::collective::Share::Team };
1865        assert_eq!(held(&work, Some(&cfg)), [HeldRepo { repo: "lib".into(), route, episodes: vec!["e1".into(), "e4".into()] }]);
1866        // Held, not skipped: nothing marks them, and a second run asks nothing.
1867        assert!(!work.join("episodes-not-lessons.txt").exists());
1868        assert_eq!(learn_in_config(&work, Some(&cfg), None, &ask, false, 4, &mut |_| {}), (0, 0, 0));
1869        assert_eq!(calls.get(), 3);
1870        // No config, or a store that is no collective's: nothing is held.
1871        assert!(held(&work, None).is_empty() && held(&base, Some(&cfg)).is_empty());
1872        // Routed here again: learned here.
1873        let back = config("oss/none");
1874        assert!(held(&work, Some(&back)).is_empty());
1875        assert_eq!(learn_in_config(&work, Some(&back), None, &ask, false, 5, &mut |_| {}), (2, 0, 0));
1876        let _ = std::fs::remove_dir_all(base);
1877    }
1878
1879    fn temp_home(name: &str) -> PathBuf {
1880        let d = std::env::temp_dir().join(format!("borg-learn-{name}-{}", std::process::id()));
1881        let _ = std::fs::remove_dir_all(&d);
1882        std::fs::create_dir_all(&d).unwrap();
1883        d
1884    }
1885
1886    fn at(id: &str, session: &str, ms: i64, engineer: &str, correction: &str) -> Episode {
1887        Episode { ms, session: session.into(), engineer: engineer.into(), ..ep(id, correction) }
1888    }
1889
1890    fn asked(home: &Path, session: &str, ms: i64, ids: &[&str]) {
1891        let o = outcome::Outcome {
1892            kind: outcome::ASKED.into(),
1893            session: session.into(),
1894            lessons: ids.iter().map(|i| i.to_string()).collect(),
1895            ms,
1896            repo: "r".into(),
1897            ..Default::default()
1898        };
1899        outcome::append(&home.join("outcomes.jsonl"), &o).unwrap();
1900    }
1901
1902    fn regret_at(id: &str, decision_ms: i64) -> Episode {
1903        let r = regret::Regret { decision_session: "s-dec".into(), decision_ms, signals: vec!["subject".into()], ..regret::Regret::default() };
1904        Episode { kind: regret::KIND.into(), regret: Some(r), ..at(id, "s-fix", 100, "bob@x", "Fixed later (3d, subject): fix: take the lock") }
1905    }
1906
1907    /// The model's verdict on a regret: a lesson, or (`lesson` false) not one.
1908    fn regret_ask(lesson: bool) -> impl Fn(&str) -> Option<Value> {
1909        move |_| {
1910            Some(json!({"is_lesson": lesson, "question": "Before reading a shared flag, has a writer been considered?", "situations": [],
1911            "consequence": "a race", "same_as": "", "supersedes": []}))
1912        }
1913    }
1914
1915    #[test]
1916    fn a_settled_lesson_says_why_the_judge_gave_it_no_verdict() {
1917        let home = temp_home("unjudged");
1918        asked(&home, "s1", 10, &["A", "B", "C", "D"]);
1919        let ignored =
1920            outcome::Outcome { kind: outcome::IGNORED.into(), session: "s1".into(), lessons: vec!["A".into()], ms: 12, repo: "r".into(), ..Default::default() };
1921        outcome::append(&home.join("outcomes.jsonl"), &ignored).unwrap();
1922        std::fs::create_dir_all(home.join("state")).unwrap();
1923        let ctx = "#borg-asked A,B,C,D @0123456789ab\n#borg-judged B=pending\n#borg-judged B=pending\n#borg-unjudged C why=slots\n";
1924        std::fs::write(home.join("state/s1.ctx"), ctx).unwrap();
1925        learn_in(&home, Some("s1"), &fake_ask, true, 100);
1926        let why: Vec<(String, String)> = outcome::read(&home.join("outcomes.jsonl"))
1927            .into_iter()
1928            .filter(|o| o.kind == outcome::NOT_CORRECTED)
1929            .map(|o| (o.lessons.join(","), o.reason))
1930            .collect();
1931        let want = [("A", outcome::UNJUDGED_NONE), ("B", outcome::UNJUDGED_TRIES), ("C", outcome::UNJUDGED_SLOTS), ("D", outcome::UNJUDGED_NO_CALL)];
1932        assert_eq!(why, want.map(|(l, r)| (l.to_string(), r.to_string())));
1933        let _ = std::fs::remove_dir_all(home);
1934    }
1935
1936    /// Settle waits for a session's end: idle by the log (no ask or episode for a day) is not enough while
1937    /// its state file is still being written.
1938    #[test]
1939    fn an_idle_session_still_working_is_not_settled() {
1940        let home = temp_home("open");
1941        asked(&home, "s1", 10, &["A"]);
1942        std::fs::create_dir_all(home.join("state")).unwrap();
1943        std::fs::write(home.join("state/s1.ctx"), "a prompt just now\n").unwrap();
1944        let now = now_ms();
1945        learn_in(&home, None, &fake_ask, true, now);
1946        assert!(results(&home).is_empty(), "still open: {:?}", results(&home));
1947        // A day after its last line, it ended.
1948        learn_in(&home, None, &fake_ask, true, now + outcome::IDLE_MS + 60_000);
1949        assert_eq!(results(&home), [(outcome::NOT_CORRECTED.into(), "s1".into(), "A".into())]);
1950        let _ = std::fs::remove_dir_all(home);
1951    }
1952
1953    /// A regret is a correction of the deciding session's work too: the lesson it merges into, graded
1954    /// applying in that session before the decision and never asked there, was a question needed and not
1955    /// asked.
1956    #[test]
1957    fn a_regret_merged_into_a_lesson_graded_but_not_asked_before_the_decision_is_a_miss() {
1958        let home = temp_home("regret-miss");
1959        append_jsonl(&home.join("episodes.jsonl"), &at("e0", "s0", 1, "ann@x", "use the lock on the flag")).unwrap();
1960        learn_in(&home, Some("s0"), &fake_ask, true, 2);
1961        let graded = |ms: i64| outcome::Outcome {
1962            kind: outcome::GRADED.into(),
1963            session: "s-dec".into(),
1964            lessons: vec!["Le0".into()],
1965            ms,
1966            repo: "r".into(),
1967            grades: vec![2],
1968            ..Default::default()
1969        };
1970        for g in [graded(10), graded(30)] {
1971            outcome::append(&home.join("outcomes.jsonl"), &g).unwrap();
1972        }
1973        append_jsonl(&home.join("episodes.jsonl"), &regret_at("R1", 20)).unwrap();
1974        let merge = |_: &str| Some(json!({"is_lesson": true, "question": "Q", "situations": [], "consequence": "c", "same_as": "Le0", "supersedes": []}));
1975        learn_in(&home, None, &merge, true, 200);
1976        let missed: Vec<(String, String, i64, String)> = outcome::read(&home.join("outcomes.jsonl"))
1977            .into_iter()
1978            .filter(|o| o.kind == outcome::MISSED)
1979            .map(|o| (o.lessons.join(","), o.session, o.ms, o.cause))
1980            .collect();
1981        assert_eq!(missed, [("Le0".to_string(), "s-dec".to_string(), 20, "R1".to_string())]);
1982        learn_in(&home, None, &merge, true, 300);
1983        assert_eq!(outcome::read(&home.join("outcomes.jsonl")).iter().filter(|o| o.kind == outcome::MISSED).count(), 1, "once");
1984        let _ = std::fs::remove_dir_all(home);
1985    }
1986
1987    #[test]
1988    fn a_regret_blames_lessons_asked_before_the_decision_once() {
1989        let regretted = |home: &Path| -> Vec<(String, String, f64, String)> {
1990            outcome::read(&home.join("outcomes.jsonl"))
1991                .into_iter()
1992                .filter(|o| o.kind == outcome::REGRETTED)
1993                .map(|o| (o.lessons.join(","), o.session, o.credit, o.cause))
1994                .collect()
1995        };
1996        let setup = |name: &str, e: Episode| {
1997            let home = temp_home(name);
1998            asked(&home, "s-dec", 10, &["Le0", "Le1"]);
1999            asked(&home, "s-dec", 50, &["Le2"]);
2000            append_jsonl(&home.join("episodes.jsonl"), &e).unwrap();
2001            home
2002        };
2003        let home = setup("regret", regret_at("R1", 20));
2004        assert_eq!(learn_in(&home, None, &regret_ask(true), true, 200), (1, 0, 0));
2005        let got = regretted(&home);
2006        assert_eq!(
2007            got,
2008            [("Le0".into(), "s-dec".into(), 0.25, "R1".into()), ("Le1".into(), "s-dec".into(), 0.25, "R1".into())],
2009            "Le2 was asked after the decision"
2010        );
2011        let before = std::fs::read_to_string(home.join("outcomes.jsonl")).unwrap();
2012        learn_in(&home, None, &regret_ask(true), true, 300);
2013        assert_eq!(std::fs::read_to_string(home.join("outcomes.jsonl")).unwrap(), before, "credited once");
2014        let _ = std::fs::remove_dir_all(home);
2015        // Not a lesson, no decision time, not tracking: nothing is blamed.
2016        let home = setup("regret-no", regret_at("R1", 20));
2017        learn_in(&home, None, &regret_ask(false), true, 200);
2018        assert!(regretted(&home).is_empty(), "not a lesson");
2019        let _ = std::fs::remove_dir_all(home);
2020        let home = setup("regret-old", regret_at("R1", 0));
2021        learn_in(&home, None, &regret_ask(true), true, 200);
2022        assert!(regretted(&home).is_empty(), "a regret from before decision times were kept");
2023        let _ = std::fs::remove_dir_all(home);
2024        let home = setup("regret-untracked", regret_at("R1", 20));
2025        learn_in(&home, None, &regret_ask(true), false, 200);
2026        assert!(regretted(&home).is_empty(), "BORG_LEARN=off writes nothing");
2027        let _ = std::fs::remove_dir_all(home);
2028    }
2029
2030    #[test]
2031    fn records_outcomes_of_asked_lessons() {
2032        let home = temp_home("outcomes");
2033        // Session s0 teaches lessons Le0 (TOCTOU) and Le1 (spec).
2034        for e in [at("e0", "s0", 1, "ann@x", "check TOCTOU on the flag"), at("e1", "s0", 2, "", "check against the spec")] {
2035            append_jsonl(&home.join("episodes.jsonl"), &e).unwrap();
2036        }
2037        assert_eq!(learn_in(&home, Some("s0"), &fake_ask, true, 3), (2, 0, 0));
2038        assert!(outcome::read(&home.join("outcomes.jsonl")).is_empty(), "nothing asked, nothing settled");
2039        // s1: both asked at 10; the TOCTOU concern is corrected again at 20 (merged into Le0).
2040        // Le1 asked again after that correction, which must not count as corrected.
2041        asked(&home, "s1", 10, &["Le0", "Le1"]);
2042        append_jsonl(&home.join("episodes.jsonl"), &at("e2", "s1", 20, "bob@x", "flag still racy merge:Le0")).unwrap();
2043        asked(&home, "s1", 30, &["Le1"]);
2044        assert_eq!(learn_in(&home, Some("s1"), &fake_ask, true, 40), (0, 1, 0));
2045        let result: Vec<(String, String, i64)> =
2046            outcome::read(&home.join("outcomes.jsonl")).into_iter().filter(|o| o.kind != outcome::ASKED).map(|o| (o.kind, o.lessons.join(","), o.ms)).collect();
2047        assert_eq!(result, [(outcome::CORRECTED.into(), "Le0".into(), 20), (outcome::NOT_CORRECTED.into(), "Le1".into(), 40)]);
2048        // Re-running learn for the same session appends nothing: deterministic and idempotent.
2049        let before = std::fs::read_to_string(home.join("outcomes.jsonl")).unwrap();
2050        learn_in(&home, Some("s1"), &fake_ask, true, 50);
2051        assert_eq!(std::fs::read_to_string(home.join("outcomes.jsonl")).unwrap(), before);
2052        // Engineers accumulate on the merged lesson.
2053        let l0 = lessons(&home.join("lessons.jsonl")).into_iter().find(|l| l.id == "Le0").unwrap();
2054        assert_eq!(l0.engineers, ["ann@x", "bob@x"]);
2055        // The same email in another case is the same engineer: not added, no join recorded.
2056        append_jsonl(&home.join("episodes.jsonl"), &at("e3", "s2", 60, "Ann@X", "flag racy again merge:Le0")).unwrap();
2057        assert_eq!(learn_in(&home, Some("s2"), &fake_ask, true, 70), (0, 1, 0));
2058        let l0 = lessons(&home.join("lessons.jsonl")).into_iter().find(|l| l.id == "Le0").unwrap();
2059        assert_eq!((l0.engineers.as_slice(), l0.joined.get("Ann@X")), (&["ann@x".to_string(), "bob@x".to_string()][..], None));
2060        let _ = std::fs::remove_dir_all(home);
2061    }
2062
2063    #[test]
2064    fn idle_sessions_settle_without_their_end_hook() {
2065        let home = temp_home("idle");
2066        for e in [at("e0", "s0", 1, "", "check TOCTOU on the flag"), at("e1", "s0", 2, "", "check against the spec")] {
2067            append_jsonl(&home.join("episodes.jsonl"), &e).unwrap();
2068        }
2069        learn_in(&home, Some("s0"), &fake_ask, true, 3);
2070        // s1 and s3 never send `end`. s1 asks both and is corrected on Le0; s3 asks Le0, no correction.
2071        asked(&home, "s1", 10, &["Le0", "Le1"]);
2072        append_jsonl(&home.join("episodes.jsonl"), &at("e2", "s1", 20, "", "flag still racy merge:Le0")).unwrap();
2073        asked(&home, "s3", 5, &["Le0"]);
2074        let idle = 20 + outcome::IDLE_MS;
2075        let results = || -> Vec<(String, String, String)> {
2076            outcome::read(&home.join("outcomes.jsonl"))
2077                .into_iter()
2078                .filter(|o| o.kind != outcome::ASKED)
2079                .map(|o| (o.kind, o.session, o.lessons.join(",")))
2080                .collect()
2081        };
2082        let not = |s: &str, l: &str| (outcome::NOT_CORRECTED.to_string(), s.to_string(), l.to_string());
2083        // s2 ends a day later: its own ask settles, and idle s3; s1 not: it is still open (its state file
2084        // is fresh), so its episode is still pending.
2085        let ctx = home.join("state/s1.ctx");
2086        std::fs::create_dir_all(ctx.parent().unwrap()).unwrap();
2087        std::fs::write(&ctx, "").unwrap();
2088        asked(&home, "s2", idle - 1, &["Le1"]);
2089        learn_in(&home, Some("s2"), &fake_ask, true, idle);
2090        assert_eq!(results(), [not("s3", "Le0"), not("s2", "Le1")]);
2091        // s1 goes quiet; a plain `borg learn` folds e2 in and settles s1 too, so s1 gives both kinds of outcome.
2092        std::fs::remove_file(&ctx).unwrap();
2093        assert_eq!(learn_in(&home, None, &fake_ask, true, idle), (0, 1, 0));
2094        let corrected = (outcome::CORRECTED.to_string(), "s1".to_string(), "Le0".to_string());
2095        assert_eq!(results(), [not("s3", "Le0"), not("s2", "Le1"), corrected.clone(), not("s1", "Le1")]);
2096        // A fresh session is not settled, and re-running appends nothing.
2097        asked(&home, "s4", idle, &["Le1"]);
2098        learn_in(&home, None, &fake_ask, true, idle + 1);
2099        assert_eq!(results().len(), 4);
2100        let _ = std::fs::remove_dir_all(home);
2101    }
2102
2103    /// A session whose live lock another process holds (a session-end pass still judging it) is not
2104    /// settled under it; once the lock is gone, it is.
2105    #[test]
2106    fn a_session_another_pass_still_judges_is_not_settled_under_it() {
2107        let home = temp_home("held");
2108        asked(&home, "s2", 10, &["Le1"]);
2109        let ctx = home.join("state/s2.ctx");
2110        std::fs::create_dir_all(ctx.parent().unwrap()).unwrap();
2111        std::fs::write(crate::live::lock_file(&ctx), format!("{} {}\n", std::process::id(), now_ms())).unwrap();
2112        let settled = || outcome::read(&home.join("outcomes.jsonl")).into_iter().filter(|o| o.kind == outcome::NOT_CORRECTED).count();
2113        learn_in(&home, Some("s2"), &fake_ask, true, 20);
2114        assert_eq!(settled(), 0, "held: not settled");
2115        std::fs::remove_file(crate::live::lock_file(&ctx)).unwrap();
2116        learn_in(&home, Some("s2"), &fake_ask, true, 30);
2117        assert_eq!(settled(), 1, "let go: settled");
2118        let _ = std::fs::remove_dir_all(home);
2119    }
2120
2121    #[test]
2122    fn a_quiet_sessions_episodes_are_learned_by_another_sessions_end() {
2123        let home = temp_home("quiet");
2124        // s1 never ends; the transcript sweep added its last correction after it went quiet (sweep.rs `idle`).
2125        append_jsonl(&home.join("episodes.jsonl"), &at("e0", "s1", 10, "", "check TOCTOU on the flag")).unwrap();
2126        // s9 is busy (an episode an hour ago) and open; s2 ends now.
2127        append_jsonl(&home.join("episodes.jsonl"), &at("e1", "s9", 10 + outcome::IDLE_MS, "", "check against the spec")).unwrap();
2128        append_jsonl(&home.join("episodes.jsonl"), &at("e2", "s2", 20 + outcome::IDLE_MS, "", "no, keep the cache")).unwrap();
2129        let now = 30 + outcome::IDLE_MS;
2130        // While s1's state file is fresh it is open, and waits for its own end.
2131        let ctx = home.join("state/s1.ctx");
2132        std::fs::create_dir_all(ctx.parent().unwrap()).unwrap();
2133        std::fs::write(&ctx, "").unwrap();
2134        assert_eq!(learn_in(&home, Some("s2"), &fake_ask, true, now), (1, 0, 0), "s2's own episode only");
2135        // Quiet: s2's end learns it too; busy s9 still waits.
2136        std::fs::remove_file(&ctx).unwrap();
2137        append_jsonl(&home.join("episodes.jsonl"), &at("e3", "s3", 25 + outcome::IDLE_MS, "", "no, the other crate")).unwrap();
2138        assert_eq!(learn_in(&home, Some("s3"), &fake_ask, true, now), (2, 0, 0), "s3's own and quiet s1's");
2139        let learned: HashSet<String> = lessons(&home.join("lessons.jsonl")).into_iter().flat_map(|l| l.episodes).collect();
2140        assert!(learned.contains("e0") && !learned.contains("e1"), "{learned:?}");
2141        let _ = std::fs::remove_dir_all(home);
2142    }
2143
2144    #[test]
2145    fn no_outcomes_written_without_tracking() {
2146        let home = temp_home("off");
2147        append_jsonl(&home.join("episodes.jsonl"), &at("e0", "s0", 1, "", "check TOCTOU")).unwrap();
2148        learn_in(&home, Some("s0"), &fake_ask, true, 2);
2149        asked(&home, "s1", 10, &["Le0"]);
2150        let before = std::fs::read_to_string(home.join("outcomes.jsonl")).unwrap();
2151        append_jsonl(&home.join("episodes.jsonl"), &at("e1", "s1", 20, "", "TOCTOU again merge:Le0")).unwrap();
2152        assert_eq!(learn_in(&home, Some("s1"), &fake_ask, false, 30), (0, 1, 0));
2153        assert_eq!(std::fs::read_to_string(home.join("outcomes.jsonl")).unwrap(), before);
2154        let _ = std::fs::remove_dir_all(home);
2155    }
2156
2157    #[test]
2158    fn failed_model_calls_leave_episodes_waiting_and_stop_early() {
2159        let home = temp_home("failed");
2160        for i in 0..5 {
2161            append_jsonl(&home.join("episodes.jsonl"), &at(&format!("e{i}"), "s0", i, "", &format!("check case {i}"))).unwrap();
2162        }
2163        asked(&home, "s0", 0, &["Lx"]);
2164        let calls = std::cell::Cell::new(0);
2165        let down = |_: &str| -> Option<Value> {
2166            calls.set(calls.get() + 1);
2167            None
2168        };
2169        // Quota hit: nothing is marked "not a lesson", the run gives up after MAX_FAILURES calls,
2170        // and the session is not settled while its episodes wait.
2171        assert_eq!(learn_in(&home, Some("s0"), &down, true, 1 + outcome::IDLE_MS), (0, 0, 5));
2172        assert_eq!(calls.get(), MAX_FAILURES);
2173        assert!(!home.join("episodes-not-lessons.txt").exists());
2174        assert!(outcome::read(&home.join("outcomes.jsonl")).iter().all(|o| o.kind == outcome::ASKED));
2175        // The model is back: the next run retries every episode.
2176        assert_eq!(learn_in(&home, Some("s0"), &fake_ask, true, 2 + outcome::IDLE_MS), (5, 0, 0));
2177        let _ = std::fs::remove_dir_all(home);
2178    }
2179
2180    #[test]
2181    fn area_and_supersedes_come_from_the_same_call() {
2182        let lib = vec![
2183            generalise(&ep("e1", "use a lock here"), &[], &|_| {
2184                Some(json!({"is_lesson": true, "question": "Is the flag guarded by a lock?", "situations": ["flag across threads"],
2185                    "consequence": "race", "same_as": "", "area": "Concurrency", "supersedes": ["Lnope"]}))
2186            })
2187            .unwrap()
2188            .unwrap(),
2189        ];
2190        assert_eq!(lib[0].area, "concurrency", "case-insensitive; normalised");
2191        assert!(lib[0].supersedes.is_empty(), "an id the model was not shown is dropped");
2192        // A lesson without an area (older store, or a model that ignores it) is "other".
2193        let old: Lesson = serde_json::from_str(
2194            r#"{"id":"L","origin_ms":1,"repo":"r","question":"q","situations":[],"consequence":"c",
2195            "support":1,"episodes":[],"history":[],"version":1}"#,
2196        )
2197        .unwrap();
2198        assert_eq!((old.area.as_str(), old.supersedes.len()), ("other", 0));
2199        let id = lib[0].id.clone();
2200        let reversed = generalise(&ep("e2", "no, the lock is wrong here, use the atomic flag"), &lib, &move |_| {
2201            Some(json!({"is_lesson": true, "question": "Is a lock-free atomic enough for the flag?", "situations": ["flag across threads"],
2202                "consequence": "lock contention", "same_as": "", "area": "made-up", "supersedes": [id.clone()]}))
2203        })
2204        .unwrap()
2205        .unwrap();
2206        assert_eq!((reversed.area.as_str(), reversed.supersedes.clone()), ("other", vec![lib[0].id.clone()]));
2207        assert!(!serde_json::to_string(&lib[0]).unwrap().contains("supersedes"), "empty supersedes stays off the line");
2208        // The merge target is never superseded; a merge keeps what the old version superseded.
2209        let (rid, lid) = (reversed.id.clone(), lib[0].id.clone());
2210        let merged = generalise(&ep("e3", "again: atomic, not a lock"), &[lib[0].clone(), reversed.clone()], &move |_| {
2211            Some(json!({"is_lesson": true, "question": "Is a lock-free atomic enough?", "situations": ["flag across threads"],
2212                "consequence": "lock contention", "same_as": rid.clone(), "area": "concurrency", "supersedes": [rid.clone(), lid.clone()]}))
2213        })
2214        .unwrap()
2215        .unwrap();
2216        assert_eq!((merged.id.as_str(), merged.version, merged.supersedes.clone()), (reversed.id.as_str(), 2, vec![lib[0].id.clone()]));
2217    }
2218
2219    #[test]
2220    fn new_lesson_then_merge_keeps_history() {
2221        let first = generalise(&ep("e1", "carefully check TOCTOU on the atomic flags"), &[], &|_| {
2222            Some(json!({"is_lesson": true, "question": "Is every check-then-act on a shared flag TOCTOU-safe?",
2223                "situations": ["check-then-act on a shared mutable flag across threads"], "consequence": "race", "same_as": ""}))
2224        })
2225        .unwrap()
2226        .unwrap();
2227        assert_eq!((first.version, first.support), (1, 1));
2228        let id = first.id.clone();
2229        let merged = generalise(&ep("e2", "again: the closed flag check is racy"), std::slice::from_ref(&first), &move |p| {
2230            assert!(p.contains(&id), "the similar lesson must be offered for merging");
2231            Some(json!({"is_lesson": true, "question": "Is each check-then-act on shared flags atomic or harmless on race?",
2232                "situations": ["check-then-act on a shared mutable flag across threads"], "consequence": "race", "same_as": id}))
2233        })
2234        .unwrap()
2235        .unwrap();
2236        assert_eq!((merged.id.as_str(), merged.version, merged.support, merged.history.len()), (first.id.as_str(), 2, 2, 1));
2237        let skipped =
2238            generalise(&ep("e3", "stop"), &[], &|_| Some(json!({"is_lesson": false, "question": "", "situations": [], "consequence": "", "same_as": ""})));
2239        assert!(matches!(skipped, Ok(None)), "a verdict: not a lesson");
2240        assert_eq!(generalise(&ep("e4", "no, the lock"), &[], &|_| None).map(|_| ()), Err(AskFailed), "no answer is no verdict");
2241        let malformed = generalise(&ep("e5", "no, the lock"), &[], &|_| Some(json!({"error": "rate limited"})));
2242        assert_eq!(malformed.map(|_| ()), Err(AskFailed));
2243    }
2244}