Skip to main content

borg/
hook.rs

1//! The hook: every harness's events, one canonical, Claude-shaped event each (session_id, cwd, prompt,
2//! tool_name, tool_input, error, last_assistant_message, transcript_path, stop_hook_active), answered
3//! Claude-shaped ({"decision":"block","reason"} at stop, `hookSpecificOutput` with context at a prompt
4//! or tool call); adapters translate both ways (harness/). What differs per harness comes from its
5//! [`Harness`]: capabilities, the turn's work and the drone's last message, never its name.
6//!
7//! `main` routes and records presence; [`dispatch`] has one function per event. Hooks always fail open:
8//! any error lets the drone continue untouched.
9
10use std::io::{Read, Write};
11use std::path::{Path, PathBuf};
12
13use borg_trace::fingerprint::bash_read_paths;
14use serde_json::{Value, json};
15
16use crate::harness::{self, Harness};
17use crate::learn::learning;
18use crate::{collective, conscience, learn, live, outcome, predict, prefilter};
19
20pub(crate) fn state_file(session: &str) -> PathBuf {
21    let dir = std::env::var_os("BORG_STATE").map_or_else(|| PathBuf::from("/tmp/borg"), PathBuf::from);
22    let _ = std::fs::create_dir_all(&dir);
23    state_file_in(&dir, session)
24}
25
26/// `session`'s state file in the state directory `dir` (`state_file`, where the hooks' is).
27pub(crate) fn state_file_in(dir: &Path, session: &str) -> PathBuf {
28    let safe: String = session.chars().filter(|c| c.is_ascii_alphanumeric() || *c == '-').collect();
29    dir.join(format!("{safe}.ctx"))
30}
31
32/// One line (or several, kept together) at the end of the state file, in one write: a check running in
33/// the background appends to the same file, and its lines never land between these.
34pub(crate) fn append(session: &str, text: &str) {
35    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(state_file(session)) {
36        let _ = f.write_all(format!("{text}\n").as_bytes());
37    }
38}
39
40/// What a tool call touched, as text for matching: paths and search patterns.
41pub(crate) fn touched(input: &Value, tool: &str) -> String {
42    let s = |k: &str| input.get(k).and_then(Value::as_str).unwrap_or_default().to_string();
43    match tool {
44        "Bash" => {
45            let cmd = s("command");
46            let paths = bash_read_paths(&cmd);
47            if paths.is_empty() { cmd.chars().take(COMMAND_CHARS).collect() } else { paths.join(" ") }
48        }
49        "Grep" => format!("{} {}", s("pattern"), s("path")),
50        "Glob" => s("pattern"),
51        _ => {
52            let mut t = [s("file_path"), s("notebook_path"), s("path")].join(" ").trim().to_string();
53            // An edit of several files (a Codex patch) names the further ones in its `edits`.
54            for p in further_files(input) {
55                if !t.is_empty() {
56                    t.push(' ');
57                }
58                t.push_str(p);
59            }
60            t
61        }
62    }
63}
64
65/// A shell command's text kept as its touch when it names no path (`touched`). Guard: bounds one
66/// state-file line (a heredoc would otherwise go in whole). borg-trace's `TOUCH_CHARS`, which a replay
67/// of this hook mirrors.
68const COMMAND_CHARS: usize = borg_trace::dialogue::TOUCH_CHARS;
69
70/// The files an edit names in its `edits` beside its own `file_path` (a Codex patch of several files;
71/// Claude's MultiEdit names none there). They reach `.ctx` and `.edits`; presence (`borg top`) shows
72/// one file per drone, the edit's own `file_path`.
73pub(crate) fn further_files(input: &Value) -> impl Iterator<Item = &str> {
74    input.get("edits").and_then(Value::as_array).into_iter().flatten().filter_map(|e| e.get("file_path").and_then(Value::as_str)).filter(|p| !p.is_empty())
75}
76
77pub(crate) fn repo_of(cwd: &str) -> String {
78    borg_trace::repo_key(cwd).rsplit('/').next().unwrap_or_default().to_string()
79}
80
81/// Counts one event of this session in the collective's statistics (stats.rs); `fill` adds details.
82fn stat(session: &str, input: &Value, kind: &str, fill: impl FnOnce(&mut crate::stats::Event)) {
83    let home = learn::home();
84    let mut ev = crate::stats::Event::new(kind, learn::now_ms());
85    ev.session = session.into();
86    ev.repo = repo_of(input.get("cwd").and_then(Value::as_str).unwrap_or_default());
87    // The engineer as presence.rs already resolved it at this hook: no second `git config` call.
88    ev.engineer = crate::presence::read(&home, session).map(|p| p.engineer).unwrap_or_default();
89    fill(&mut ev);
90    crate::stats::record(&home, &ev);
91}
92
93/// The engineer's prompts this session (the queen's criterion 3; the state file mixes in touches).
94fn prompts_file(session: &str) -> PathBuf {
95    state_file(session).with_extension("prompts")
96}
97
98/// Where the session works: the repo it runs in and the paths it touched (`where_now` adds the latest
99/// prompt). Scopes are judged here only, so a system merely mentioned in a command's text or in an
100/// earlier part of the conversation is not a topic.
101fn where_file(session: &str) -> PathBuf {
102    state_file(session).with_extension("where")
103}
104
105/// Where the session works now: the repo and touched paths so far, plus the engineer's latest prompt
106/// (an earlier prompt's system is not this turn's work).
107fn where_now(session: &str) -> String {
108    let paths = std::fs::read_to_string(where_file(session)).unwrap_or_default();
109    let latest = std::fs::read_to_string(state_file(session).with_extension("lastprompt")).unwrap_or_default();
110    format!("{paths}\n{latest}")
111}
112
113fn note_where(session: &str, text: &str) {
114    if text.trim().is_empty() {
115        return;
116    }
117    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(where_file(session)) {
118        let _ = writeln!(f, "{text}");
119    }
120}
121
122fn task_file(session: &str) -> PathBuf {
123    state_file(session).with_extension("task")
124}
125
126/// A tool call refused without words, waiting for the engineer's next prompt (their reason).
127/// Pending only while the drone waits: a later successful tool call or the session's end clears it,
128/// because omp lets the model carry on after a denial and the next prompt may be about anything.
129fn denied_file(session: &str) -> PathBuf {
130    state_file(session).with_extension("denied")
131}
132
133/// A correcting prompt becomes an episode (task, the drone's last proposal, the correction);
134/// so does the first prompt after a wordless tool denial. Any other prompt becomes the session's task.
135fn capture_prompt(h: &dyn Harness, session: &str, input: &Value, prompt: &str) {
136    if !learning() {
137        return;
138    }
139    let task = std::fs::read_to_string(task_file(session)).unwrap_or_default();
140    if let Ok(tool) = std::fs::read_to_string(denied_file(session)) {
141        let _ = std::fs::remove_file(denied_file(session));
142        let said = prompt.trim();
143        if said.len() > 3 && !said.starts_with('<') {
144            record(session, input, task, proposal(h, input), Some(tool), said.chars().take(learn::EXCERPT_CHARS).collect());
145            return;
146        }
147    }
148    if !learn::is_correction(prompt) || task.is_empty() {
149        let task: String = prompt.chars().take(learn::TASK_CHARS).collect();
150        let _ = std::fs::write(task_file(session), task);
151        return;
152    }
153    record(session, input, task, proposal(h, input), None, prompt.chars().take(learn::EXCERPT_CHARS).collect());
154}
155
156/// A veto said in the session (veto.rs): its episode, the shown lessons' credit and its stat. A pending
157/// wordless denial is answered by it, so it never becomes a correction later.
158fn veto(h: &dyn Harness, session: &str, input: &Value, reason: &str) {
159    let _ = std::fs::remove_file(denied_file(session));
160    let cwd = input.get("cwd").and_then(Value::as_str).unwrap_or_default();
161    let task = std::fs::read_to_string(task_file(session)).unwrap_or_default();
162    let said = crate::veto::Said {
163        session,
164        repo: &repo_of(cwd),
165        engineer: &learn::engineer(cwd),
166        task: &task,
167        agent: &proposal(h, input),
168        reason,
169        ms: learn::now_ms(),
170        via: "prompt",
171    };
172    if let Ok(v) = crate::veto::record(&learn::home(), &said) {
173        append(session, &format!("#borg-episode {}", v.episode));
174    }
175}
176
177/// A rejected tool call or plan that carries the engineer's words becomes an episode; a wordless
178/// denial waits for the next prompt (`capture_prompt`).
179fn capture_rejection(h: &dyn Harness, session: &str, input: &Value) {
180    if !learning() {
181        return;
182    }
183    let error = input.get("error").and_then(Value::as_str).unwrap_or_default();
184    let tool = format!(
185        "{} {}",
186        input.get("tool_name").and_then(Value::as_str).unwrap_or_default(),
187        input.get("tool_input").map(|v| v.to_string()).unwrap_or_default().chars().take(learn::EXCERPT_CHARS).collect::<String>()
188    );
189    let Some(said) = learn::rejection(error) else {
190        if learn::denied(error) {
191            let _ = std::fs::write(denied_file(session), tool);
192        }
193        return;
194    };
195    let task = std::fs::read_to_string(task_file(session)).unwrap_or_default();
196    record(session, input, task, proposal(h, input), Some(tool), said);
197}
198
199/// The drone's last proposal, its last `learn::PROPOSAL_CHARS` characters (`harness::last_message`: the
200/// adapter's, else the transcript's).
201fn proposal(h: &dyn Harness, input: &Value) -> String {
202    let t = full_proposal(h, input);
203    t.chars().rev().take(learn::PROPOSAL_CHARS).collect::<Vec<_>>().into_iter().rev().collect()
204}
205
206/// The drone's whole last proposal (`proposal` keeps its end): the reasoning note
207/// redacts it before it cuts (notes.rs).
208fn full_proposal(h: &dyn Harness, input: &Value) -> String {
209    harness::last_message(h.caps(), input)
210}
211
212fn record(session: &str, input: &Value, task: String, agent: String, rejected_tool: Option<String>, correction: String) {
213    let ms = learn::now_ms();
214    let cwd = input.get("cwd").and_then(Value::as_str).unwrap_or_default();
215    let e = learn::Episode {
216        id: format!("{}-{ms}", session.chars().take(8).collect::<String>()),
217        ms,
218        repo: repo_of(cwd),
219        session: session.into(),
220        task,
221        agent,
222        rejected_tool,
223        correction,
224        engineer: learn::engineer(cwd),
225        ..learn::Episode::default()
226    };
227    let _ = learn::record_episode(&e);
228    append(session, &format!("#borg-episode {}", e.id));
229}
230
231/// The lessons the conscience reads (BORG_CONSCIENCE, the BORG_BEFORE cut-off) as `session` sees them:
232/// weighted by their memory with this session's usefulness sample (memory.rs), without the ones it may
233/// not deliver here (systems, reach, absorbed). Every hook path that hands lessons to a drone reads them
234/// here.
235fn lessons(session: &str) -> Vec<conscience::Challenge> {
236    loaded(session).0
237}
238
239/// Every lesson the judge gives a verdict on: `all` and the absorbed ones withheld from it (`loaded`).
240/// A delivery made before its lesson was absorbed is evidence about the lesson, not a delivery, so the
241/// judge still sees it (`Work::judged`).
242fn judged(all: &[conscience::Challenge], withheld: Vec<conscience::Challenge>) -> std::borrow::Cow<'_, [conscience::Challenge]> {
243    if withheld.is_empty() { all.into() } else { all.iter().cloned().chain(withheld).collect::<Vec<_>>().into() }
244}
245
246/// `lessons`, and the lessons the absorbed filter withheld from it (absorb.rs `withhold`).
247fn loaded(session: &str) -> (Vec<conscience::Challenge>, Vec<conscience::Challenge>) {
248    let before = std::env::var("BORG_BEFORE").ok().and_then(|v| v.parse().ok());
249    // The routed collective's lessons and this engineer's personal ones of it (reach.rs), nothing else.
250    let mut all = conscience::load_store_for(&lessons_path(), before, Some(session));
251    // The system gate (systems.rs): where the repo says which system this is, lessons scoped to other
252    // systems score 0, however much the conversation names them.
253    let where_text = std::fs::read_to_string(where_file(session)).unwrap_or_default();
254    let repos = crate::systems::work_repos(&where_text);
255    if crate::systems::on() {
256        crate::systems::gate(&mut all, &repos);
257    }
258    // Reach (reach.rs): a tombstone is never delivered, a repo convention only in its repo.
259    all.retain(|c| crate::reach::deliverable(c, &repos));
260    // Absorbed (absorb.rs): held outside Borg now, so delivered on no path; every path reads this list.
261    let withheld = crate::absorb::withhold_beside(&lessons_path(), &mut all, before);
262    // Spike (concepts.rs): lessons sharing engineering concepts with the code written, the touched paths
263    // and the latest prompt rank higher; never a filter.
264    if crate::concepts::on() {
265        let where_text = std::fs::read_to_string(where_file(session)).unwrap_or_default();
266        let edits = std::fs::read_to_string(edits_file(session)).unwrap_or_default();
267        let latest = std::fs::read_to_string(state_file(session).with_extension("lastprompt")).unwrap_or_default();
268        crate::concepts::apply(&mut all, &crate::concepts::of_work(&edits, &where_text, &latest), crate::concepts::beta());
269    }
270    (all, withheld)
271}
272
273/// The lessons file the conscience reads: BORG_CONSCIENCE, default $BORG_HOME/lessons.jsonl.
274fn lessons_path() -> PathBuf {
275    std::env::var_os("BORG_CONSCIENCE").map(PathBuf::from).unwrap_or_else(|| learn::home().join("lessons.jsonl"))
276}
277
278/// What the drone did, for R1 (`pending`, `early_acted`): one line per Edit/Write/MultiEdit/NotebookEdit
279/// (path and the start of the new text); at each Stop block the blocked message and `#borg-asked <ids>`;
280/// the drone's answer at the turn end after it (`#borg-answer <message>`); `#borg-early <ids>` per early
281/// delivery. Kept out of the state file so the drone's code never feeds the conscience's matching.
282fn edits_file(session: &str) -> PathBuf {
283    state_file(session).with_extension("edits")
284}
285
286/// What Borg cost this session (R3b), one line per delivery: `block questions=N chars=M` for a Stop
287/// block (an extra round), `free event=E lessons=N chars=M` for context on a hook that forces no round.
288/// `borg status` sums them (prefilter.rs `tally`).
289fn cost_file(session: &str) -> PathBuf {
290    state_file(session).with_extension("cost")
291}
292
293/// BORG_STOP_GATE=edits: one line per tool call this turn that changed state (prefilter.rs `changes`);
294/// the Stop removes it.
295fn changes_file(session: &str) -> PathBuf {
296    state_file(session).with_extension("changes")
297}
298
299fn log_cost(session: &str, line: &str) {
300    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(cost_file(session)) {
301        let _ = writeln!(f, "{line}");
302    }
303}
304
305/// Lessons delivered early (`#borg-early` markers in the .edits file, and graded lessons handed over free,
306/// `#borg-queued`) that the drone acted on since: its work after the delivery brings in the lesson's words
307/// (memory.rs `resolve`, `final_msg` included). Stop leaves them alone; an early lesson not acted on can
308/// still be asked at Stop. A later delivery's own work line (`WORK`: a free delivery's, a placebo's; a live
309/// ask's before 0.7.6) is neither: it is the work before that delivery, written whatever its route, so how
310/// a later lesson was routed never decides whether this one was acted on.
311fn early_acted(session: &str, all: &[conscience::Challenge], final_msg: &str, upto: Option<usize>) -> std::collections::HashSet<String> {
312    let edits = edits_upto(session, upto);
313    let state = std::fs::read_to_string(state_file(session)).unwrap_or_default();
314    let lines: Vec<&str> = edits.lines().filter(|l| !l.starts_with(WORK)).collect();
315    let mut out = std::collections::HashSet::new();
316    for (p, l) in lines.iter().enumerate() {
317        let Some(list) = l.strip_prefix("#borg-early ").or_else(|| l.strip_prefix("#borg-queued ")) else { continue };
318        let ids: std::collections::HashSet<&str> = ids_in(list).collect();
319        let asked: Vec<&conscience::Challenge> = all.iter().filter(|c| ids.contains(c.id.as_str())).collect();
320        let after: Vec<&str> = lines[p + 1..].iter().copied().filter(|l| !l.starts_with("#borg-")).collect();
321        let upto = state.find(&format!("{l}\n")).map_or(state.as_str(), |i| &state[..i]);
322        let before = format!("{upto}\n{}", lines[..p].join("\n"));
323        out.extend(crate::memory::resolve(&asked, all, &before, &after.join("\n"), final_msg, crate::memory::NOVEL_WORDS).0);
324    }
325    out
326}
327
328/// The .edits file up to its length `upto` (a detached check's stop, check_async.rs `Job::edits`): the
329/// turn it judges and the deliveries before it, never the next turn's lines; all of it for None.
330fn edits_upto(session: &str, upto: Option<usize>) -> String {
331    let mut edits = std::fs::read_to_string(edits_file(session)).unwrap_or_default();
332    if let Some(n) = upto.filter(|n| edits.is_char_boundary(*n)) {
333        edits.truncate(n);
334    }
335    edits
336}
337
338/// Done in this session: asked as a question (at a stop, or live before 0.7.6), or handed over (early,
339/// free) and acted on since (`early_acted`, its `acted`). The one definition a stop's pool leaves out: a
340/// lesson handed over but not acted on may be graded again, and when its grade asks it the question
341/// reaches the drone.
342fn done_in<'a>(state: &'a str, acted: &std::collections::HashSet<String>) -> std::collections::HashSet<&'a str> {
343    let mut done = listed(state, "#borg-asked ");
344    done.extend(listed(state, "#borg-early ").into_iter().chain(listed(state, "#borg-queued ")).filter(|i| acted.contains(*i)));
345    done
346}
347
348/// The lessons an earlier graded check of the session applied (its `#borg-graded` lines, `graded_line`):
349/// in play for the pool log (pools.rs `IN_PLAY`).
350fn applied_in(state: &str) -> impl Iterator<Item = &str> {
351    state.lines().filter_map(|l| l.strip_prefix("#borg-graded n=")).flat_map(|l| l.split_once(' ').map_or("", |(_, items)| items).split(',')).filter_map(
352        |item| {
353            let id = item.rsplitn(3, ':').nth(2).unwrap_or_default();
354            (!id.is_empty()).then_some(id)
355        },
356    )
357}
358
359/// Asked or handed over in this session, acted on or not: free context comes once per session, so
360/// none of these goes free again (`queued`, `early`), done or not.
361fn had_in(state: &str) -> std::collections::HashSet<&str> {
362    let mut had = listed(state, "#borg-asked ");
363    had.extend(listed(state, "#borg-early "));
364    had.extend(listed(state, "#borg-queued "));
365    had
366}
367
368/// The .edits line before a free delivery's or a placebo's markers (a live ask's before 0.7.6): the work so far
369/// (`work_so_far`), the evidence `pending` reads before it as a question's blocked message. Not work
370/// itself: `early_acted` and a judge's "work since" skip it, as every `#borg-` line, so an earlier
371/// delivery's evidence never depends on how a later lesson was routed (`note_work`).
372const WORK: &str = "#borg-work ";
373
374/// An edit tool's paths (every file it names, so the cut never drops one) and the start of its new
375/// text, on one line; None for other tools.
376pub(crate) fn edit_text(tool: &str, input: &Value) -> Option<String> {
377    if !["edit", "write", "multiedit", "notebookedit"].contains(&tool.to_ascii_lowercase().as_str()) {
378        return None;
379    }
380    let s = |v: &Value, k: &str| v.get(k).and_then(Value::as_str).unwrap_or_default().to_string();
381    let mut text = [s(input, "file_path"), s(input, "notebook_path")].join(" ");
382    for p in further_files(input) {
383        text.push(' ');
384        text.push_str(p);
385    }
386    for k in ["new_string", "content", "new_source"] {
387        text.push(' ');
388        text.push_str(&s(input, k));
389    }
390    for e in input.get("edits").and_then(Value::as_array).into_iter().flatten() {
391        text.push(' ');
392        text.push_str(&s(e, "new_string"));
393    }
394    Some(text.split_whitespace().collect::<Vec<_>>().join(" ").chars().take(EDIT_CHARS).collect())
395}
396
397/// An edit's line in the `.edits` and turn files (`edit_text`): its paths and the start of its new text.
398/// Guard: bounds one evidence line (a whole written file would otherwise go in).
399const EDIT_CHARS: usize = 600;
400
401/// The longest drone message one evidence line of the `.edits` file keeps: the blocked message before
402/// a question, the work so far before a free delivery, the answer after a block (`pending`).
403/// Guard: bounds a line of a file every judge call reads; at least what the judge reads after a
404/// delivery (`predict::JUDGE_PART` three times), so the cut never takes evidence the judge would see.
405/// Not a context budget: none of it is injected into the drone.
406const EVIDENCE_CHARS: usize = 4000;
407const _: () = assert!(EVIDENCE_CHARS >= 3 * predict::JUDGE_PART);
408
409/// `text` on one line, at most `n` characters.
410pub(crate) fn one_line(text: &str, n: usize) -> String {
411    text.split_whitespace().collect::<Vec<_>>().join(" ").chars().take(n).collect()
412}
413
414/// R1: the first turn end after a Stop block is the drone's answer to it. Its message (with any
415/// `<id>: applies|skip`) goes to the .edits file as `#borg-answer`: evidence for the judge (`pending`),
416/// never a label by itself.
417fn note_answer(h: &dyn Harness, session: &str, input: &Value) {
418    let edits = std::fs::read_to_string(edits_file(session)).unwrap_or_default();
419    let lines: Vec<&str> = edits.lines().collect();
420    let Some(i) = lines.iter().rposition(|l| l.starts_with("#borg-asked ")) else { return };
421    if lines[i + 1..].iter().any(|l| l.starts_with("#borg-answer ")) {
422        return;
423    }
424    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(edits_file(session)) {
425        let _ = writeln!(f, "#borg-answer {}", one_line(&full_proposal(h, input), EVIDENCE_CHARS));
426    }
427}
428
429/// Pending lessons sent to a check's judge per call at most (the most recently delivered). Guard: bounds
430/// one check's judge section, whatever a session delivered, so the stop's latency stays bounded; derived
431/// from one full block (conscience.rs `CAP`), whose lessons share their evidence in the prompt. A lesson
432/// it leaves out is logged (`#borg-unjudged <ids> why=slots`), so its hit rate is measured, and is judged
433/// at the session's end at the latest, which has no slots (`judge_session`).
434const MAX_PENDING: usize = conscience::CAP;
435/// A lesson whose verdicts stay pending (unclear, missing) is sent to a check's judge this often at
436/// most; after that it waits for the session-end judge (`judge_session`), which judges it once more
437/// whatever its tries, then for `borg learn` to settle it as `asked_not_corrected` (censored, reason
438/// `tries`). Guard: a lesson the judge never settles would otherwise ride on every later call. A failed
439/// call is no try: its lessons stay pending as they were (`#borg-unjudged … why=failed`).
440const MAX_JUDGE_TRIES: usize = 2;
441
442/// The session-end judge's prompt budget (`end_budget`) before any evidence of the judge's effective
443/// length: characters of a batch's whole prompt (`predict::judge_prompt`). Baseline (0.7.6
444/// pre-registration, 2026-11-01): 0.7.5's 24,000 characters of work and its 40 lessons at
445/// `predict::JUDGE_PART` (1,200) each. Not the largest prompt 0.7.5's one session-end call could send:
446/// with every lesson's full evidence (1,200 before, 3,600 after, its question and consequence) that was
447/// about 220,000; the pre-registered value is kept, well below it, so batching never asks more of the
448/// judge than 0.7.5 could before the measured effective length says it may.
449pub(crate) const END_START_BUDGET: usize = 72_000;
450
451/// A session-end batch's lessons (`predict::judge_section`) take at most 1/this of its budget; the
452/// session's work gets the rest (`batches`). Guard: the work, every lesson's evidence, is never crowded
453/// out by the lessons beside it.
454const END_LESSONS_SHARE: usize = 2;
455
456/// The session-end judge's budget for one pass (`end_budget`).
457#[derive(Debug, Clone, Copy, PartialEq)]
458pub(crate) struct EndBudget {
459    /// Characters a batch's whole prompt may take (`predict::judge_prompt`).
460    pub chars: usize,
461    /// When the pass drew whether to explore a larger budget: the chance it explored and whether it
462    /// did (the cost line's `psi=` and `explore=`).
463    pub draw: Option<(f64, bool)>,
464}
465
466/// The session-end judge's prompt budget for this pass, for the judge `model` (`judge_model`): its
467/// measured effective length from the store's judge cost lines (judge_length.rs `length`, the `.cost`
468/// files of `state`), or, with the chance `psi` the estimate gives while the step above it is open, that
469/// step, drawn from `seed` and logged (`draw`). `BORG_END_BUDGET=<chars>` pins it (an evaluation switch:
470/// no estimate, no draw).
471pub(crate) fn end_budget(state: &std::path::Path, model: &str, seed: &str) -> EndBudget {
472    if let Some(pinned) = std::env::var("BORG_END_BUDGET").ok().and_then(|v| v.trim().parse::<usize>().ok()).filter(|n| *n > 0) {
473        return EndBudget { chars: pinned, draw: None };
474    }
475    let len = crate::judge_length::length(state, model);
476    let mut h = blake3::Hasher::new();
477    h.update(b"borg-end-explore-v1\0");
478    h.update(seed.as_bytes());
479    let bytes: [u8; 8] = h.finalize().as_bytes()[..8].try_into().unwrap_or_default();
480    let u = crate::memory::Rng::new(u64::from_le_bytes(bytes)).unit();
481    match len.explore {
482        Some(up) if u < len.psi => EndBudget { chars: up, draw: Some((len.psi, true)) },
483        _ => EndBudget { chars: len.chars, draw: Some((len.psi, false)) },
484    }
485}
486
487/// The judge's model as cost lines name it (`model=`): the first runner `spec` picks and the model it
488/// asks (runner.rs `Choice::label`); empty when none.
489pub(crate) fn judge_model(spec: &str) -> String {
490    crate::runner::choose(Some(spec)).first().map(crate::runner::Choice::label).unwrap_or_default()
491}
492
493/// How a session-end batch cuts a part of the session's work longer than its share of the budget
494/// (`BORG_END_WORK`): its newest characters (`newest`, 0.7.5's), from the batch's earliest delivery on
495/// (`after`), or half right after that delivery and half the end (`ends`). A part without delivery
496/// markers (the agent's messages) keeps its newest characters under each. Which one reads more verdicts
497/// out of a long session was the 0.7.6 backtest's question (docs/evidence.md, Lesson selection, 0.7.6
498/// pre-registration): `ends` answered the most on lessons delivered long before the end (196 of 196,
499/// `newest` 195, `after` 192), so it is the default.
500#[derive(Debug, Clone, Copy, PartialEq, Eq)]
501enum Cut {
502    Newest,
503    After,
504    Ends,
505}
506
507impl Cut {
508    /// `BORG_END_WORK`, else `ends`; an unknown word is the default too.
509    fn from_env() -> Cut {
510        match std::env::var("BORG_END_WORK").unwrap_or_default().trim().to_ascii_lowercase().as_str() {
511            "newest" => Cut::Newest,
512            "after" => Cut::After,
513            _ => Cut::Ends,
514        }
515    }
516
517    /// Its word on the cost line (`cut=`).
518    fn word(self) -> &'static str {
519        match self {
520            Cut::Newest => "newest",
521            Cut::After => "after",
522            Cut::Ends => "ends",
523        }
524    }
525
526    /// `n` of `text`'s characters (`n` < its length), `at` the batch's earliest delivery in it (none:
527    /// the newest), each cut marked `…`. `after`: from `at` on, its start first, or the newest `n` when
528    /// what follows `at` is shorter; `ends`: half from `at`, half the end, or `after`'s when the two meet.
529    fn window(self, text: &[char], n: usize, at: Option<usize>) -> String {
530        let all = text.len();
531        let s = |r: std::ops::Range<usize>| text[r].iter().collect::<String>();
532        let mark = |cut: bool| if cut { "…" } else { "" };
533        let at = at.map(|a| a.min(all));
534        match (self, at) {
535            (Cut::After, Some(a)) => {
536                let start = a.min(all - n);
537                format!("{}{}{}", mark(start > 0), s(start..start + n), mark(start + n < all))
538            }
539            (Cut::Ends, Some(a)) if a + n / 2 < all - (n - n / 2) => format!("{}{}…{}", mark(a > 0), s(a..a + n / 2), s(all - (n - n / 2)..all)),
540            (Cut::Ends, Some(_)) => Cut::After.window(text, n, at),
541            _ => format!("…{}", s(all - n..all)),
542        }
543    }
544}
545
546/// How long one session-end judge call (`judge_session`) may take: a learn-side call's limit
547/// (`BORG_LEARN_TIMEOUT_S`), at most twice the check's (`BORG_PREDICT_TIMEOUT_S`). Derived: the session
548/// lock it holds reads stale after twice the check's limit plus a slack (live.rs `stale_ms`); the
549/// pass renews the lock before each call (live.rs `Guard::renew`), so a stop of a session resumed
550/// meanwhile never takes the lock over and judges the same deliveries again.
551fn end_timeout() -> std::time::Duration {
552    learn::learn_timeout().min(predict::timeout().saturating_mul(2))
553}
554
555/// After a session-end judge call, in the state file: `#borg-end-judged <ids>`, the lessons it was
556/// shown. Their latest deliveries are settled by it whatever it said (`pending`), so neither a later
557/// check nor a second session-end pass judges them again; one it left pending settles at `borg learn`
558/// as having used its tries (`unjudged`). A later delivery of the lesson opens it again.
559const END_JUDGED: &str = "#borg-end-judged ";
560
561/// When a judge call runs (`pending`), which decides what it judges: a turn end (a stop's check) or the
562/// session's end (`judge_session`).
563#[derive(Debug, Clone, Copy, PartialEq, Eq)]
564enum When {
565    Stop,
566    End,
567}
568
569/// The markers of a delivery of lessons to the drone, in the state and .edits files: asked at a Stop
570/// (`#borg-asked`; live, before 0.7.6, with a `#borg-live` line after it), delivered early, handed over
571/// free (`#borg-queued`); and a placebo (`#borg-placebo`), never shown but judged as a free delivery is.
572/// Each ends in ` @<check>`, the check that routed its lessons (select.rs `check_id`), once checks had ids.
573const DELIVERED: [&str; 4] = ["#borg-asked ", "#borg-early ", "#borg-queued ", "#borg-placebo "];
574
575/// Right after a stop's `#borg-asked` line: `#borg-routed <id>=<check>,…`, the lessons of the block
576/// another check routed (a live check's ask still waiting at the stop, before 0.7.6), where the block's
577/// own tag is the stop's check. Read, no longer written.
578const ROUTED: &str = "#borg-routed ";
579
580/// Whether `line` is part of a delivery's markers (`DELIVERED`, `#borg-live`, `ROUTED`): one delivery
581/// writes its work line once, then a marker per routing check.
582fn is_marker(line: &str) -> bool {
583    DELIVERED.iter().any(|m| line.starts_with(m)) || line.starts_with("#borg-live ") || line.starts_with(ROUTED)
584}
585
586/// Whether the judge sees question `lesson` of `check` without the drone's `<id>: applies|skip`: a
587/// fair coin from blake3 over both, so a retry strips alike and a replay knows it (outcome.rs
588/// `stripped`; a test predicts a judge prompt by it). Method: a random half, to measure what the
589/// declaration adds to a verdict.
590#[must_use]
591pub fn stripped(check: &str, lesson: &str) -> bool {
592    let mut h = blake3::Hasher::new();
593    h.update(b"borg-strip-v1\0");
594    h.update(format!("{check}\0{lesson}").as_bytes());
595    h.finalize().as_bytes()[0] & 1 == 1
596}
597
598/// Of the lessons delivered in a session's state file `state`, since each one's latest delivery: how many
599/// verdicts it had (its tries) and whether one settled it (a label other than pending, or a session-end
600/// judge's `END_JUDGED`). A delivery opens its lesson again.
601fn verdicts_in(state: &str) -> (std::collections::HashMap<&str, usize>, std::collections::HashSet<&str>) {
602    let mut tries: std::collections::HashMap<&str, usize> = std::collections::HashMap::new();
603    let mut settled = std::collections::HashSet::new();
604    for l in state.lines() {
605        if let Some(l) = l.strip_prefix("#borg-judged ") {
606            let (id, label) = l.split_once('=').unwrap_or((l, ""));
607            *tries.entry(id).or_default() += 1;
608            if label.trim() != "pending" {
609                settled.insert(id);
610            }
611        } else if let Some(list) = l.strip_prefix(END_JUDGED) {
612            settled.extend(ids_in(list));
613        } else if let Some(list) = DELIVERED.iter().find_map(|m| l.strip_prefix(m)) {
614            for id in ids_in(list) {
615                tries.remove(id);
616                settled.remove(id);
617            }
618        }
619    }
620    (tries, settled)
621}
622
623/// The lessons of `state` whose latest delivery no verdict settled yet: what a session-end judge
624/// (`judge_session`) judges, all of them.
625fn open_in(state: &str) -> std::collections::HashSet<&str> {
626    let (_, settled) = verdicts_in(state);
627    let mut open: std::collections::HashSet<&str> = DELIVERED.iter().flat_map(|m| listed(state, m)).collect();
628    open.retain(|id| !settled.contains(id));
629    open
630}
631
632/// What one judge call takes (`pending`): the pending lessons with their evidence, whether a question
633/// is among them, and the ids the call has no slot for (`MAX_PENDING`; none at the session's end).
634struct Judging<'a> {
635    pending: Vec<predict::Pending<'a>>,
636    question: bool,
637    crowded: Vec<String>,
638}
639
640/// R1: lessons delivered earlier in this session (asked at a Stop, or live before 0.7.6, delivered early,
641/// handed over free, or a placebo) that have no verdict yet, with the evidence for the consequence model
642/// (predict.rs `Pending`), the same for every route (judge v2): the work before the delivery (at a Stop the
643/// blocked message, before a free delivery or a placebo the work so far), and after it "work since" (a
644/// question's answer, then the drone's edits and later messages) and this turn's message (`final_msg`),
645/// each capped by `predict::JUDGE_PART`. A random half of the questions (`stripped`) lose the drone's
646/// declaration from all of the call's evidence: their own, every sibling's, and this turn's message, which
647/// may be the answer itself (`check` takes it out of the work the call shows too). `#borg-judged
648/// <id>=<label>` lines in the state file record verdicts, each for the lesson's latest delivery before it:
649/// a lesson delivered early, judged, then asked at a Stop is judged again on the question. A delivery still
650/// open after a stop's call (or at a stop that makes none) waits for a later stop, and at the latest for
651/// the session-end judge (`judge_session`). Of the `MAX_PENDING` slots, questions take theirs first (the
652/// newest), free deliveries what is left: a free hand-over never pushes a question out of its judge. At the
653/// session's end (`When::End`) nothing is held, a lesson's tries do not bar it (`MAX_JUDGE_TRIES`) and
654/// there are no slots: every open delivery is judged, in batches (`batches`), in delivery order (the order
655/// of the result). A lesson a session-end judge was shown (`END_JUDGED`) is settled for every later call.
656/// `judging`: this call's check, the coin's seed for a delivery from before checks had ids.
657fn pending<'a>(session: &str, state: &str, all: &'a [conscience::Challenge], final_msg: &str, when: When, judging: &str) -> Judging<'a> {
658    pending_upto(session, state, all, final_msg, when, judging, None)
659}
660
661/// `pending` with the .edits file read up to `edits` (`edits_upto`): a detached check's, as at its stop.
662fn pending_upto<'a>(
663    session: &str,
664    state: &str,
665    all: &'a [conscience::Challenge],
666    final_msg: &str,
667    when: When,
668    judging: &str,
669    edits: Option<usize>,
670) -> Judging<'a> {
671    let (tries, settled) = verdicts_in(state);
672    let edits = edits_upto(session, edits);
673    let lines: Vec<&str> = edits.lines().collect();
674    let part = predict::JUDGE_PART;
675    /// One delivery's evidence, unclipped, and its fresh lessons: (id, routing check, stripped).
676    struct Delivery<'l> {
677        channel: &'static str,
678        /// A question (`#borg-asked`), not a free delivery or a placebo.
679        question: bool,
680        before: &'l str,
681        answers: Vec<&'l str>,
682        work: Vec<&'l str>,
683        fresh: Vec<(&'l str, &'l str, bool)>,
684    }
685    let mut deliveries: Vec<Delivery> = Vec::new();
686    for (p, l) in lines.iter().enumerate() {
687        let Some((marker, list)) = DELIVERED.iter().find_map(|m| l.strip_prefix(m).map(|rest| (*m, rest))) else { continue };
688        let question = marker == "#borg-asked ";
689        let later = &lines[p + 1..];
690        let channel = match marker {
691            "#borg-asked " if later.first().is_some_and(|n| n.starts_with("#borg-live ")) => "live",
692            "#borg-asked " => "stop",
693            "#borg-early " => "early",
694            "#borg-queued " => "queued",
695            _ => outcome::PLACEBO,
696        };
697        let answers: Vec<&str> = if question { later.iter().filter_map(|l| l.strip_prefix("#borg-answer ")).collect() } else { Vec::new() };
698        let work: Vec<&str> = later.iter().copied().filter(|l| !l.starts_with("#borg-")).collect();
699        let (list, tag) = tagged(list);
700        // The work before: the line before this delivery's markers, a delivery's own work line (`WORK`)
701        // read as its text; nothing when that is another bookkeeping line.
702        let before = match lines[..p].iter().rev().find(|b| !is_marker(b)) {
703            Some(b) if b.starts_with(WORK) => &b[WORK.len()..],
704            Some(b) if !b.starts_with("#borg-") => b,
705            _ => "",
706        };
707        let routed: std::collections::HashMap<&str, &str> =
708            later.first().and_then(|n| n.strip_prefix(ROUTED)).into_iter().flat_map(|r| r.split(',')).filter_map(|it| it.split_once('=')).collect();
709        let fresh: Vec<&str> = list
710            .split(',')
711            .map(str::trim)
712            .filter(|i| !i.is_empty() && !settled.contains(i) && (when == When::End || tries.get(i).copied().unwrap_or(0) < MAX_JUDGE_TRIES))
713            .filter(|i| all.iter().any(|c| c.id == *i))
714            .collect();
715        let check_of = |id: &str| routed.get(id).copied().or(tag).unwrap_or_default();
716        let fresh =
717            fresh.into_iter().map(|id| (id, check_of(id), question && stripped(if check_of(id).is_empty() { judging } else { check_of(id) }, id))).collect();
718        deliveries.push(Delivery { channel, question, before, answers, work, fresh });
719    }
720    // Every stripped question's declaration leaves all of this call's evidence before any cut could halve
721    // one: its own, a sibling's (a later blocked message may hold the answer) and this turn's message.
722    let strip: Vec<&str> = deliveries.iter().flat_map(|d| &d.fresh).filter(|f| f.2).map(|f| f.0).collect();
723    let blind = |text: &str| strip.iter().fold(text.to_string(), |t, id| crate::memory::strip_declaration(&t, id));
724    let turn = predict::clip(&one_line(&blind(final_msg), part + 1), part);
725    // Each with whether it was a question.
726    let mut out: Vec<(predict::Pending, bool)> = Vec::new();
727    for d in deliveries {
728        // One "after" for the whole delivery, so lessons delivered together share it.
729        let answer = blind(&d.answers.join(" "));
730        let said = answer.to_lowercase();
731        let since: Vec<&str> = std::iter::once(answer.as_str()).filter(|a| !a.is_empty()).chain(d.work.iter().copied()).collect();
732        let after = format!("work since: {}\nthis turn: {turn}", predict::clip(&blind(&since.join(" | ")), 2 * part));
733        let before = blind(d.before);
734        for (id, check, stripped) in d.fresh {
735            let Some(c) = all.iter().find(|c| c.id == id) else { continue };
736            // A later delivery of the same lesson is the one judged: its evidence replaces the earlier one's.
737            out.retain(|x| x.0.lesson.id != id);
738            let declared = if stripped { None } else { crate::memory::declared(&said, &id.to_lowercase()) };
739            let pending =
740                predict::Pending { lesson: c, before: before.clone(), after: after.clone(), declared, check: check.to_string(), channel: d.channel, stripped };
741            out.push((pending, d.question));
742        }
743    }
744    let slots = if when == When::End { usize::MAX } else { MAX_PENDING };
745    let questions = out.iter().filter(|x| x.1).count().min(slots);
746    let mut room = [slots - questions, questions];
747    let (mut kept, mut crowded): (Vec<(predict::Pending, bool)>, Vec<String>) = (Vec::new(), Vec::new());
748    for x in out.into_iter().rev() {
749        let r = &mut room[usize::from(x.1)];
750        if *r > 0 {
751            *r -= 1;
752            kept.push(x);
753        } else {
754            crowded.push(x.0.lesson.id.clone());
755        }
756    }
757    kept.reverse();
758    crowded.sort();
759    Judging { pending: kept.into_iter().map(|x| x.0).collect(), question: questions > 0, crowded }
760}
761
762/// `text` without the declarations of the stripped questions among `pending` (`stripped`): the work a
763/// judge call shows beside them, which may hold the answer to one (this turn's message at the stop).
764fn without_declarations<'t>(text: &'t str, pending: &[predict::Pending]) -> std::borrow::Cow<'t, str> {
765    pending.iter().filter(|p| p.stripped).fold(std::borrow::Cow::Borrowed(text), |t, p| crate::memory::strip_declaration(&t, &p.lesson.id).into())
766}
767
768/// `#borg-unjudged <ids> why=<reason>`: pending lessons a judge call did not judge, and why
769/// (outcome.rs `UNJUDGED_SLOTS`, `UNJUDGED_FAILED`), so `borg learn` can say why each settled without
770/// a verdict (`unjudged`). Nothing for no ids.
771fn note_unjudged(session: &str, ids: &[String], why: &str) {
772    if !ids.is_empty() {
773        append(session, &format!("#borg-unjudged {} why={why}", ids.join(",")));
774    }
775}
776
777/// Why each lesson delivered in a session's state file `state` has no verdict, as `borg learn` settles
778/// it (outcome.rs `settle_with`): its tries ran out on unclear or missing verdicts since its latest
779/// delivery, the session-end judge's included (`END_JUDGED`; `tries`), else the last judge call that
780/// left it out (`slots`, `failed`), else no call came after its delivery (`no_call`).
781pub(crate) fn unjudged(state: &str) -> std::collections::HashMap<String, &'static str> {
782    let mut why: std::collections::HashMap<String, &'static str> = std::collections::HashMap::new();
783    let mut tries: std::collections::HashMap<&str, usize> = std::collections::HashMap::new();
784    for l in state.lines() {
785        if let Some(list) = DELIVERED.iter().find_map(|m| l.strip_prefix(m)) {
786            for id in ids_in(list) {
787                tries.remove(id);
788                why.insert(id.to_string(), outcome::UNJUDGED_NO_CALL);
789            }
790        } else if let Some(l) = l.strip_prefix("#borg-judged ") {
791            let (id, label) = l.split_once('=').unwrap_or((l, ""));
792            if label.trim() == "pending" {
793                let n = tries.entry(id).or_default();
794                *n += 1;
795                if *n >= MAX_JUDGE_TRIES {
796                    why.insert(id.to_string(), outcome::UNJUDGED_TRIES);
797                }
798            }
799        } else if let Some(list) = l.strip_prefix(END_JUDGED) {
800            // The session-end judge was its last try, whatever its count.
801            for id in ids_in(list) {
802                tries.insert(id, MAX_JUDGE_TRIES);
803                why.insert(id.to_string(), outcome::UNJUDGED_TRIES);
804            }
805        } else if let Some(rest) = l.strip_prefix("#borg-unjudged ") {
806            let (list, reason) = rest.rsplit_once(" why=").unwrap_or((rest, ""));
807            let reason = match reason.trim() {
808                "slots" => outcome::UNJUDGED_SLOTS,
809                "failed" => outcome::UNJUDGED_FAILED,
810                _ => continue,
811            };
812            for id in list.split(',').map(str::trim).filter(|i| !i.is_empty()) {
813                if tries.get(id).copied().unwrap_or(0) < MAX_JUDGE_TRIES {
814                    why.insert(id.to_string(), reason);
815                }
816            }
817        }
818    }
819    why
820}
821
822/// R1: the consequence model's verdicts on the pending lessons it was shown (predict.rs `Judged`), each
823/// recorded under its own kind (predict.rs `Verdict::kind`): `asked_then_changed` for a change that
824/// avoids the consequence, `asked_ignored` for no change on a point that came up, `asked_not_arisen`
825/// for a point that never came up (censored for beliefs); these three settle the delivery
826/// (`#borg-judged <id>=<verdict>`), a `not_arisen` only at the session's end (`at` outcome.rs
827/// `AT_END`): at a check the point may still come up later, so it stays pending for later stops and
828/// the session-end judge (`judge_session`), a try as an unclear one is. An unclear verdict (`asked_unclear`) and a change that still leads
829/// to the consequence (`asked_consequence_remains`) are measurements only and stay pending
830/// (`#borg-judged <id>=pending`), as a missing verdict does. Each result names its delivery's channel
831/// (outcome.rs `PLACEBO` for a placebo, never a result about the lesson), the check that routed it, the
832/// judge's version (predict.rs `JUDGE_VERSION`), whether the declaration was stripped, per lesson the
833/// judge's redacted reason (`reasons`), the session's `repo`, and where it was judged (`at`: empty at a
834/// check, outcome.rs `AT_END` at the session's end). Logs the call's judging cost in the .cost file
835/// (`judge_line`, then ` model=<judge>`, then ` at=<at>` when not at a check, then `logged`'s fields).
836fn record_verdicts(session: &str, repo: &str, at: &str, pending: &[predict::Pending], pred: &predict::Prediction, logged: &Logged) {
837    // One record per (kind, channel, check, stripped): the statistics split what the deliveries led to by
838    // how they were made. Each lesson with its reason.
839    type Key<'k> = (&'k str, &'k str, &'k str, bool);
840    let mut records: std::collections::BTreeMap<Key, Vec<(String, String)>> = std::collections::BTreeMap::new();
841    for p in pending {
842        let judged = pred.resolved.iter().find(|j| j.id == p.lesson.id);
843        // A point that has not come up by a check may still come up later in the session: only the
844        // session-end judge's `not_arisen`, with the whole session in view, settles the delivery.
845        let word = match judged {
846            Some(j) if j.label().is_some() && (j.verdict != predict::Verdict::NotArisen || at == outcome::AT_END) => j.verdict.word(),
847            _ => "pending",
848        };
849        append(session, &format!("#borg-judged {}={word}", p.lesson.id));
850        let Some(j) = judged else { continue };
851        records.entry((j.verdict.kind(), p.channel, p.check.as_str(), p.stripped)).or_default().push((p.lesson.id.clone(), j.reason.clone()));
852    }
853    let at_field = if at.is_empty() { String::new() } else { format!(" at={at}") };
854    let line = judge_line(logged.call, pending, &pred.resolved, logged.chars, &pred.returned);
855    log_cost(session, &format!("{line} model={}{at_field}{}", logged.model, logged.fields));
856    if learning() {
857        let (home, ms) = (learn::home(), learn::now_ms());
858        // The statistics count one event per kind and channel, whatever checks routed its lessons.
859        let mut counted: std::collections::BTreeMap<(&str, &str), Vec<String>> = std::collections::BTreeMap::new();
860        for ((kind, channel, check, stripped), judged) in records {
861            let (lessons, reasons): (Vec<String>, Vec<String>) = judged.into_iter().unzip();
862            counted.entry((kind, channel)).or_default().extend(lessons.iter().cloned());
863            let o = outcome::Outcome {
864                kind: kind.into(),
865                session: session.into(),
866                lessons,
867                ms,
868                repo: repo.into(),
869                channel: channel.into(),
870                check: check.into(),
871                judge: predict::JUDGE_VERSION,
872                stripped,
873                at: at.into(),
874                // None at all: left out, as an older record.
875                reasons: if reasons.iter().all(String::is_empty) { Vec::new() } else { reasons },
876                ..Default::default()
877            };
878            let _ = learn::append_jsonl(&home.join("outcomes.jsonl"), &o);
879        }
880        for ((kind, channel), lessons) in counted {
881            outcome::count_result(
882                &home,
883                &outcome::Outcome { kind: kind.into(), session: session.into(), lessons, ms, repo: repo.into(), channel: channel.into(), ..Default::default() },
884            );
885        }
886    }
887}
888
889/// What a judge call's cost line says besides its verdicts (`record_verdicts`): the call's number
890/// among the session's model calls, its prompt's characters (`judge_line`), the judge's model
891/// (`judge_model`), and fields of its own after ` at=` (a check's whole prompt's characters, `prompt=`; a
892/// session-end batch's `batch=`, `budget=`, `psi=` and `explore=` when it drew, `cut=`).
893struct Logged {
894    call: usize,
895    chars: usize,
896    model: String,
897    fields: String,
898}
899
900/// A judge call's cost line (`record_verdicts`): `judge call=<nth model call of the session> pending=N
901/// resolved=K chars=<prompt chars added>`, K the verdicts that settle (changed, ignored, not_arisen);
902/// then the ids the answer named (`returned=a,b`, predict.rs `Prediction::returned`) and those of them
903/// no pending lesson has (`unmatched=`), each empty for none; then the verdicts on pending lessons by
904/// word, every word in `Verdict::WORDS` order (`verdicts=changed:1,ignored:0,…`), and the characters
905/// of their reasons (`reason_chars=`, what the answer added). A pending lesson in neither list was left
906/// out by the model; one returned but not resolved got a malformed verdict. Zero tokens, no prompt text.
907fn judge_line(call: usize, pending: &[predict::Pending], resolved: &[predict::Judged], chars: usize, returned: &[String]) -> String {
908    let unmatched: Vec<&str> = returned.iter().map(String::as_str).filter(|id| !pending.iter().any(|p| p.lesson.id == *id)).collect();
909    let mine: Vec<&predict::Judged> = resolved.iter().filter(|j| pending.iter().any(|p| p.lesson.id == j.id)).collect();
910    let settled = mine.iter().filter(|j| j.label().is_some()).count();
911    let verdicts: Vec<String> = predict::Verdict::ALL.iter().map(|v| format!("{}:{}", v.word(), mine.iter().filter(|j| j.verdict == *v).count())).collect();
912    let reason_chars: usize = mine.iter().map(|j| j.reason.chars().count()).sum();
913    format!(
914        "judge call={call} pending={} resolved={settled} chars={chars} returned={} unmatched={} verdicts={} reason_chars={reason_chars}",
915        pending.len(),
916        returned.join(","),
917        unmatched.join(","),
918        verdicts.join(",")
919    )
920}
921
922/// The session-end judge (0.7.6), run by `borg learn --session` before it learns (main.rs `learn_cmd`):
923/// every delivery of `session` (any channel) whose latest delivery no verdict settled yet (none came, or
924/// only unclear, consequence_remains or missing ones, its tries spent or not) is judged, beside as much of
925/// the session's work as the prompt's budget leaves (`end_budget`, `session_sources`, `within`). The
926/// lessons go in batches by delivery order, each prompt within the budget (`batches`), each batch its
927/// own model call, `runner::JOBS` at a time, each within `end_timeout`, under the session lock,
928/// renewed before each call (live.rs `Guard::renew`). Each batch records as soon as its call returns,
929/// while the others still run, so a pass stopped partway keeps what it was already told (and paid for).
930/// An answered batch records as a check's judge does (`record_verdicts`, `at` outcome.rs `AT_END`, its
931/// cost line naming `batch=<i>/<n>`, `budget=` and `cut=`) and writes the state lines `#borg-judged`,
932/// then its own `END_JUDGED`: what it judged settles there, and `borg learn` then settles the rest as
933/// `asked_not_corrected` with its reason (`unjudged`). Each call is counted as a `model_call` with why
934/// `judge`. A failed batch is no try, for its own lessons alone (`#borg-unjudged … why=failed`), and its
935/// cost line `judge-failed` keeps its lessons as sent and unanswered for the judge's effective length
936/// (judge_length.rs); after `runner::MAX_FAILURES` failed calls before the first answer the judge is
937/// unavailable, and the batches not yet asked fail alike, without a call. A pass whose lock was taken over
938/// asks nothing more; what it did not ask stays open for the lock's new holder, and `borg learn` settles
939/// no session whose lock another holds (learn.rs). A lock held by another (a stop, or an earlier pass
940/// of a resumed session) is waited for, as long as it is renewed: it reads stale `live::stale_ms`
941/// after its holder stops. Nothing runs, and nothing is written, without learning (`BORG_LEARN=off`),
942/// without a runner for the consequence model (`BORG_PREDICT_MODEL=off`, none installed), or with
943/// nothing open. Returns how many lessons it judged.
944pub fn judge_session(session: &str) -> usize {
945    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
946    if !learning() {
947        return 0;
948    }
949    let ctx = state_file(session);
950    let opened = |state: &str| state.lines().any(|l| DELIVERED.iter().any(|m| l.starts_with(m)));
951    if !opened(&std::fs::read_to_string(&ctx).unwrap_or_default()) {
952        return 0;
953    }
954    let model = std::env::var("BORG_PREDICT_MODEL").unwrap_or_default();
955    if crate::runner::choose(Some(&model)).is_empty() {
956        return 0;
957    }
958    // A stop, or an earlier pass of this session (a resumed session's end), judges the same lessons:
959    // wait for it while it holds the lock. A pass renews it before each call, so it may hold it longer
960    // than one wait; a holder that stops renewing lets it go stale within one.
961    let wait = u64::try_from(live::stale_ms()).unwrap_or(u64::MAX);
962    while live::wait_free(&ctx, wait) == live::Lock::Held {}
963    let Some(guard) = live::acquire(&ctx, learn::now_ms()) else { return 0 };
964    let guard = std::sync::Mutex::new(guard);
965    let state = std::fs::read_to_string(&ctx).unwrap_or_default();
966    let (all, withheld) = loaded(session);
967    let judged = judged(&all, withheld);
968    let last = crate::notes::proposal(&crate::notes::state_dir(), session);
969    let check = crate::select::check_id(session, "end", learn::now_ms(), std::process::id());
970    let judging = pending(session, &state, &judged, &last, When::End, &check);
971    let pend = &judging.pending;
972    if pend.is_empty() {
973        return 0;
974    }
975    let label = judge_model(&model);
976    // The store's judge cost lines are beside the session's: its .cost file is in the same directory.
977    let budget = end_budget(ctx.parent().unwrap_or(std::path::Path::new(".")), &label, &check);
978    let cut = Cut::from_env();
979    let sources = session_sources(session, &state, pend, budget.chars);
980    let groups = batches(pend, budget.chars);
981    let (next, failures, answers, lost) = (AtomicUsize::new(0), AtomicUsize::new(0), AtomicUsize::new(0), AtomicBool::new(false));
982    let repo = where_repo(session);
983    let n = groups.len();
984    let draw = budget.draw.map_or(String::new(), |(psi, explored)| format!(" psi={psi:.4} explore={}", u8::from(explored)));
985    // The judge's call numbering counts every answered model call of the session: the checks', then these
986    // in the order they answer (`batch=` names each one's place).
987    let mut call = state.lines().filter(|l| l.starts_with("#borg-predict-cost")).count();
988    let mut done = 0;
989    let mut asked = vec![false; n];
990    let (tx, rx) = std::sync::mpsc::channel::<(usize, Batch)>();
991    std::thread::scope(|s| {
992        for _ in 0..crate::runner::JOBS.min(n) {
993            let tx = tx.clone();
994            let (groups, next, failures, answers, lost, guard, sources, model) = (&groups, &next, &failures, &answers, &lost, &guard, &sources, &model);
995            s.spawn(move || {
996                while !lost.load(Ordering::SeqCst) && (answers.load(Ordering::SeqCst) > 0 || failures.load(Ordering::SeqCst) < crate::runner::MAX_FAILURES) {
997                    let i = next.fetch_add(1, Ordering::SeqCst);
998                    let Some(range) = groups.get(i) else { break };
999                    // Taken over: another holds the session now. Not written (I/O): still ours, ask on.
1000                    if guard.lock().map_or(true, |mut g| g.renew(learn::now_ms()) == live::Renewal::Lost) {
1001                        lost.store(true, Ordering::SeqCst);
1002                        break;
1003                    }
1004                    let b = ask_batch(&pend[range.clone()], sources, budget.chars, cut, model);
1005                    (if b.pred.is_some() { answers } else { failures }).fetch_add(1, Ordering::SeqCst);
1006                    if tx.send((i, b)).is_err() {
1007                        break;
1008                    }
1009                }
1010            });
1011        }
1012        drop(tx);
1013        // One writer: each batch records here as its call returns, while the others still run.
1014        for (i, b) in rx {
1015            asked[i] = true;
1016            let batch = &pend[groups[i].clone()];
1017            let ids: Vec<String> = batch.iter().map(|p| p.lesson.id.clone()).collect();
1018            crate::stats::count(&learn::home(), crate::stats::MODEL_CALL, session, |ev| {
1019                ev.why = "judge".into();
1020                ev.repo = repo.clone();
1021                ev.n = batch.len() as u64;
1022                ev.usd = b.pred.as_ref().map_or(0.0, |p| p.cost);
1023                ev.dur_ms = b.ms;
1024                if b.pred.is_none() {
1025                    ev.text = "no answer".into();
1026                }
1027            });
1028            let fields = format!(" batch={}/{n} budget={}{draw} cut={}", i + 1, budget.chars, if b.cut { cut.word() } else { "none" });
1029            let Some(pred) = b.pred else {
1030                // Sent and not answered: the judge's effective length counts it (judge_length.rs).
1031                log_cost(session, &format!("judge-failed pending={} chars={} model={label} at={}{fields}", batch.len(), b.chars, outcome::AT_END));
1032                note_unjudged(session, &ids, "failed");
1033                continue;
1034            };
1035            call += 1;
1036            record_verdicts(session, &repo, outcome::AT_END, batch, &pred, &Logged { call, chars: b.chars, model: label.clone(), fields });
1037            append(session, &format!("{END_JUDGED}{}", ids.join(",")));
1038            done += ids.len();
1039        }
1040    });
1041    // Not asked: the judge was unavailable (failed, no try), or the lock was lost (open for its holder).
1042    if !lost.load(Ordering::SeqCst) {
1043        let unasked: Vec<String> =
1044            groups.iter().zip(&asked).filter(|(_, a)| !**a).flat_map(|(r, _)| pend[r.clone()].iter().map(|p| p.lesson.id.clone())).collect();
1045        note_unjudged(session, &unasked, "failed");
1046    }
1047    done
1048}
1049
1050/// The session-end judge's batches of `pending` (`judge_session`, in delivery order), as ranges of it:
1051/// each takes the next lessons while their judge section (`predict::judge_section`) stays within
1052/// 1/`END_LESSONS_SHARE` of `budget`, and at least one. Consecutive, so each batch covers its own span of
1053/// the session, from its earliest delivery on. A lesson whose evidence alone exceeds that share goes
1054/// alone (its evidence is bounded by `predict::JUDGE_PART`).
1055fn batches(pending: &[predict::Pending], budget: usize) -> Vec<std::ops::Range<usize>> {
1056    let room = budget / END_LESSONS_SHARE;
1057    let mut out = Vec::new();
1058    let mut start = 0;
1059    for end in 2..=pending.len() {
1060        if end - start > 1 && predict::judge_section(&pending[start..end]).chars().count() > room {
1061            out.push(start..end - 1);
1062            start = end - 1;
1063        }
1064    }
1065    if start < pending.len() {
1066        out.push(start..pending.len());
1067    }
1068    out
1069}
1070
1071/// One session-end batch's call (`ask_batch`): its prompt's characters (`predict::judge_prompt`), whether
1072/// the session's work was cut to fit, the judge's answer (None: failed) and how long it took.
1073struct Batch {
1074    chars: usize,
1075    cut: bool,
1076    pred: Option<predict::Prediction>,
1077    ms: u64,
1078}
1079
1080/// Asks the session-end judge about `batch` (`judge_session`): beside its lessons, the session's work
1081/// in what `budget` leaves (`within`), cut by `cut` at the batch's earliest delivery, within
1082/// `end_timeout`.
1083fn ask_batch(batch: &[predict::Pending], sources: &[Source], budget: usize, cut: Cut, model: &str) -> Batch {
1084    let fixed = predict::judge_prompt("", batch).chars().count();
1085    let ids: Vec<&str> = batch.iter().map(|p| p.lesson.id.as_str()).collect();
1086    let anchors: Vec<Option<usize>> = sources.iter().map(|s| s.anchor(&ids)).collect();
1087    let (work, was_cut) = within(sources, &anchors, budget.saturating_sub(fixed), cut);
1088    let chars = fixed + work.chars().count();
1089    let started = std::time::Instant::now();
1090    let pred = predict::judge(&work, batch, model, end_timeout());
1091    Batch { chars, cut: was_cut, pred, ms: started.elapsed().as_millis() as u64 }
1092}
1093
1094/// One source of the session's work as the session-end judge reads it (`session_sources`): its label,
1095/// its text, and the delivery markers (`DELIVERED`) that stood in it: where (a character offset of the
1096/// text) and the lessons each delivered.
1097struct Source {
1098    label: &'static str,
1099    text: Vec<char>,
1100    marks: Vec<(usize, Vec<String>)>,
1101}
1102
1103impl Source {
1104    /// The lines of `text` that `keep` gives, each delivery marker a mark between them; each run of lines
1105    /// between two marks without the declarations of the stripped questions among `pending`
1106    /// (`without_declarations`). Empty when only whitespace is left.
1107    fn of<'t>(label: &'static str, text: &'t str, keep: impl Fn(&'t str) -> Option<&'t str>, pending: &[predict::Pending]) -> Source {
1108        let mut out = Source { label, text: Vec::new(), marks: Vec::new() };
1109        let mut run: Vec<&str> = Vec::new();
1110        for l in text.lines() {
1111            if let Some(list) = DELIVERED.iter().find_map(|m| l.strip_prefix(m)) {
1112                out.push(&run, pending);
1113                run.clear();
1114                // Where the next run starts: after the newline that will set it apart.
1115                out.marks.push((out.text.len() + usize::from(!out.text.is_empty()), ids_in(list).map(str::to_string).collect()));
1116            } else if let Some(kept) = keep(l) {
1117                run.push(kept);
1118            }
1119        }
1120        out.push(&run, pending);
1121        if out.text.iter().all(|c| c.is_whitespace()) {
1122            out.text.clear();
1123        }
1124        out
1125    }
1126
1127    /// Appends a run of `lines`, on lines of their own, without the stripped declarations.
1128    fn push(&mut self, lines: &[&str], pending: &[predict::Pending]) {
1129        if lines.is_empty() {
1130            return;
1131        }
1132        if !self.text.is_empty() {
1133            self.text.push('\n');
1134        }
1135        self.text.extend(without_declarations(&lines.join("\n"), pending).chars());
1136    }
1137
1138    /// Where the earliest of `ids`' latest deliveries stood in the text (the delivery each is judged on,
1139    /// `pending`); None when no marker names one.
1140    fn anchor(&self, ids: &[&str]) -> Option<usize> {
1141        ids.iter().filter_map(|id| self.marks.iter().rev().find(|(_, l)| l.iter().any(|x| x == id)).map(|m| m.0)).min()
1142    }
1143}
1144
1145/// The session's whole work as the session-end judge reads it (`judge_session`), by source: the
1146/// engineer's prompts and the agent's touches (the state file without its bookkeeping), the agent's
1147/// edits and its answers to questions (the .edits file), and its messages (its harness's transcript,
1148/// harness.rs `session_texts`, where it writes one: its newest `cap` characters, more than any batch
1149/// reads). Each without the declarations of the stripped questions among `pending`
1150/// (`without_declarations`); the first two with where each delivery stood in them (`Source`).
1151fn session_sources(session: &str, state: &str, pending: &[predict::Pending], cap: usize) -> Vec<Source> {
1152    let edits = std::fs::read_to_string(edits_file(session)).unwrap_or_default();
1153    let said = harness::session_texts(harness::current(), session, cap);
1154    vec![
1155        Source::of("The engineer's prompts and the agent's touches", state, |l| (!l.starts_with("#borg-")).then_some(l), pending),
1156        Source::of("The agent's edits and answers", &edits, |l| l.strip_prefix("#borg-answer ").or_else(|| (!l.starts_with("#borg-")).then_some(l)), pending),
1157        Source { label: "The agent's messages", text: without_declarations(&said, pending).chars().collect(), marks: Vec::new() },
1158    ]
1159}
1160
1161/// `sources` in `budget` characters in all, framing included, each as `### <label>` and its text: all
1162/// whole when they fit; else each an equal share of what is left, the shortest first, so a part shorter
1163/// than its share leaves the rest to the others, and a longer one is cut by `cut` (`Cut::window`) at its
1164/// `anchors` entry (the batch's earliest delivery in it). An empty part is left out. With whether a part
1165/// was cut.
1166fn within(sources: &[Source], anchors: &[Option<usize>], budget: usize, cut: Cut) -> (String, bool) {
1167    let len: Vec<usize> = sources.iter().map(|s| s.text.len()).collect();
1168    let mut order: Vec<usize> = (0..sources.len()).filter(|i| len[*i] > 0).collect();
1169    // `### <label>\n` and the closing newline of each part; a cut part's marks (`…`), two at most.
1170    let frame: usize = order.iter().map(|&i| sources[i].label.chars().count() + 6).sum();
1171    let total: usize = len.iter().sum();
1172    let mut left = if frame + total <= budget { total } else { budget.saturating_sub(frame + 2 * order.len()) };
1173    order.sort_by_key(|i| len[*i]);
1174    let mut keep = vec![0; sources.len()];
1175    for (k, &i) in order.iter().enumerate() {
1176        keep[i] = len[i].min(left / (order.len() - k));
1177        left -= keep[i];
1178    }
1179    let (mut out, mut was_cut) = (String::new(), false);
1180    for (i, s) in sources.iter().enumerate() {
1181        let n = keep[i];
1182        if n == 0 {
1183            continue;
1184        }
1185        let text = if n < len[i] {
1186            was_cut = true;
1187            cut.window(&s.text, n, anchors.get(i).copied().flatten())
1188        } else {
1189            s.text.iter().collect()
1190        };
1191        out.push_str(&format!("### {}\n{text}\n", s.label));
1192    }
1193    (out, was_cut)
1194}
1195
1196/// The repo the session works in: the latest repo the hook noted at a prompt (`.where`, a repo name on
1197/// a line of its own, systems.rs `work_repos`); empty when none.
1198fn where_repo(session: &str) -> String {
1199    let text = std::fs::read_to_string(where_file(session)).unwrap_or_default();
1200    text.lines().map(str::trim).rfind(|l| !l.is_empty() && !l.contains(['/', ' '])).unwrap_or_default().to_string()
1201}
1202
1203/// Ids listed on the state file's `marker` lines (`#borg-asked a,b @<check>`).
1204pub(crate) fn listed<'a>(context: &'a str, marker: &str) -> std::collections::HashSet<&'a str> {
1205    context.lines().filter_map(|l| l.strip_prefix(marker)).flat_map(ids_in).collect()
1206}
1207
1208/// The ids of a marker's or a queue line's list, without its ` @<check>` tag (`tagged`).
1209fn ids_in(list: &str) -> impl Iterator<Item = &str> {
1210    tagged(list).0.split(',').map(str::trim).filter(|i| !i.is_empty())
1211}
1212
1213/// ` @<check>` after a list, or nothing for a check without an id (a line of an older binary's state).
1214fn tag(check: &str) -> String {
1215    if check.is_empty() { String::new() } else { format!(" @{check}") }
1216}
1217
1218/// Early delivery (conscience.rs `early`): once the session's prompts and touches name a lesson scope
1219/// (a topic) and that scope's lessons score on them, the top ones go to the drone at once, as context,
1220/// instead of waiting for the Stop checkpoint: at the prompt or the first touch on the topic (`text` is
1221/// this event's, already in the state file). Each lesson once per session (`#borg-early`; Stop does not
1222/// ask it again) and an `asked` outcome with channel early (`record_delivery`). A topic is settled (`#borg-topic`) once a delivery covered it or none of its lessons is
1223/// left; until then each prompt or touch looks again. At most `MAX_EARLY_DELIVERIES` per session. Only
1224/// where the harness takes it (`Caps::early_delivery`, Claude Code so far): omp and opencode take a prompt's
1225/// and a tool call's context too, but early delivery to them waits for an evaluation, so for
1226/// them the lessons stay with Stop. Standing lessons (ratified constraints, conscience.rs `standing_pool`)
1227/// are no ranking guess and need no evaluation: they come on every harness that takes the context (a
1228/// prompt's, or a tool call's for a named scope), whenever their scope is named (an unscoped one at a
1229/// prompt), settled topic or not and past `MAX_EARLY_DELIVERIES`, until each came once. `BORG_EARLY=off`
1230/// disables it; so do the eval modes BORG_ONLY and BORG_CRITIC.
1231/// Early deliveries per session at most (each up to `conscience::render_early`'s lesson cap). Guard: a
1232/// long session naming many systems would otherwise get a delivery at every touch until each topic's
1233/// lessons ran out (standing lessons are exempt).
1234const MAX_EARLY_DELIVERIES: usize = 3;
1235
1236fn early(h: &dyn Harness, so_far: &SoFar, hook_event: &str, session: &str, input: &Value, text: &str) -> Option<String> {
1237    if text.trim().is_empty()
1238        || std::env::var("BORG_EARLY").is_ok_and(|v| v.trim().eq_ignore_ascii_case("off"))
1239        || std::env::var_os("BORG_ONLY").is_some()
1240        || std::env::var_os("BORG_CRITIC").is_some()
1241    {
1242        return None;
1243    }
1244    let challenges = lessons(session);
1245    let context = std::fs::read_to_string(state_file(session)).unwrap_or_default();
1246    // Free, but not free of noise: a long session naming many systems would otherwise get a delivery
1247    // at every touch until each topic's lessons ran out. Standing lessons are exempt.
1248    let plain = h.caps().early_delivery && context.lines().filter(|l| l.starts_with("#borg-early ")).count() < MAX_EARLY_DELIVERIES;
1249    let standing = challenges.iter().any(conscience::Challenge::standing);
1250    if !standing && (!plain || challenges.iter().all(|c| c.scope.is_empty() && !c.consolidated)) {
1251        return None;
1252    }
1253    let settled = listed(&context, "#borg-topic ");
1254    let named = conscience::named_scopes(&challenges, &where_now(session));
1255    let mut topics: Vec<String> = if plain { named.iter().filter(|t| !settled.contains(t.as_str())).cloned().collect() } else { Vec::new() };
1256    // Consolidated lessons (memory.rs) need no scope: the first prompt is their topic (`ANY`).
1257    if plain
1258        && topics.is_empty()
1259        && hook_event == "UserPromptSubmit"
1260        && !settled.contains(conscience::ANY)
1261        && challenges.iter().any(|c| c.consolidated && !c.standing())
1262    {
1263        topics.push(conscience::ANY.into());
1264    }
1265    // A standing lesson's topic is every named scope, settled or not, and any prompt for an unscoped one.
1266    let mut standing_topics = named;
1267    if hook_event == "UserPromptSubmit" {
1268        standing_topics.push(conscience::ANY.into());
1269    }
1270    let done = had_in(&context);
1271    let mut pool = if standing { conscience::standing_pool(&challenges, &standing_topics, &done) } else { Vec::new() };
1272    if !topics.is_empty() {
1273        pool.extend(conscience::early_pool(&challenges, &conscience::work_text(&context), &topics, &done));
1274    }
1275    // The header names what came: the plain topics, and the scopes of the standing lessons beside them.
1276    let mut shown_topics = topics.clone();
1277    for t in standing_topics.into_iter().filter(|t| pool.iter().any(|(_, c)| c.standing() && c.scope.contains(t))) {
1278        if !shown_topics.contains(&t) {
1279            shown_topics.push(t);
1280        }
1281    }
1282    // Today's early caps (`EARLY_CAP`, `EARLY_CHARS`, `RELATIVE`), kept until a learned early rule has
1283    // data: each early delivery's record carries every lesson's score relative to the best (`early_rel`).
1284    let selected: Vec<&conscience::Challenge> = conscience::explored(pool.iter().copied()).into_iter().take(conscience::EARLY_CAP).collect();
1285    let shown = conscience::render_early(&selected, &shown_topics);
1286    let delivered = &selected[..shown.as_ref().map_or(0, |(_, n)| *n)];
1287    // Plain topics only: a standing lesson left over (past `EARLY_CAP` or `EARLY_CHARS`) comes at the next
1288    // prompt or touch on its scope whatever is settled.
1289    let now_settled: Vec<&str> = topics
1290        .iter()
1291        .filter(|t| {
1292            // One delivery per topic: whatever was delivered now covers every topic looked at now.
1293            !delivered.is_empty()
1294                || !challenges.iter().any(|c| {
1295                    !c.standing()
1296                        && (c.scope.contains(t) || (t.as_str() == conscience::ANY && c.consolidated))
1297                        && !done.contains(c.id.as_str())
1298                        && !delivered.iter().any(|d| d.id == c.id)
1299                })
1300        })
1301        .map(String::as_str)
1302        .collect();
1303    if !now_settled.is_empty() {
1304        append(session, &format!("#borg-topic {}", now_settled.join(",")));
1305    }
1306    let (text, _) = shown?;
1307    let ids = delivered.iter().map(|c| c.id.as_str()).collect::<Vec<_>>().join(",");
1308    // An early delivery is its own decision: its id ties its markers to its record.
1309    let check = crate::select::check_id(session, "early", learn::now_ms(), std::process::id());
1310    let line = format!("#borg-early {ids} @{check}");
1311    append(session, &line);
1312    note_work(session, so_far, &line);
1313    let p = conscience::early_propensity(&pool, session, &shown_topics);
1314    let p: Vec<f64> = delivered.iter().map(|c| p.get(&c.id).copied().unwrap_or(0.0)).collect();
1315    record_delivery(session, input, delivered, "early", &check, p, early_rel(&challenges, &conscience::work_text(&context), &topics, delivered));
1316    log_cost(session, &format!("free event={hook_event} lessons={} chars={}", delivered.len(), text.len()));
1317    crate::stats::count(&learn::home(), crate::stats::EARLY, session, |e| {
1318        e.n = delivered.len() as u64;
1319        e.lessons = delivered.iter().map(|c| c.id.clone()).collect();
1320        e.channel = "early".into();
1321    });
1322    Some(text)
1323}
1324
1325/// Each of `delivered`'s word-match score (conscience.rs `rank` over `work`) as a share of the best
1326/// on-topic match of `topics`, done or not (the match `RELATIVE` bars against: `early_pool` over no done
1327/// lesson starts with it), rounded as a propensity is: what a learned early rule needs of each early
1328/// delivery (outcome.rs `rel`). A standing lesson's too, though no score gates it (0 when it matches no
1329/// word). Empty when no plain lesson is on topic to scale by (a standing lesson's delivery alone).
1330fn early_rel(challenges: &[conscience::Challenge], work: &str, topics: &[String], delivered: &[&conscience::Challenge]) -> Vec<f64> {
1331    let Some(&(top, _)) = conscience::early_pool(challenges, work, topics, &std::collections::HashSet::new()).first() else { return Vec::new() };
1332    let ranked = conscience::rank(challenges, work);
1333    delivered.iter().map(|c| outcome::round_p(ranked.iter().find(|(_, x)| x.id == c.id).map_or(0.0, |(s, _)| *s) / top)).collect()
1334}
1335
1336/// A delivery's markers (`lines`: a free delivery's, a placebo's) in the .edits file, in
1337/// one write after the work so far (`WORK`): the evidence before it is the same whatever the route
1338/// (`pending`), and no route's line reads as work after an earlier delivery.
1339fn note_work(session: &str, so_far: &SoFar, lines: &str) {
1340    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(edits_file(session)) {
1341        let _ = writeln!(f, "{WORK}{}\n{lines}", one_line(so_far.get(), EVIDENCE_CHARS));
1342    }
1343}
1344
1345/// The `asked` outcome of a free delivery (`channel` early or queued) of `delivered`, routed by `check`,
1346/// with each one's propensity `p` and, early, its relative score `rel` (`early_rel`): delivered as
1347/// context, the lessons wait for a result as a Stop's questions do. The next judge (`pending`) labels
1348/// them changed or ignored, a merged correction corrected (outcome.rs `contradicted`), and `borg learn`
1349/// settles the rest. Nothing with learning off.
1350fn record_delivery(session: &str, input: &Value, delivered: &[&conscience::Challenge], channel: &str, check: &str, p: Vec<f64>, rel: Vec<f64>) {
1351    if !learning() || delivered.is_empty() {
1352        return;
1353    }
1354    let lessons: Vec<String> = delivered.iter().map(|c| c.id.clone()).collect();
1355    let (grades, ranks, applied) = graded_of(session, &lessons);
1356    let asked = outcome::Outcome {
1357        kind: outcome::ASKED.into(),
1358        session: session.into(),
1359        lessons,
1360        ms: learn::now_ms(),
1361        repo: repo_of(input.get("cwd").and_then(Value::as_str).unwrap_or_default()),
1362        groups: conscience::disagreement_ids(delivered),
1363        channel: channel.into(),
1364        grades,
1365        ranks,
1366        applied,
1367        p: p.into_iter().map(outcome::round_p).collect(),
1368        check: check.into(),
1369        rel,
1370        ..Default::default()
1371    };
1372    let _ = outcome::append(&learn::home().join("outcomes.jsonl"), &asked);
1373}
1374
1375/// One `delivery` record (outcome.rs `DELIVERY`): what delivery did with `lessons`, routed by `check`:
1376/// `route` (outcome.rs `ROUTES`) and why. Nothing with learning off or without lessons.
1377fn record_route(session: &str, input: &Value, check: &str, lessons: Vec<String>, channel: &str, route: &str, reason: &str) {
1378    if !learning() || lessons.is_empty() {
1379        return;
1380    }
1381    let o = outcome::Outcome {
1382        kind: outcome::DELIVERY.into(),
1383        session: session.into(),
1384        lessons,
1385        ms: learn::now_ms(),
1386        repo: repo_of(input.get("cwd").and_then(Value::as_str).unwrap_or_default()),
1387        channel: channel.into(),
1388        check: check.into(),
1389        route: route.into(),
1390        reason: reason.into(),
1391        ..Default::default()
1392    };
1393    let _ = outcome::append(&learn::home().join("outcomes.jsonl"), &o);
1394}
1395
1396/// `record_route` for each check of `routed` (lesson, check) pairs, in first-seen order.
1397fn record_routes(session: &str, input: &Value, routed: &[(&conscience::Challenge, &str)], channel: &str, route: &str, reason: &str) {
1398    for (check, ids) in by_check(routed) {
1399        record_route(session, input, check, ids.into_iter().map(String::from).collect(), channel, route, reason);
1400    }
1401}
1402
1403/// `routed` (lesson, check) pairs grouped by check, checks in first-seen order, lessons in theirs.
1404fn by_check<'a>(routed: &[(&'a conscience::Challenge, &'a str)]) -> Vec<(&'a str, Vec<&'a str>)> {
1405    let mut out: Vec<(&str, Vec<&str>)> = Vec::new();
1406    for (c, check) in routed {
1407        match out.iter_mut().find(|(k, _)| k == check) {
1408            Some((_, ids)) => ids.push(c.id.as_str()),
1409            None => out.push((check, vec![c.id.as_str()])),
1410        }
1411    }
1412    out
1413}
1414
1415/// Graded Stop (default): what a graded check left for later, one line per check, waiting for the session's
1416/// next prompt or touch (`queued`): `<ids> @<check>`, the 2s and 1s of a check that did not block, to hand
1417/// over free; `placebo <ids> @<check>`, a check's placebo draws (select.rs `Routed`); and `instead-next
1418/// <ids> @<check>`, a check's free lessons handed over in place of questions the fixed rule asks
1419/// (`INSTEAD_NEXT`); and `live <ids> @<check>`, a detached check's asks (`ASK_NEXT`, check_async.rs).
1420/// A line of an older binary has no tag, and may be a live check's (`ASK_NEXT`, `LEGACY_INSTEAD`).
1421fn queued_file(session: &str) -> PathBuf {
1422    state_file(session).with_extension("queued")
1423}
1424
1425/// What a queue line routes its lessons to (`queued_file`), in the order delivery takes them (`queued`).
1426#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
1427enum Queued {
1428    /// A detached check's ask, or a live check's from before 0.7.6 (`ASK_NEXT`): handed over, recorded as
1429    /// the ask it was.
1430    Ask,
1431    /// Free, in place of a question the fixed rule asks (`INSTEAD_NEXT`, `LEGACY_INSTEAD`).
1432    Instead,
1433    Free,
1434    Placebo,
1435}
1436
1437/// Protocol: the prefix of the free lessons a check hands over in place of questions the fixed rule
1438/// asks (`queued_file`): the evidence rule's on a check with a 3 (select/policy.rs `EVIDENCE`), which
1439/// valued them as hand-overs (their chance of arriving included), and any of a stop's. A free delivery
1440/// carries them whatever its budget (`queued`, conscience.rs `render_owing`), as the questions they
1441/// replace would have been carried; no stop asks them.
1442const INSTEAD_NEXT: &str = "instead-next ";
1443
1444/// Protocol: the prefixes of the asks a check queues for the next prompt or tool call: a detached
1445/// check's (check_async.rs, `live <ids> @<check>`, since 0.7.7), and a live check's from before 0.7.6
1446/// (live mode is gone; a session that ran across that upgrade may still have one waiting). The next
1447/// prompt or tool call hands them over, all and first, and records each as the ask its check decided
1448/// (channel live, `#borg-asked` then `#borg-live`, as 0.7.5 did), so the selection fold and the judge
1449/// join it; no stop blocks for them. A detached check's come as questions about the turn it judged
1450/// (`PROPOSAL`, conscience.rs `render_late`); a live check's with the free lessons, as 0.7.5 did.
1451const ASK_NEXT: [&str; 2] = ["live ", "live-next "];
1452
1453/// Protocol: the prefix of a live check's question-count test hand-overs in place of questions, from
1454/// before 0.7.6: free, as an `INSTEAD_NEXT` line is.
1455const LEGACY_INSTEAD: &str = "instead ";
1456
1457/// A queue line's route, ids and check (empty on an older binary's line).
1458fn queue_line(line: &str) -> (Queued, Vec<&str>, &str) {
1459    let ask = ASK_NEXT.iter().find_map(|p| line.strip_prefix(p));
1460    let instead = line.strip_prefix(INSTEAD_NEXT).or_else(|| line.strip_prefix(LEGACY_INSTEAD));
1461    let (route, list) = match (ask, line.strip_prefix("placebo "), instead) {
1462        (Some(l), _, _) => (Queued::Ask, l),
1463        (_, Some(l), _) => (Queued::Placebo, l),
1464        (_, _, Some(l)) => (Queued::Instead, l),
1465        _ => (Queued::Free, line),
1466    };
1467    (route, ids_in(list).collect(), tagged(list).1.unwrap_or_default())
1468}
1469
1470/// `lines` at the end of the `.queued` file in one write under its lock (live.rs `locked`): never into a
1471/// file a hook is taking at the same moment. A detached check's (`offset`, its stop's state length) only
1472/// while its session lives (`gone`), read under that lock, which the end hook reads the queue under
1473/// (`note_never_delivered`): false when it ended, nothing written then.
1474fn enqueue_all(session: &str, lines: &[String], offset: Option<usize>) -> bool {
1475    if lines.is_empty() {
1476        return true;
1477    }
1478    let q = queued_file(session);
1479    live::locked(&q, || {
1480        if offset.is_some_and(|o| gone(session, o)) {
1481            return false;
1482        }
1483        let text: String = lines.iter().map(|l| format!("{l}\n")).collect();
1484        let _ = std::fs::OpenOptions::new().create(true).append(true).open(&q).and_then(|mut f| f.write_all(text.as_bytes()));
1485        true
1486    })
1487}
1488
1489/// The whole `.queued` file, taken under its lock (a rename, then read): what a check appends after
1490/// starts a new file, delivered next time.
1491fn take_queue(session: &str) -> Option<String> {
1492    let q = queued_file(session);
1493    let taken = state_file(session).with_extension(format!("queued.{}", std::process::id()));
1494    live::locked(&q, || std::fs::rename(&q, &taken)).ok()?;
1495    let text = std::fs::read_to_string(&taken).unwrap_or_default();
1496    let _ = std::fs::remove_file(&taken);
1497    Some(text)
1498}
1499
1500/// The lessons a `.queued` file's `text` holds to be handed over (every line but a placebo's): decided
1501/// already by the check that queued them, so no other check decides them again (the stop's pool,
1502/// `decide`'s `had`). A placebo waiting was routed nowhere: it stays eligible. For a detached check
1503/// (`regraded`) so does a waiting ask (`ASK_NEXT`): it grades it again against newer work and takes its
1504/// line off the queue (`supersede`), so a question never reaches the drone about work it has since
1505/// moved past. A stop's own check leaves it to its line, as before 0.7.7.
1506fn waiting_in(text: &str, regraded: bool) -> impl Iterator<Item = &str> {
1507    text.lines().map(queue_line).filter(move |(route, _, _)| *route != Queued::Placebo && !(regraded && *route == Queued::Ask)).flat_map(|(_, ids, _)| ids)
1508}
1509
1510/// A detached check's step before it queues its own (`settle`): the waiting asks of earlier checks
1511/// (`ASK_NEXT` lines) that it showed its model again (`shown`) leave the queue: superseded when it asks
1512/// them too (its own line carries them; outcome.rs `ROUTE_SUPERSEDED`), else settled by the later work
1513/// it graded (`ROUTE_DROPPED_DONE`); each with its `delivery` record. The `.queued` file is rewritten whole under its
1514/// lock (live.rs `locked`), so no hook takes it half-written. The ids that left.
1515fn supersede(session: &str, input: &Value, shown: &[&conscience::Challenge], asks: &[&conscience::Challenge]) -> Vec<String> {
1516    let q = queued_file(session);
1517    // (check, ids, why) of each waiting ask taken off.
1518    let mut left: Vec<(String, Vec<String>, &'static str, &'static str)> = Vec::new();
1519    live::locked(&q, || {
1520        let Ok(text) = std::fs::read_to_string(&q) else { return };
1521        let mut kept = String::new();
1522        for line in text.lines() {
1523            let (route, ids, check) = queue_line(line);
1524            let again: Vec<&str> = ids.iter().copied().filter(|i| route == Queued::Ask && shown.iter().any(|c| c.id == *i)).collect();
1525            if again.is_empty() {
1526                kept.push_str(&format!("{line}\n"));
1527                continue;
1528            }
1529            let (asked, settled): (Vec<&str>, Vec<&str>) = again.iter().copied().partition(|i| asks.iter().any(|c| c.id == *i));
1530            for (ids, route, why) in [(asked, outcome::ROUTE_SUPERSEDED, "superseded"), (settled, outcome::ROUTE_DROPPED_DONE, "settled by later work")] {
1531                if !ids.is_empty() {
1532                    left.push((check.to_string(), ids.into_iter().map(String::from).collect(), route, why));
1533                }
1534            }
1535            let rest: Vec<&str> = ids.into_iter().filter(|i| !again.contains(i)).collect();
1536            if let Some(prefix) = ASK_NEXT.iter().find(|p| line.starts_with(**p))
1537                && !rest.is_empty()
1538            {
1539                kept.push_str(&format!("{prefix}{}{}\n", rest.join(","), tag(check)));
1540            }
1541        }
1542        if kept.is_empty() {
1543            let _ = std::fs::remove_file(&q);
1544            return;
1545        }
1546        let tmp = q.with_extension(format!("queued.{}.new", std::process::id()));
1547        if std::fs::write(&tmp, &kept).and_then(|()| std::fs::rename(&tmp, &q)).is_err() {
1548            let _ = std::fs::remove_file(&tmp);
1549            left.clear();
1550        }
1551    });
1552    let mut out = Vec::new();
1553    for (check, ids, route, why) in left {
1554        out.extend(ids.iter().cloned());
1555        record_route(session, input, &check, ids, "live", route, why);
1556    }
1557    out
1558}
1559
1560/// Protocol: the .edits line a detached check writes beside its asks (`settle`), `#borg-proposal
1561/// <check> <the drone's message it judged, on one line>`: what their delivery (`queued`) names as the
1562/// turn the questions are about (conscience.rs `render_late`) and gives the judge as the work before them
1563/// (`pending`), as a block's blocked message is. A bookkeeping line: no reader takes it for work.
1564const PROPOSAL: &str = "#borg-proposal ";
1565
1566/// The excerpt of the judged message a detached check's questions name (conscience.rs `render_late`), in
1567/// characters. Display: enough of a message's start to recognise the turn.
1568const LATE_EXCERPT: usize = 160;
1569
1570fn note_proposal(session: &str, check: &str, proposal: &str) {
1571    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(edits_file(session)) {
1572        let _ = writeln!(f, "{PROPOSAL}{check} {}", one_line(proposal, EVIDENCE_CHARS));
1573    }
1574}
1575
1576/// The message `check` judged, from its `PROPOSAL` line in the .edits file's `edits`; None for a check
1577/// that wrote none (a live check's ask from before 0.7.6).
1578fn proposal_of<'e>(edits: &'e str, check: &str) -> Option<&'e str> {
1579    if check.is_empty() {
1580        return None;
1581    }
1582    edits.lines().rev().find_map(|l| l.strip_prefix(PROPOSAL)?.strip_prefix(check)?.strip_prefix(' '))
1583}
1584
1585/// A detached check's asks of `check` delivered (`members`, the drone's message it judged `said`), as
1586/// `queued` and a wake (listen.rs, `commit_asks`) mark them: `#borg-asked` then `#borg-live` (with `via`
1587/// when a wake carried them) in the state file, the pair after the judged message in the .edits file,
1588/// the work before the questions (`pending`), so the next stop records the drone's answer
1589/// (`note_answer`); each recorded as asked on channel live (`record_asked`); the cost line, `event` the
1590/// hook's or `wake`; and the stat.
1591fn mark_late(session: &str, input: &Value, check: &str, members: &[&conscience::Challenge], said: &str, event: &str, chars: usize) {
1592    let list = members.iter().map(|c| c.id.as_str()).collect::<Vec<_>>().join(",");
1593    let via = if event == WAKE_EVENT { format!(" via={WAKE_EVENT}") } else { String::new() };
1594    let marker = format!("#borg-asked {list}{}\n#borg-live {list} ms={}{via}", tag(check), learn::now_ms());
1595    append(session, &marker);
1596    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(edits_file(session)) {
1597        let _ = writeln!(f, "{said}\n{marker}");
1598    }
1599    record_asked(session, input, members, "live", check);
1600    log_cost(session, &format!("late event={event} questions={} chars={chars}", members.len()));
1601    crate::stats::count(&learn::home(), crate::stats::LIVE_ASK, session, |e| {
1602        e.n = members.len() as u64;
1603        e.lessons = members.iter().map(|c| c.id.clone()).collect();
1604        e.channel = "live".into();
1605        e.why = if event == WAKE_EVENT { WAKE_EVENT.into() } else { String::new() };
1606    });
1607}
1608
1609/// Protocol: the event a wake's delivery of asks is logged under (`mark_late`: the cost line's `event=`,
1610/// `#borg-live … via=wake`, the stat's why).
1611const WAKE_EVENT: &str = "wake";
1612
1613/// Protocol: the id a wake and an answer's `shown_ids` name a detached check's ask by,
1614/// `L@<check>:<lesson>` (listen.rs): beside message ids (msg.rs `is_id`), never one.
1615#[must_use]
1616pub(crate) fn lesson_wake_id(check: &str, lesson: &str) -> String {
1617    format!("L@{check}:{lesson}")
1618}
1619
1620/// The check and the lesson of a wake id (`lesson_wake_id`); None for any other id.
1621#[must_use]
1622pub(crate) fn lesson_of_wake_id(id: &str) -> Option<(&str, &str)> {
1623    id.strip_prefix("L@")?.split_once(':').filter(|(check, lesson)| !check.is_empty() && !lesson.is_empty())
1624}
1625
1626/// The asks a wake has taken from the `.queued` file and not yet delivered (`reserve_asks`):
1627/// `<s>.queued.wake`, `ASK_NEXT` lines, under the `.queued` file's lock. Delivered by the wake
1628/// (`commit_asks`), or put back (`release_asks`) when the wake never went in.
1629fn reserved_file(session: &str) -> PathBuf {
1630    state_file(session).with_extension("queued.wake")
1631}
1632
1633/// The detached checks' asks waiting in the `.queued` file (`ASK_NEXT` lines whose check left its judged
1634/// message, `PROPOSAL`), as `queued` would deliver them: each check with its lesson ids, in queue order.
1635/// A wake carries them (listen.rs).
1636#[must_use]
1637pub(crate) fn asks_waiting(session: &str) -> Vec<(String, Vec<String>)> {
1638    let text = std::fs::read_to_string(queued_file(session)).unwrap_or_default();
1639    if !text.lines().any(|l| queue_line(l).0 == Queued::Ask) {
1640        return Vec::new();
1641    }
1642    let edits = std::fs::read_to_string(edits_file(session)).unwrap_or_default();
1643    let mut out: Vec<(String, Vec<String>)> = Vec::new();
1644    for (route, ids, check) in text.lines().map(queue_line) {
1645        if route != Queued::Ask || proposal_of(&edits, check).is_none() {
1646            continue;
1647        }
1648        match out.iter_mut().find(|(c, _)| c == check) {
1649            Some((_, have)) => {
1650                for id in ids {
1651                    if !have.iter().any(|h| h == id) {
1652                        have.push(id.to_string());
1653                    }
1654                }
1655            }
1656            None => out.push((check.to_string(), ids.into_iter().map(String::from).collect())),
1657        }
1658    }
1659    out
1660}
1661
1662/// A wake's asks, taken off the `.queued` file into the reserved file (`reserved_file`) under the
1663/// queue's lock, so no prompt delivers them meanwhile (`queued`); with the text the wake carries
1664/// (conscience.rs `render_late` per check, as `queued` renders them) and their wake ids
1665/// (`lesson_wake_id`). Only the asks of `wanted` still waiting, deliverable and not done
1666/// (`done_in`, as `queued` drops them) are taken, each lesson once: the rest stay for the next prompt's
1667/// delivery and its records. None when none is taken, and while a detached check of the session runs
1668/// (check_async.rs `running`): it may grade a waiting ask again against newer work (`supersede`), so a
1669/// wake never carries a question that work settled; the waiter looks again once that check writes
1670/// `<s>.checked`.
1671#[must_use]
1672pub(crate) fn reserve_asks(session: &str, wanted: &[(String, Vec<String>)]) -> Option<(String, Vec<String>)> {
1673    if crate::check_async::running(session) {
1674        return None;
1675    }
1676    let all = lessons(session);
1677    let state = std::fs::read_to_string(state_file(session)).unwrap_or_default();
1678    let done = done_in(&state, &early_acted(session, &all, "", None));
1679    let ok =
1680        |check: &str, id: &str| wanted.iter().any(|(c, ids)| c == check && ids.iter().any(|i| i == id)) && !done.contains(id) && all.iter().any(|c| c.id == id);
1681    let q = queued_file(session);
1682    let mut taken: Vec<(String, Vec<String>)> = Vec::new();
1683    live::locked(&q, || {
1684        let Ok(text) = std::fs::read_to_string(&q) else { return };
1685        let mut kept = String::new();
1686        let mut moved = String::new();
1687        // A lesson waiting under two checks' lines goes once (the first); the other line keeps it, and
1688        // the prompt drops it there as delivered (`queued`).
1689        let mut once: std::collections::HashSet<&str> = std::collections::HashSet::new();
1690        for line in text.lines() {
1691            let (route, ids, check) = queue_line(line);
1692            let (mine, rest): (Vec<&str>, Vec<&str>) = ids.iter().copied().partition(|i| route == Queued::Ask && ok(check, i) && once.insert(*i));
1693            let Some(prefix) = ASK_NEXT.iter().find(|p| line.starts_with(**p)).filter(|_| !mine.is_empty()) else {
1694                kept.push_str(&format!("{line}\n"));
1695                continue;
1696            };
1697            moved.push_str(&format!("{prefix}{}{}\n", mine.join(","), tag(check)));
1698            if !rest.is_empty() {
1699                kept.push_str(&format!("{prefix}{}{}\n", rest.join(","), tag(check)));
1700            }
1701            taken.push((check.to_string(), mine.into_iter().map(String::from).collect()));
1702        }
1703        if taken.is_empty() {
1704            return;
1705        }
1706        let r = reserved_file(session);
1707        let wrote = std::fs::OpenOptions::new().create(true).append(true).open(&r).and_then(|mut f| f.write_all(moved.as_bytes()));
1708        let tmp = q.with_extension(format!("queued.{}.new", std::process::id()));
1709        let swapped =
1710            wrote.and_then(|()| if kept.is_empty() { std::fs::remove_file(&q) } else { std::fs::write(&tmp, &kept).and_then(|()| std::fs::rename(&tmp, &q)) });
1711        if swapped.is_err() {
1712            // The queue stays whole: the reserved copy goes, nothing is taken.
1713            let _ = std::fs::remove_file(&tmp);
1714            let _ = unreserve(&r, &moved);
1715            taken.clear();
1716        }
1717    });
1718    if taken.is_empty() {
1719        return None;
1720    }
1721    let edits = std::fs::read_to_string(edits_file(session)).unwrap_or_default();
1722    let mut parts = Vec::new();
1723    let mut wake = Vec::new();
1724    for (check, ids) in &taken {
1725        let members: Vec<&conscience::Challenge> = ids.iter().filter_map(|i| all.iter().find(|c| &c.id == i)).collect();
1726        parts.push(conscience::render_late(&members, &one_line(proposal_of(&edits, check).unwrap_or_default(), LATE_EXCERPT)));
1727        wake.extend(ids.iter().map(|id| lesson_wake_id(check, id)));
1728    }
1729    Some((parts.join("\n\n"), wake))
1730}
1731
1732/// `moved` (lines of the reserved file `r`) out of it again: the rest of `r` stays.
1733fn unreserve(r: &Path, moved: &str) -> std::io::Result<()> {
1734    let text = std::fs::read_to_string(r).unwrap_or_default();
1735    let rest = text.strip_suffix(moved).unwrap_or(&text);
1736    if rest.is_empty() { std::fs::remove_file(r) } else { std::fs::write(r, rest) }
1737}
1738
1739/// Takes the reserved asks the wake ids `ids` name (`lesson_wake_id`; None: all) off the reserved file,
1740/// under the queue's lock: each check's ids taken, in reserve order. `then` carries them out under the
1741/// same lock, before they leave the reserved file: a reader without the lock (`askable`, `decide`'s
1742/// `had`) finds each ask reserved, delivered or queued again, never in none of them. `then` must not
1743/// take the queue's lock (an flock held twice by one process waits on itself). When `then` fails (false:
1744/// a full or read-only state directory) the asks stay reserved, none taken: a release that wrote nothing
1745/// never loses them (reclaimed later, or recorded `never_delivered` at the session's end).
1746fn take_reserved(session: &str, ids: Option<&[String]>, then: impl FnOnce(&[(String, Vec<String>)]) -> bool) -> Vec<(String, Vec<String>)> {
1747    let q = queued_file(session);
1748    let r = reserved_file(session);
1749    let named = |check: &str, id: &str| ids.is_none_or(|ids| ids.iter().any(|w| lesson_of_wake_id(w) == Some((check, id))));
1750    let mut taken: Vec<(String, Vec<String>)> = Vec::new();
1751    live::locked(&q, || {
1752        let Ok(text) = std::fs::read_to_string(&r) else { return };
1753        let mut kept = String::new();
1754        for line in text.lines() {
1755            let (_, ids, check) = queue_line(line);
1756            let (mine, rest): (Vec<&str>, Vec<&str>) = ids.iter().copied().partition(|i| named(check, i));
1757            if !rest.is_empty() {
1758                kept.push_str(&format!("{}{}{}\n", ASK_NEXT[0], rest.join(","), tag(check)));
1759            }
1760            if !mine.is_empty() {
1761                taken.push((check.to_string(), mine.into_iter().map(String::from).collect()));
1762            }
1763        }
1764        if taken.is_empty() {
1765            return;
1766        }
1767        if !then(&taken) {
1768            taken.clear();
1769            return;
1770        }
1771        let _ = if kept.is_empty() { std::fs::remove_file(&r) } else { std::fs::write(&r, &kept) };
1772    });
1773    taken
1774}
1775
1776/// A wake that went in (listen.rs) delivered the reserved asks `ids` names (`lesson_wake_id`): each
1777/// check's marked and recorded as `queued` marks a delivered one (`mark_late`, event `wake`; `cwd` the
1778/// session's, for the records' repo), then off the reserved file. The lessons it delivered; none already
1779/// taken (an earlier answer, `release_asks`).
1780pub(crate) fn commit_asks(session: &str, cwd: &str, ids: &[String]) -> Vec<String> {
1781    let all = lessons(session);
1782    let edits = std::fs::read_to_string(edits_file(session)).unwrap_or_default();
1783    let input = json!({"session_id": session, "cwd": cwd});
1784    let mut out = Vec::new();
1785    take_reserved(session, Some(ids), |taken| {
1786        for (check, ids) in taken {
1787            let members: Vec<&conscience::Challenge> = ids.iter().filter_map(|i| all.iter().find(|c| &c.id == i)).collect();
1788            let said = proposal_of(&edits, check).unwrap_or_default();
1789            let chars = conscience::render_late(&members, &one_line(said, LATE_EXCERPT)).len();
1790            mark_late(session, &input, check, &members, said, WAKE_EVENT, chars);
1791            out.extend(members.iter().map(|c| c.id.clone()));
1792        }
1793        true
1794    });
1795    out
1796}
1797
1798/// The reserved asks `ids` names (None: all of them) back on the `.queued` file: the wake never went
1799/// in, so the next prompt or wake delivers them. At a waiter's start and end every reserved ask goes
1800/// back (a waiter that ended or died with one reserved; `reclaim_reserved`). The wake ids put back; none
1801/// when the queue could not be written (they stay reserved, `take_reserved`).
1802pub(crate) fn release_asks(session: &str, ids: Option<&[String]>) -> Vec<String> {
1803    let q = queued_file(session);
1804    let taken = take_reserved(session, ids, |taken| {
1805        // Appended as `enqueue_all` does, under the lock `take_reserved` holds.
1806        let lines: String = taken.iter().map(|(check, ids)| format!("{}{}{}\n", ASK_NEXT[0], ids.join(","), tag(check))).collect();
1807        std::fs::OpenOptions::new().create(true).append(true).open(&q).and_then(|mut f| f.write_all(lines.as_bytes())).is_ok()
1808    });
1809    taken.iter().flat_map(|(check, ids)| ids.iter().map(|id| lesson_wake_id(check, id))).collect()
1810}
1811
1812/// The reserved asks back on the queue (`release_asks`) when no waiter of the session runs (listen.rs
1813/// `waiter_runs`): one that ended or died with a wake unanswered never delivers them, and until they are
1814/// back no prompt delivers them and no check grades them again (`askable`). At a prompt or tool call
1815/// (`queued`) and a detached check's stop.
1816fn reclaim_reserved(session: &str) {
1817    if reserved_file(session).exists() && !crate::listen::waiter_runs(&learn::home(), session) {
1818        release_asks(session, None);
1819    }
1820}
1821
1822/// At the session's end: whatever still waits in the `.queued` file was never delivered
1823/// (`never_delivered`, outcome.rs `DELIVERY`). Read, not taken: a resumed session still gets it at its
1824/// next prompt, and that delivery's record supersedes this one.
1825fn note_never_delivered(session: &str, input: &Value) {
1826    // And the asks a wake reserved that never went in (`reserve_asks`). Under the queue's lock: a
1827    // detached check queues before it (and is read here) or finds the session ended (`enqueue_all`).
1828    let text = live::locked(&queued_file(session), || {
1829        std::fs::read_to_string(queued_file(session)).unwrap_or_default() + &std::fs::read_to_string(reserved_file(session)).unwrap_or_default()
1830    });
1831    for line in text.lines() {
1832        let (route, ids, check) = queue_line(line);
1833        let channel = match route {
1834            Queued::Ask => "live",
1835            Queued::Instead | Queued::Free => "queued",
1836            Queued::Placebo => outcome::PLACEBO,
1837        };
1838        record_route(session, input, check, ids.into_iter().map(String::from).collect(), channel, outcome::ROUTE_NEVER, "session ended");
1839    }
1840}
1841
1842/// For each of `ids`, its value on the latest `prefix` line (`#borg-p `, `#borg-pf `) tagged `check`;
1843/// empty when any of them has none there (unknown, never a guess).
1844fn tagged_p(state: &str, prefix: &str, check: &str, ids: &[&str]) -> Vec<f64> {
1845    let mut seen: std::collections::HashMap<&str, f64> = std::collections::HashMap::new();
1846    for line in state.lines().filter_map(|l| l.strip_prefix(prefix)) {
1847        let (items, at) = tagged(line);
1848        if at != Some(check) {
1849            continue;
1850        }
1851        for (id, p) in items.split(',').filter_map(|it| it.rsplit_once('=')) {
1852            if let Ok(p) = p.trim().parse::<f64>() {
1853                seen.insert(id.trim(), p);
1854            }
1855        }
1856    }
1857    ids.iter().map(|i| seen.get(i).copied()).collect::<Option<Vec<f64>>>().unwrap_or_default()
1858}
1859
1860/// Lessons, each with the check that routed it.
1861type Pairs<'a> = Vec<(&'a conscience::Challenge, &'a str)>;
1862
1863/// Where a lesson the queue drops at delivery came from and why: its check, channel, route (outcome.rs
1864/// `ROUTES`) and reason.
1865type Drop<'a> = (&'a str, &'static str, &'static str, &'static str);
1866
1867/// The waiting graded lessons, once (`queued_file`): the asks a check queued (`ASK_NEXT`) unless done
1868/// (`done_in`, the definition its check's pool used), handed over all and first and marked and recorded
1869/// as asks on channel live (`record_asked`), a detached check's as questions about the turn it judged
1870/// (`PROPOSAL`, conscience.rs `render_late`, that message the work before them in the .edits file, as
1871/// before a block), a live check's from before 0.7.6 with the free lessons; the free lessons not asked or handed over in
1872/// the session yet (`had_in`), those in place of a question (`INSTEAD_NEXT`, `LEGACY_INSTEAD`) all and next, the
1873/// others as many as one early delivery fits beside them (`conscience::render_owing`); and last the
1874/// placebos not shown meanwhile (asked or handed over, now or before; one the budget cut stays a placebo),
1875/// never shown but marked for the judge (`#borg-placebo`, judged as a free delivery is). Free deliveries
1876/// are marked `#borg-queued <ids> @<check>` in the state and .edits files (after the work so far, `WORK`),
1877/// so neither this queue nor early delivery repeats them, and Stop treats them as early lessons (done once
1878/// acted on); an `asked` outcome per routing check with channel queued and that check's free propensity
1879/// (`#borg-pf`, `record_delivery`). Every lesson of the queue gets its `delivery` record: asked, free,
1880/// placebo, or dropped and why.
1881fn queued(so_far: &SoFar, hook_event: &str, session: &str, input: &Value) -> Option<String> {
1882    reclaim_reserved(session);
1883    let text = take_queue(session)?;
1884    let all = lessons(session);
1885    let state = std::fs::read_to_string(state_file(session)).unwrap_or_default();
1886    let had = had_in(&state);
1887    let mut done: Option<std::collections::HashSet<&str>> = None;
1888    let mut lines: Vec<(Queued, Vec<&str>, &str)> = text.lines().map(queue_line).collect();
1889    // Asks, then free lessons in place of questions, then the other free ones, then placebos
1890    // (`Queued`'s order), each in its lines' order.
1891    lines.sort_by_key(|l| l.0);
1892    let placebos = lines.iter().position(|l| l.0 == Queued::Placebo).unwrap_or(lines.len());
1893    // The messages detached checks judged (`PROPOSAL`), read only when an ask waits.
1894    let edits = if lines.iter().any(|l| l.0 == Queued::Ask) { std::fs::read_to_string(edits_file(session)).unwrap_or_default() } else { String::new() };
1895    // A detached check's asks (`late`), apart from the lessons the free delivery renders (`wanted`).
1896    let (mut wanted, mut placebo, mut late): (Pairs, Pairs, Pairs) = (Vec::new(), Vec::new(), Vec::new());
1897    let mut dropped: Vec<(Drop, &str)> = Vec::new();
1898    // The first `owed` of `wanted`: the legacy asks (its first `asks`) and the free lessons in place of
1899    // questions, which sort first.
1900    let (mut owed, mut asks) = (0, 0);
1901    for (route, ids, check) in &lines[..placebos] {
1902        let channel = if *route == Queued::Ask { "live" } else { "queued" };
1903        for &id in ids {
1904            let Some(c) = all.iter().find(|c| c.id == id) else {
1905                dropped.push(((check, channel, outcome::ROUTE_DROPPED_DONE, "not deliverable"), id));
1906                continue;
1907            };
1908            let why = if wanted.iter().chain(&late).any(|(c, _)| c.id == id) {
1909                Some("queued twice")
1910            } else if *route == Queued::Ask {
1911                done.get_or_insert_with(|| done_in(&state, &early_acted(session, &all, "", None))).contains(id).then_some("done")
1912            } else {
1913                had.contains(id).then_some("given before")
1914            };
1915            match why {
1916                Some(why) => dropped.push(((check, channel, outcome::ROUTE_DROPPED_DONE, why), id)),
1917                None if *route == Queued::Ask && proposal_of(&edits, check).is_some() => late.push((c, check)),
1918                None => {
1919                    owed += usize::from(*route != Queued::Free);
1920                    asks += usize::from(*route == Queued::Ask);
1921                    wanted.push((c, check));
1922                }
1923            }
1924        }
1925    }
1926    let free: Vec<&conscience::Challenge> = wanted.iter().map(|(c, _)| *c).collect();
1927    let shown = conscience::render_owing(&free, &[], owed);
1928    let n = shown.as_ref().map_or(0, |(_, n)| *n);
1929    dropped.extend(wanted[n..].iter().map(|(c, check)| ((*check, "queued", outcome::ROUTE_DROPPED_CHARS, "budget"), c.id.as_str())));
1930    // A placebo is a lesson the drone was not shown: decided against what this delivery shows (the free
1931    // lessons within the budget) and what it was shown before.
1932    for (_, ids, check) in &lines[placebos..] {
1933        for &id in ids {
1934            let Some(c) = all.iter().find(|c| c.id == id) else {
1935                dropped.push(((check, outcome::PLACEBO, outcome::ROUTE_DROPPED_DONE, "not deliverable"), id));
1936                continue;
1937            };
1938            let why = if placebo.iter().any(|(c, _)| c.id == id) {
1939                Some("queued twice")
1940            } else {
1941                (wanted[..n].iter().chain(&late).any(|(c, _)| c.id == id) || had.contains(id)).then_some("delivered")
1942            };
1943            match why {
1944                Some(why) => dropped.push(((check, outcome::PLACEBO, outcome::ROUTE_DROPPED_DONE, why), id)),
1945                None => placebo.push((c, check)),
1946            }
1947        }
1948    }
1949    let mut parts: Vec<String> = Vec::new();
1950    let mut markers: Vec<String> = Vec::new();
1951    // A detached check's asks, first and all (owed, as the block they replace), one delivery per check
1952    // naming the turn it judged; its marker pair after that turn's message in the .edits file, the work
1953    // before the questions (`pending`), so the next stop records the drone's answer (`note_answer`).
1954    for (check, ids) in by_check(&late) {
1955        let members: Vec<&conscience::Challenge> = ids.iter().filter_map(|i| late.iter().find(|(c, _)| c.id == *i).map(|(c, _)| *c)).collect();
1956        let said = proposal_of(&edits, check).unwrap_or_default();
1957        let text = conscience::render_late(&members, &one_line(said, LATE_EXCERPT));
1958        mark_late(session, input, check, &members, said, hook_event, text.len());
1959        so_far.late.borrow_mut().extend(ids.iter().map(|id| lesson_wake_id(check, id)));
1960        parts.push(text);
1961    }
1962    if let Some((text, n)) = shown {
1963        // The legacy asks, all shown (owed): one write per check, `#borg-live` right after its
1964        // `#borg-asked` (`pending` reads the pair as a live ask).
1965        let now = learn::now_ms();
1966        for (check, ids) in by_check(&wanted[..asks]) {
1967            let list = ids.join(",");
1968            markers.push(format!("#borg-asked {list}{}\n#borg-live {list} ms={now}", tag(check)));
1969            let members: Vec<&conscience::Challenge> = ids.iter().filter_map(|i| free.iter().find(|c| c.id == *i).copied()).collect();
1970            record_asked(session, input, &members, "live", check);
1971        }
1972        for (check, ids) in by_check(&wanted[asks..n]) {
1973            markers.push(format!("#borg-queued {}{}", ids.join(","), tag(check)));
1974            let delivered: Vec<&conscience::Challenge> = ids.iter().filter_map(|i| free.iter().find(|c| c.id == *i).copied()).collect();
1975            // The routing check's chance of handing each over free; 1 on an older binary's line (the
1976            // fixed rule's certain hand-over).
1977            let p = if check.is_empty() { vec![1.0; ids.len()] } else { tagged_p(&state, "#borg-pf ", check, &ids) };
1978            record_delivery(session, input, &delivered, "queued", check, p, Vec::new());
1979        }
1980        record_routes(session, input, &wanted[asks..n], "queued", outcome::ROUTE_FREE, "");
1981        log_cost(session, &format!("free event={hook_event} lessons={n} chars={} graded dropped={}", text.len(), wanted.len() - n));
1982        crate::stats::count(&learn::home(), crate::stats::QUEUED_FREE, session, |e| {
1983            e.n = n as u64;
1984            e.lessons = wanted[..n].iter().map(|(c, _)| c.id.clone()).collect();
1985            e.channel = "queued".into();
1986        });
1987        parts.push(text);
1988    }
1989    // Placebos: nothing reaches the drone; the next judge call sees each as it would a free delivery.
1990    for (check, ids) in by_check(&placebo) {
1991        markers.push(format!("#borg-placebo {}{}", ids.join(","), tag(check)));
1992    }
1993    record_routes(session, input, &placebo, outcome::PLACEBO, outcome::ROUTE_PLACEBO, "");
1994    if !markers.is_empty() {
1995        let markers = markers.join("\n");
1996        append(session, &markers);
1997        note_work(session, so_far, &markers);
1998    }
1999    let mut groups: Vec<(Drop, Vec<String>)> = Vec::new();
2000    for (key, id) in dropped {
2001        match groups.iter_mut().find(|(k, _)| *k == key) {
2002            Some((_, ids)) => ids.push(id.to_string()),
2003            None => groups.push((key, vec![id.to_string()])),
2004        }
2005    }
2006    for ((check, channel, route, why), ids) in groups {
2007        record_route(session, input, check, ids, channel, route, why);
2008    }
2009    (!parts.is_empty()).then(|| parts.join("\n\n"))
2010}
2011
2012/// This turn's work so far (harness `work_so_far`): from the transcript where the harness
2013/// writes one, else the turn's tool calls (`turn_file`): no prose.
2014fn work_so_far(h: &dyn Harness, input: &Value) -> String {
2015    let session = input.get("session_id").and_then(Value::as_str).unwrap_or("unknown");
2016    harness::work_so_far(h.caps(), input, &turn_file(session), EVIDENCE_CHARS)
2017}
2018
2019/// One hook's work so far (`work_so_far`), read when a delivery first needs it and then shared: the free
2020/// deliveries and placebos of one prompt or tool call write the same work line (`note_work`)
2021/// from one read of the transcript, and a hook that delivers nothing reads none.
2022struct SoFar<'a> {
2023    h: &'a dyn Harness,
2024    input: &'a Value,
2025    text: std::cell::OnceCell<String>,
2026    /// The detached checks' asks this hook delivered, as wake ids (`lesson_wake_id`): its answer's
2027    /// `shown_ids` (`free_context_with`).
2028    late: std::cell::RefCell<Vec<String>>,
2029}
2030
2031impl<'a> SoFar<'a> {
2032    fn new(h: &'a dyn Harness, input: &'a Value) -> Self {
2033        SoFar { h, input, text: std::cell::OnceCell::new(), late: std::cell::RefCell::default() }
2034    }
2035
2036    fn get(&self) -> &str {
2037        self.text.get_or_init(|| work_so_far(self.h, self.input))
2038    }
2039}
2040
2041/// This turn's tool calls, one `[tool] touch` line each, and their edit text: the work so far where the
2042/// harness sends no transcript. Emptied at each prompt.
2043fn turn_file(session: &str) -> PathBuf {
2044    state_file(session).with_extension("turn")
2045}
2046
2047/// One open queen question for the engineer (queen.rs `deliver`), at the first prompt of the session
2048/// where one of its work repos (the cwd's and the touched ones, systems.rs `work_repos`) has one: never
2049/// another repo's. Its commands run this binary in the session's `cwd`, wherever the drone's shell has
2050/// gone since. 0 model calls; fails open. Off with `BORG_QUEEN_ASK=off` or `BORG_LEARN=off` (nothing
2051/// recorded; eval drones cannot answer), and in the eval modes BORG_ONLY and BORG_CRITIC.
2052fn queen_ask(h: &dyn Harness, session: &str, cwd: &str) -> Option<String> {
2053    if !learning() || !crate::queen::ask_enabled() || std::env::var_os("BORG_ONLY").is_some() || std::env::var_os("BORG_CRITIC").is_some() {
2054        return None;
2055    }
2056    let repos = || {
2057        let mut repos = crate::systems::work_repos(&std::fs::read_to_string(where_file(session)).unwrap_or_default());
2058        let here = repo_of(cwd);
2059        if !here.is_empty() && !repos.contains(&here) {
2060            repos.insert(0, here);
2061        }
2062        repos
2063    };
2064    let scope = if cwd.is_empty() { String::new() } else { format!(" --dir {}", crate::queen::sh_quote(cwd)) };
2065    let mut q = crate::queen::deliver(&learn::home(), session, repos, &scope, learn::now_ms())?;
2066    if let Ok(exe) = std::env::current_exe() {
2067        q = q.run_by(&crate::queen::sh_quote(&exe.display().to_string()));
2068    }
2069    let text = crate::queen::render_delivery(&q, h.caps().question_dialog);
2070    log_cost(session, &format!("free event=UserPromptSubmit queen={} chars={}", q.id, text.len()));
2071    Some(text)
2072}
2073
2074/// The free context of a prompt or touch: waiting graded lessons first, then an early delivery.
2075fn free_context(h: &dyn Harness, hook_event: &str, session: &str, input: &Value, text: &str) -> Option<Value> {
2076    free_context_with(h, hook_event, session, input, text, None)
2077}
2078
2079/// `free_context` plus `extra` (a queen question) last.
2080/// The answer names the detached checks' asks it delivered in `"shown_ids"` (`lesson_wake_id`), as it
2081/// names the messages it shows (msg.rs `deliver`): an adapter drops a wake it holds for them
2082/// (listen.rs), and `for_harness` strips the key where no adapter reads it.
2083fn free_context_with(h: &dyn Harness, hook_event: &str, session: &str, input: &Value, text: &str, extra: Option<String>) -> Option<Value> {
2084    let so_far = SoFar::new(h, input);
2085    let parts: Vec<String> =
2086        [queued(&so_far, hook_event, session, input), early(h, &so_far, hook_event, session, input, text), extra].into_iter().flatten().collect();
2087    let mut out = (!parts.is_empty()).then(|| json!({"hookSpecificOutput": {"hookEventName": hook_event, "additionalContext": parts.join("\n\n")}}))?;
2088    let late = so_far.late.take();
2089    if !late.is_empty() {
2090        out["shown_ids"] = json!(late);
2091    }
2092    Some(out)
2093}
2094
2095/// What one model check of the work decided (`check`).
2096enum Verdict {
2097    /// The free gate skipped the model call: the score, or `noedit`.
2098    Skip(String),
2099    /// The model gave no answer: fail open.
2100    Failed,
2101    /// The ids to ask, in a block at the stop.
2102    Ask(Vec<String>),
2103}
2104
2105/// The work a check judges, as the stop reads it.
2106struct Work<'a> {
2107    /// Every lesson of the session it may deliver (the ungraded widening looks here).
2108    all: &'a [conscience::Challenge],
2109    /// `all` and the absorbed lessons withheld from it (`judged`): the judge's pending lessons look here,
2110    /// so a delivery made before an absorb still gets its verdict.
2111    judged: &'a [conscience::Challenge],
2112    /// The lessons that may be asked: not done (`done_in`), not decided already (waiting in the queue to
2113    /// be asked or handed over, `waiting_in`), not of another system.
2114    challenges: &'a [conscience::Challenge],
2115    done: &'a std::collections::HashSet<&'a str>,
2116    /// The state file as read before the check.
2117    state: &'a str,
2118    /// The state file plus the drone's work text: the gate finds the turn in it by its markers; the pool
2119    /// and the model read it without them (`conscience::work_text`).
2120    context: &'a str,
2121    scope_text: &'a str,
2122    /// The drone's message for the judge (`pending`): its final message at the stop.
2123    message: &'a str,
2124    /// BORG_STOP_GATE=edits: whether the turn changed anything.
2125    changed: bool,
2126    /// Whether this check judges the pending lessons: not while another (the session-end judge, holding
2127    /// the session lock) may be judging the same ones.
2128    judge: bool,
2129    /// The harness's stop budget (none: the model's own limit) and when the hook started: the model call
2130    /// gets what is left of it when it starts (live.rs `call_limit`).
2131    budget: Option<std::time::Duration>,
2132    started: std::time::Instant,
2133    /// A detached check's own limit for its model call (check_async.rs `call_limit`), in place of the
2134    /// stop's (`budget`); it decides on channel live and queues its asks for the next prompt
2135    /// (`settle`). None at the stop.
2136    detached: Option<std::time::Duration>,
2137    /// A detached check's .edits length at its stop (check_async.rs `Job::edits`): what its judge reads
2138    /// (`pending_upto`). None at the stop: all of it.
2139    edits: Option<usize>,
2140    /// This check's id (select.rs `check_id`), computed once when it started: on its `graded` record and
2141    /// on every marker of what it routed.
2142    check: &'a str,
2143    /// The selection summary beside the lessons file, read at the check's start (select/summary.rs
2144    /// `refresh`): what sizes the pool before the model call, and what a decision by a policy that reads
2145    /// no gate reads; a learned decision reads it again, under the store's decision lock (`decide`).
2146    summary: &'a crate::select::summary::Summary,
2147}
2148
2149impl Work<'_> {
2150    /// The model call's limit when it starts: a detached check's own, else what the stop's budget leaves
2151    /// (live.rs `call_limit`).
2152    fn limit(&self) -> std::time::Duration {
2153        self.detached.unwrap_or_else(|| live::call_limit(self.budget, self.started.elapsed()))
2154    }
2155
2156    /// The channel the check decides on: `stop`, or `live` for a detached check, whose asks reach the
2157    /// drone at its next prompt or tool call (select/summary.rs reads `live` as asked, no block).
2158    fn channel(&self) -> &'static str {
2159        if self.detached.is_some() { "live" } else { "stop" }
2160    }
2161}
2162
2163/// What a check's `graded` record says of its pool and its model call (`record_graded`).
2164struct Call<'a> {
2165    /// Lessons the pool allowed, and those shown in the order they filled its budget (whole.rs
2166    /// `select`; the shortlist's order without the whole pool).
2167    eligible: usize,
2168    order: &'a [&'a conscience::Challenge],
2169    /// The tail sample shown after the pool (whole.rs `tail`): each lesson with its place in the full
2170    /// fill order (`order.len()` + its place among the lessons the budget dropped), and each one's
2171    /// chance of the draw (0 without a tail).
2172    tail: &'a [(usize, &'a conscience::Challenge)],
2173    pt: f64,
2174    /// The boundary draw shown after the pool (whole.rs `explore`), each lesson with its place in the
2175    /// full fill order: an apply's fill position, never a tail lesson's or one the summary folds.
2176    boundary: &'a [(usize, &'a conscience::Challenge)],
2177    /// The pool's token budget the check filled to (0 without the whole pool), outcome.rs `budget`.
2178    budget: usize,
2179    model: &'a str,
2180    prompt_hash: String,
2181    dur_ms: u64,
2182    /// The call's cost as the CLI reported it (0 for a failed call).
2183    usd: f64,
2184    /// The compact prompt's lesson line format (whole.rs `line_format`), empty without it; the answer's
2185    /// design moves, and per apply its why (aligned with the record's applies): outcome.rs `line`,
2186    /// `moves`, `whys`.
2187    line: &'static str,
2188    moves: Vec<String>,
2189    whys: Vec<String>,
2190}
2191
2192/// The consequence model's check of `w` at the stop: the free gate (`gate`), the pool (with its tail
2193/// sample, whole.rs `tail`; `pool`), one graded call that also judges the pending lessons (`call`), and
2194/// the graded split (select.rs `graded_under`), decided and recorded as one step (`settle`, `decide`:
2195/// under the store's decision lock for the learned policy, which carries out a decision only once its
2196/// record landed where its gate is folded from). Its decision is logged before anything carries it out
2197/// (the `graded` record, then `#borg-graded`, `#borg-select`, `#borg-p`, `#borg-pf` in one write, then
2198/// the queue): a hook killed in between asks nothing and leaves no marker of an unlogged decision. The
2199/// free lessons and the placebos wait in the `.queued` file; what to ask is returned. Logged as `check`.
2200fn check(session: &str, input: &Value, w: &Work) -> Verdict {
2201    if let Some(score) = gate(w.challenges, w.context, w.changed) {
2202        return Verdict::Skip(score);
2203    }
2204    // What the pool and the model read: the work, without the state file's bookkeeping lines.
2205    let text = conscience::work_text(w.context);
2206    let pool = match pool(session, w, &text) {
2207        Ok(pool) => pool,
2208        Err(skip) => return skip,
2209    };
2210    let called = call(session, input, w, &pool, &text);
2211    settle(session, input, w, pool, called, None).verdict
2212}
2213
2214/// The free gate (prefilter.rs): a turn whose words touch none of `challenges` costs no model call; the
2215/// conscience then asked nothing. BORG_STOP_GATE=edits instead: a turn that changed nothing costs no
2216/// model call. Why it skips (the score, or `noedit`); None when the model is asked.
2217fn gate(challenges: &[conscience::Challenge], context: &str, changed: bool) -> Option<String> {
2218    if prefilter::edits_gate() { (!changed).then(|| "noedit".to_string()) } else { prefilter::skip(challenges, context).map(|score| format!("{score:.2}")) }
2219}
2220
2221/// What a check shows its model (`pool`).
2222struct Pool<'a> {
2223    /// The lessons shown: the pool's, then the boundary's and the tail sample's.
2224    shortlist: Vec<&'a conscience::Challenge>,
2225    /// `Call`'s fields: the lessons the pool allowed, those shown in fill order, the tail sample and
2226    /// its chance, the boundary draw, the pool's token budget.
2227    eligible: usize,
2228    order: Vec<&'a conscience::Challenge>,
2229    tail: Vec<(usize, &'a conscience::Challenge)>,
2230    pt: f64,
2231    boundary: Vec<(usize, &'a conscience::Challenge)>,
2232    budget: usize,
2233    /// The whole pool (whole.rs `on`), not the shortlist.
2234    whole: bool,
2235    /// The pending lessons the call judges (`pending`).
2236    judging: Judging<'a>,
2237    /// What a detached check logs of its pool beside its record (pools.rs), None without the whole
2238    /// pool; and the work text the pool read, which its shadow scores read again after the call.
2239    rows: Option<crate::pools::Rows<'a>>,
2240    text: String,
2241}
2242
2243/// The pool of `w` over its work `text` (`Pool`); `Err` with the skip when there is nothing to ask and
2244/// no question to judge.
2245fn pool<'a>(session: &str, w: &Work<'a>, text: &str) -> Result<Pool<'a>, Verdict> {
2246    // BM25's top 30, plus lessons the vector index finds that word matching missed (vectors.rs; here only,
2247    // where the model call dwarfs loading the embedder). Top 30 found as much as top 20 + vectors; both
2248    // together find the most (internal evaluation "vectors", not published).
2249    let started = std::time::Instant::now();
2250    let shortlist_now = || {
2251        let (list, added) = crate::vectors::shortlist(&lessons_path(), w.challenges, text, w.scope_text, crate::vectors::SHORTLIST);
2252        if crate::vectors::enabled() {
2253            log_cost(session, &format!("vectors added={added} ms={}", started.elapsed().as_millis()));
2254        }
2255        list
2256    };
2257    // The whole pool (whole.rs, default): every lesson the Stop allows, up to a token budget; the
2258    // shortlist's order decides only when they do not all fit. BORG_STOP_FORMAT=classic asks about that
2259    // pool in the shortlist's layout (predict::prompt).
2260    let whole = crate::whole::on();
2261    let (shortlist, eligible, order, tail, pt, boundary, budget, rows) = if whole {
2262        let raw = conscience::raw(w.challenges, text);
2263        let ranked = conscience::rank_raw(w.challenges, raw.clone(), w.scope_text);
2264        let eligible = crate::whole::eligible(w.challenges, &ranked, w.scope_text);
2265        let n = eligible.len();
2266        // The pool's budget: BORG_STOP_POOL_TOKENS when set, else the policy's (select.rs `pool_tokens`:
2267        // today's `whole::TOKENS` unless the learned policy's gate may risk a larger one that fits the
2268        // model call's limit).
2269        let tokens = match std::env::var_os("BORG_STOP_POOL_TOKENS") {
2270            Some(_) => crate::whole::tokens(),
2271            None => {
2272                let limit = w.limit();
2273                crate::select::pool_tokens(w.summary, Some(u64::try_from(limit.as_millis()).unwrap_or(u64::MAX)), session)
2274            }
2275        };
2276        // Exploration (whole.rs `explore`, plan W3b "Pool"), under every policy: a pool the budget cuts
2277        // fills its last share by a draw at the boundary of the fill order, and shows a uniform tail
2278        // sample of the lessons the budget dropped after it (its stable prefix stays), so the yield of
2279        // fill positions past the budget is measured; each lesson's chance is known (pools.rs).
2280        let e = crate::whole::explore(eligible, shortlist_now, &ranked, tokens, crate::whole::boundary_share(), tail_seed(session, w.check, w.summary));
2281        let sampled = if e.tail.is_empty() { String::new() } else { format!(" tail={}", e.tail.len()) };
2282        log_cost(
2283            session,
2284            &format!(
2285                "pool mode=all eligible={n} shown={} budget={}{sampled} boundary={}",
2286                e.filled.shown.len(),
2287                if e.filled.fell { "full" } else { "fits" },
2288                e.boundary.len()
2289            ),
2290        );
2291        // A detached check logs its pool (pools.rs); a stop's own check (BORG_ASYNC=off) adds nothing.
2292        let rows = w.detached.map(|_| crate::pools::Rows::of(&e, w.challenges, &raw, tokens));
2293        let mut shown = e.filled.shown.clone();
2294        shown.extend(e.boundary.iter().chain(&e.tail).map(|(_, c)| *c));
2295        let crate::whole::Explored { filled, tail, pt, boundary, .. } = e;
2296        (shown, n, filled.order, tail, pt, boundary, tokens, rows)
2297    } else {
2298        let list = shortlist_now();
2299        (list.clone(), list.len(), list, Vec::new(), 0.0, Vec::new(), 0, None)
2300    };
2301    // R1: the same call judges what the drone did with lessons delivered earlier this session.
2302    let judging = if w.judge {
2303        pending_upto(session, w.state, w.judged, w.message, When::Stop, w.check, w.edits)
2304    } else {
2305        Judging { pending: Vec::new(), question: false, crowded: Vec::new() }
2306    };
2307    // Nothing to ask and no question to judge (no lesson in the pool, none pending): no model call.
2308    // Free deliveries alone never pay for a call: they wait for the next one, or `borg learn` settles them.
2309    if shortlist.is_empty() && !judging.question {
2310        return Err(Verdict::Skip("empty".into()));
2311    }
2312    Ok(Pool { shortlist, eligible, order, tail, pt, boundary, budget, whole, judging, rows, text: text.to_string() })
2313}
2314
2315/// A check's model call (`call`): the answer (None: failed), how long it took, and how it was asked.
2316struct Called {
2317    pred: Option<predict::Prediction>,
2318    dur_ms: u64,
2319    model: String,
2320    /// The judge section's characters (`predict::judge_section`), the cost line's `chars=`.
2321    extra: usize,
2322    graded: bool,
2323    compact: bool,
2324}
2325
2326/// The model call of `w` over `pool` and its work `text`, within `Work::limit`; counted as a `check`
2327/// (`note_check`; a detached check's once it is done).
2328fn call(session: &str, input: &Value, w: &Work, pool: &Pool, text: &str) -> Called {
2329    let model = std::env::var("BORG_PREDICT_MODEL").unwrap_or_default();
2330    let pend = &pool.judging.pending;
2331    let extra = predict::judge_section(pend).len();
2332    let graded = crate::whole::graded();
2333    let compact = pool.whole && !crate::whole::classic();
2334    let timeout = w.limit();
2335    // The work the call shows holds no stripped question's declaration either: this turn's message may
2336    // be the drone's answer to it.
2337    let shown_work = without_declarations(text, pend);
2338    let called = std::time::Instant::now();
2339    let pred = if compact {
2340        predict::predict_whole(&shown_work, &pool.shortlist, pend, &model, graded, timeout)
2341    } else {
2342        predict::predict(&shown_work, &pool.shortlist, pend, &model, graded, timeout)
2343    };
2344    let dur_ms = called.elapsed().as_millis() as u64;
2345    // A detached check's comes when it is done (`run_job`), with the peak memory of all it did.
2346    if w.detached.is_none() {
2347        note_check(session, input, pool.shortlist.len(), dur_ms, pred.as_ref().map(|p| p.cost), None);
2348    }
2349    Called { pred, dur_ms, model, extra, graded, compact }
2350}
2351
2352/// The `check` event of a model call: the lessons shown, its duration, its cost (None: it failed); a
2353/// detached check's (`limit`, its call's limit) in mode `async` with the process's peak memory.
2354fn note_check(session: &str, input: &Value, shown: usize, dur_ms: u64, cost: Option<f64>, limit: Option<std::time::Duration>) {
2355    stat(session, input, crate::stats::CHECK, |e| {
2356        e.n = shown as u64;
2357        e.dur_ms = dur_ms;
2358        e.usd = cost.unwrap_or(0.0);
2359        if cost.is_none() {
2360            e.why = "failed".into();
2361        }
2362        if let Some(limit) = limit {
2363            e.mode = crate::check_async::MODE.into();
2364            e.rss_kb = crate::check_async::rss_kb();
2365            e.limit_ms = u64::try_from(limit.as_millis()).unwrap_or(u64::MAX);
2366        }
2367    });
2368}
2369
2370/// What a detached check (check_async.rs) carries beside a stop's check when it writes its answer, under
2371/// the session lock: the state file's length at its stop (`offset`), whether the session ended since
2372/// (`gone`), and the drone's message it judged (`proposal`).
2373pub(crate) struct Late<'l> {
2374    pub offset: usize,
2375    pub gone: bool,
2376    pub proposal: &'l str,
2377    /// When its stop ran: what its pool log (pools.rs) reads of the store is what was there before.
2378    pub stop_ms: i64,
2379}
2380
2381/// What a check carried out (`settle`): its verdict (a detached check's asks are queued already), the
2382/// lessons it handed over free, the waiting asks of earlier checks it took off the queue
2383/// (`supersede`), and a detached check's pool to log once it let go of the session lock (`PoolLog`).
2384struct Settled<'a> {
2385    verdict: Verdict,
2386    free: Vec<String>,
2387    superseded: Vec<String>,
2388    pool: Option<PoolLog<'a>>,
2389}
2390
2391/// A detached check's pool and what its call made of it, for its line of the pool log (pools.rs),
2392/// scored after the check wrote its answer and let go of the session lock (`run_job`, `log_pool`).
2393struct PoolLog<'a> {
2394    rows: crate::pools::Rows<'a>,
2395    text: String,
2396    applied: Vec<(&'a conscience::Challenge, u8)>,
2397    pending: Vec<String>,
2398    line: &'static str,
2399    model: String,
2400    prompt_hash: String,
2401    failed: bool,
2402    /// Its index among the session's checks, read before its own is counted (check_async.rs `index`).
2403    idx: u64,
2404    stop_ms: i64,
2405}
2406
2407/// `p`'s line of the pool log (pools.rs `log`) for the check of `w`: the lessons the session had and
2408/// those in play read from its state at the check.
2409fn log_pool(session: &str, w: &Work, p: &PoolLog) {
2410    let delivered = had_in(w.state);
2411    let in_play: std::collections::HashSet<&str> = delivered.iter().copied().chain(applied_in(w.state)).collect();
2412    let c = crate::pools::Check {
2413        session,
2414        check: w.check,
2415        idx: p.idx,
2416        stop_ms: p.stop_ms,
2417        line: p.line,
2418        model: &p.model,
2419        prompt_hash: &p.prompt_hash,
2420        failed: p.failed,
2421        applied: &p.applied,
2422        pending: &p.pending,
2423        text: &p.text,
2424        scope_text: w.scope_text,
2425        challenges: w.challenges,
2426        delivered: &delivered,
2427        in_play: &in_play,
2428        lessons: &lessons_path(),
2429        outcomes: &graded_log(),
2430    };
2431    crate::pools::log(&learn::home(), &p.rows, &c);
2432}
2433
2434/// `w`'s answer `called` over `pool`, recorded and carried out (`check`). A detached check (`late`)
2435/// records no verdict on a lesson settled or delivered again since its stop (`touched_since`): the
2436/// session-end judge's, or a later delivery's, stands. Its asks wait for the next prompt or tool call
2437/// (`ASK_NEXT`, its judged message beside them in the .edits file, `PROPOSAL`), after it took the asks
2438/// of earlier checks it showed its model again off the queue (`supersede`); when the session ended
2439/// during its call nothing waits, and each routed lesson's `delivery` record says so.
2440fn settle<'a>(session: &str, input: &Value, w: &Work<'a>, pool: Pool<'a>, called: Called, late: Option<&Late>) -> Settled<'a> {
2441    let Pool { shortlist, eligible, order, tail, pt, boundary, budget, judging, rows, text, .. } = pool;
2442    let Called { pred, dur_ms, model, extra, graded, compact } = called;
2443    let none = |verdict| Settled { verdict, free: Vec::new(), superseded: Vec::new(), pool: None };
2444    let usd = pred.as_ref().map_or(0.0, |p| p.cost);
2445    let line = if compact { crate::whole::line_format().name() } else { "" };
2446    let mut call = Call {
2447        eligible,
2448        order: &order,
2449        tail: &tail,
2450        pt,
2451        boundary: &boundary,
2452        budget,
2453        model: &model,
2454        prompt_hash: predict::template_hash(compact, graded),
2455        dur_ms,
2456        usd,
2457        line,
2458        moves: Vec::new(),
2459        whys: Vec::new(),
2460    };
2461    note_unjudged(session, &judging.crowded, "slots");
2462    let mut pend = judging.pending;
2463    if let Some(late) = late {
2464        let state = std::fs::read_to_string(state_file(session)).unwrap_or_default();
2465        let since = touched_since(state.get(late.offset..).unwrap_or_default());
2466        pend.retain(|p| !since.contains(p.lesson.id.as_str()));
2467    }
2468    // A detached check's pool, with the applies' grades, for its line of the pool log (`PoolLog`).
2469    let pending: Vec<String> = pend.iter().map(|p| p.lesson.id.clone()).collect();
2470    let pool_log = |applied: &[(&'a conscience::Challenge, u8)], failed: bool, prompt_hash: &str| {
2471        let (rows, late) = (rows?, late?);
2472        Some(PoolLog {
2473            rows,
2474            text,
2475            applied: applied.to_vec(),
2476            pending,
2477            line,
2478            model: model.clone(),
2479            prompt_hash: prompt_hash.to_string(),
2480            failed,
2481            idx: crate::check_async::index(session),
2482            stop_ms: late.stop_ms,
2483        })
2484    };
2485    let Some(pred) = pred else {
2486        // Fail open: nothing asked, the pending lessons stay pending (no try counted).
2487        if graded {
2488            record_graded(session, input, w, &[], None, &call);
2489        }
2490        note_unjudged(session, &pend.iter().map(|p| p.lesson.id.clone()).collect::<Vec<_>>(), "failed");
2491        let pool = if graded { pool_log(&[], true, &call.prompt_hash) } else { None };
2492        return Settled { pool, ..none(Verdict::Failed) };
2493    };
2494    // The judge's call numbering counts every model call of the session.
2495    append(session, &format!("#borg-predict-cost {:.4}", pred.cost));
2496    // The session ended during a detached check's call: the session-end judge has its pending lessons
2497    // (and `borg learn` may have settled them), so no verdict of this call lands after that.
2498    let gone = late.is_some_and(|l| l.gone);
2499    if !pend.is_empty() && !gone {
2500        // A detached check counts in the state file as it is now: a check that wrote after its stop
2501        // gets its own number.
2502        let call = match late {
2503            Some(_) => std::fs::read_to_string(state_file(session)).unwrap_or_default().lines().filter(|l| l.starts_with("#borg-predict-cost")).count(),
2504            None => w.state.lines().filter(|l| l.starts_with("#borg-predict-cost")).count() + 1,
2505        };
2506        // The whole prompt's characters (`prompt=`): the judge's effective length is measured by it (judge_length.rs).
2507        let logged = Logged { call, chars: extra, model: judge_model(&model), fields: format!(" prompt={}", pred.prompt) };
2508        record_verdicts(session, &repo_of(input.get("cwd").and_then(Value::as_str).unwrap_or_default()), "", &pend, &pred, &logged);
2509    }
2510    // The applies decide; a why each, empty where the answer gave none.
2511    let (applied, whys): (Vec<(&conscience::Challenge, u8)>, Vec<String>) = pred
2512        .applies
2513        .iter()
2514        .zip(&pred.grades)
2515        .enumerate()
2516        .filter_map(|(at, (i, g))| shortlist.iter().find(|c| &c.id == i).map(|c| ((*c, *g), pred.whys.get(at).cloned().unwrap_or_default())))
2517        .unzip();
2518    call.moves.clone_from(&pred.moves);
2519    call.whys = whys;
2520    let ids = |cs: &[&conscience::Challenge]| cs.iter().map(|c| c.id.clone()).collect::<Vec<_>>();
2521    if graded {
2522        // Graded (default): select.rs decides what of the applies to ask, hand over free at the next
2523        // prompt or touch, or judge as a placebo (the fixed rule: ask only for a 3, else the 2s and 1s
2524        // free, conscience.rs `graded_without` with the session's `had`).
2525        let lessons = lessons_path();
2526        // Decided at the pool tokens the record logs, which the fit and the gate's ledger read; and, under
2527        // the learned policy, with the store's other checks one at a time (`decide`).
2528        let at = Decide {
2529            session,
2530            check: w.check,
2531            channel: w.channel(),
2532            lessons: &lessons,
2533            tokens: crate::select::shown_tokens(call.order),
2534            recorded: records_reach_gate(&lessons),
2535        };
2536        let r = decide(crate::select::policy(), &applied, &at, w.summary, DECIDE_WAIT, |r, basis| {
2537            record_graded(session, input, w, &applied, Some((r, basis)), &call)
2538        });
2539        let pool = pool_log(&applied, false, &call.prompt_hash);
2540        let routed = |cs: &[&conscience::Challenge], p: &[f64]| -> Vec<(&str, f64)> {
2541            applied.iter().zip(p).filter(|((c, _), _)| cs.iter().any(|x| x.id == c.id)).map(|((c, _), p)| (c.id.as_str(), *p)).collect()
2542        };
2543        let mut lines = vec![graded_line(&applied), r.log.line(applied.len(), w.check)];
2544        lines.extend(r.log.attention.clone());
2545        lines.extend(p_line("#borg-p", routed(&r.ask, &r.p), Some(w.check)));
2546        lines.extend(p_line("#borg-pf", routed(&r.free, &r.pf), Some(w.check)));
2547        append(session, &lines.join("\n"));
2548        let superseded = if late.is_some() && !gone { supersede(session, input, &shortlist, &r.ask) } else { Vec::new() };
2549        // What it queues, in queue order: its asks (a detached check's), the free lessons, the placebos.
2550        let mut queue: Vec<String> = Vec::new();
2551        if let Some(late) = late
2552            && !r.ask.is_empty()
2553        {
2554            // Its asks, with the message it judged (the turn their delivery names), first in the queue.
2555            note_proposal(session, w.check, late.proposal);
2556            queue.push(format!("{}{}{}", ASK_NEXT[0], ids(&r.ask).join(","), tag(w.check)));
2557        }
2558        if !r.free.is_empty() {
2559            // The learned policy's hand-overs of lessons the fixed rule asks (on a check with a 3: the
2560            // evidence rule's) are marked (`INSTEAD_NEXT`), so the free delivery's budget never cuts what
2561            // a question would have carried.
2562            let asks = if r.log.policy == crate::select::Policy::Learned.name() { conscience::graded_without(&applied, None, &[]).0 } else { Vec::new() };
2563            let (owed, rest): (Vec<&conscience::Challenge>, Vec<&conscience::Challenge>) = r.free.iter().partition(|c| asks.iter().any(|a| a.id == c.id));
2564            if !owed.is_empty() {
2565                queue.push(format!("{INSTEAD_NEXT}{}{}", ids(&owed).join(","), tag(w.check)));
2566            }
2567            if !rest.is_empty() {
2568                queue.push(format!("{}{}", ids(&rest).join(","), tag(w.check)));
2569            }
2570        }
2571        if !r.placebo.is_empty() {
2572            queue.push(format!("placebo {}{}", ids(&r.placebo).join(","), tag(w.check)));
2573        }
2574        // A detached check's lines land only while the session lives, checked under the queue's lock
2575        // the end hook reads it under (`note_never_delivered`): what it queued is either read there as
2576        // never delivered, or never queued and dropped here.
2577        let landed = !gone && enqueue_all(session, &queue, late.map(|l| l.offset));
2578        log_cost(
2579            session,
2580            &format!(
2581                "graded applied={} ask={} free={} {}",
2582                applied.len(),
2583                r.ask.len(),
2584                r.free.len(),
2585                if landed && !r.free.is_empty() { "queued" } else { "dropped" }
2586            ),
2587        );
2588        if !landed {
2589            // The session ended during the call: no next prompt comes for what it routed.
2590            for (routed, channel) in [(&r.ask, "live"), (&r.free, "queued"), (&r.placebo, outcome::PLACEBO)] {
2591                record_route(session, input, w.check, ids(routed), channel, outcome::ROUTE_DROPPED_DONE, "session ended");
2592            }
2593            return Settled { pool, superseded, ..none(Verdict::Ask(Vec::new())) };
2594        }
2595        Settled { verdict: Verdict::Ask(ids(&r.ask)), free: ids(&r.free), superseded, pool }
2596    } else {
2597        // R3b: the model decides whether to ask (at the stop, an extra round); once it does, the questions
2598        // also carry every other lesson over the low bar (conscience.rs `widen`).
2599        let predicted: Vec<&conscience::Challenge> = applied.iter().map(|(c, _)| *c).collect();
2600        let text = conscience::work_text(w.context);
2601        let pool = conscience::widen_pool(&conscience::rank_scoped(w.all, &text, w.scope_text), &predicted, w.done);
2602        let widened: Vec<&conscience::Challenge> =
2603            conscience::in_scope_first(conscience::explored(pool.iter().copied())).into_iter().filter(|c| w.challenges.iter().any(|x| x.id == c.id)).collect();
2604        append_p_of(session, &pool, &widened, w.check);
2605        // Ungraded, the stop asks whatever it widens to: so it would have asked these.
2606        none(Verdict::Ask(ids(&widened)))
2607    }
2608}
2609
2610/// The lessons the state lines `after` (past a detached check's stop) name in a verdict
2611/// (`#borg-judged`, `END_JUDGED`) or a delivery (`DELIVERED`): settled or delivered again since, so the
2612/// check records no verdict on their earlier delivery (`settle`).
2613fn touched_since(after: &str) -> std::collections::HashSet<&str> {
2614    let mut out = std::collections::HashSet::new();
2615    for l in after.lines() {
2616        if let Some(rest) = l.strip_prefix("#borg-judged ") {
2617            out.insert(rest.split_once('=').map_or(rest, |(id, _)| id));
2618        } else if let Some(list) = l.strip_prefix(END_JUDGED).or_else(|| DELIVERED.iter().find_map(|m| l.strip_prefix(m))) {
2619            out.extend(ids_in(list));
2620        }
2621    }
2622    out
2623}
2624
2625/// Where a graded check decides (`decide`): the parts of its select.rs `Ctx` the summary does not hold.
2626struct Decide<'x> {
2627    session: &'x str,
2628    check: &'x str,
2629    channel: &'static str,
2630    lessons: &'x std::path::Path,
2631    /// The check's pool tokens (select.rs `Ctx::tokens`).
2632    tokens: u32,
2633    /// Whether the check's `graded` record lands in the log its gate is folded from
2634    /// (`records_reach_gate`): a learned decision whose record would not is never debited.
2635    recorded: bool,
2636}
2637
2638/// What a graded check's decision read (`decide`), for its record: the selection summary, the lessons
2639/// the session already had or has waiting in its queue (select.rs `Ctx::had`), and why it fell back to
2640/// the fixed rule (outcome.rs `DECIDED_BUSY`, `DECIDED_NO_LOCK`, `DECIDED_NO_RECORD`; empty when it
2641/// decided by the policy in force).
2642struct Basis<'s> {
2643    summary: &'s crate::select::summary::Summary,
2644    had: &'s [String],
2645    why: &'static str,
2646}
2647
2648/// How long a learned check waits for the store's decision lock (`decide`) at most; past it, the check
2649/// takes the fixed rule and its record says why. Guard: a check that holds the lock long (a summary
2650/// rebuilt from a long log, a stopped process) would otherwise hold up every other learned check's
2651/// answer. Derived: a third of
2652/// what the stop's model call leaves for the hook's records and answer (live.rs `AFTER_CALL`), so the
2653/// wait never costs a stop its answer.
2654const DECIDE_WAIT: std::time::Duration = std::time::Duration::from_millis(live::AFTER_CALL.as_millis() as u64 / 3);
2655
2656/// Method: how often a check waiting for the decision lock tries it again (`lock_within`).
2657const DECIDE_POLL: std::time::Duration = std::time::Duration::from_millis(5);
2658
2659/// The store's decision lock (`decide`), beside the selection summary it guards.
2660fn decide_lock(lessons: &std::path::Path) -> PathBuf {
2661    crate::select::summary::cache_path(lessons).with_extension("json.lock")
2662}
2663
2664/// `file` locked exclusively, tried every `DECIDE_POLL` for at most `wait` (once for none); unlocked when
2665/// the handle drops. Err with why when it stays held (`DECIDED_BUSY`) or cannot be taken at all
2666/// (`DECIDED_NO_LOCK`).
2667fn lock_within(file: &std::path::Path, wait: std::time::Duration) -> Result<std::fs::File, &'static str> {
2668    let lock = std::fs::OpenOptions::new().create(true).truncate(false).write(true).open(file).map_err(|_| outcome::DECIDED_NO_LOCK)?;
2669    let until = std::time::Instant::now() + wait;
2670    loop {
2671        match lock.try_lock() {
2672            Ok(()) => return Ok(lock),
2673            Err(std::fs::TryLockError::WouldBlock) if std::time::Instant::now() < until => std::thread::sleep(DECIDE_POLL),
2674            Err(std::fs::TryLockError::WouldBlock) => return Err(outcome::DECIDED_BUSY),
2675            Err(std::fs::TryLockError::Error(_)) => return Err(outcome::DECIDED_NO_LOCK),
2676        }
2677    }
2678}
2679
2680/// A graded check's decision and its record (`record`, which answers whether the record landed) as one
2681/// step. Only the learned policy reads the gate (select/policy.rs `choose`): its checks decide one at a
2682/// time under the store's decision lock (`decide_lock`), never during the model call, since learned
2683/// checks that decide at once (the checks of several sessions on one store) would each take all the gate
2684/// may still risk. Under the lock the summary is loaded again
2685/// (select/summary.rs `refresh`: the log's tail since the check started, other checks' records and their
2686/// ledger entries included), decided on (select.rs `graded_under`) and recorded with that summary as the
2687/// one the decision read (outcome.rs `seen`); then released. A learned decision is carried out only once
2688/// its record landed in the log its gate is folded from, the gate's only debit: one whose record would
2689/// not land there (`Decide::recorded`) or did not (`record`'s answer) takes the fixed rule instead
2690/// (`DECIDED_NO_RECORD`), as one that cannot take the lock within `wait` does (on `stale`, the summary
2691/// it read at its start), and its record says why (`Basis::why`). The fixed rule and the evaluation
2692/// switches read no gate: they decide by themselves on `stale`, without the lock, whatever other checks
2693/// hold. Every way, the session's lessons delivered by now (`had_in`, read again: a tool call may have
2694/// handed one over during the model call) or waiting in its queue (`waiting_in`) never go free.
2695fn decide<'a>(
2696    policy: crate::select::Policy,
2697    applied: &[(&'a conscience::Challenge, u8)],
2698    at: &Decide,
2699    stale: &crate::select::summary::Summary,
2700    wait: std::time::Duration,
2701    mut record: impl FnMut(&crate::select::Routed<'a>, &Basis) -> bool,
2702) -> crate::select::Routed<'a> {
2703    use crate::select::Policy;
2704    let state = std::fs::read_to_string(state_file(at.session)).unwrap_or_default();
2705    let queue = std::fs::read_to_string(queued_file(at.session)).unwrap_or_default();
2706    let reserved = std::fs::read_to_string(reserved_file(at.session)).unwrap_or_default();
2707    // A detached check (channel live) grades the waiting asks again (`supersede`); one a wake reserved is
2708    // on its way (`reserve_asks`).
2709    let mut had: Vec<String> =
2710        had_in(&state).into_iter().chain(waiting_in(&queue, at.channel == "live")).chain(waiting_in(&reserved, false)).map(str::to_string).collect();
2711    had.sort_unstable();
2712    had.dedup();
2713    // The collective's tolerance, read once: every way below decides and logs under it (select.rs `Ctx::alpha`).
2714    let alpha = crate::select::alpha(at.lessons);
2715    // Decided under `policy` on `summary` and recorded with `why`; whether the record landed.
2716    let mut on = |summary: &crate::select::summary::Summary, policy: Policy, why: &'static str| {
2717        let ctx =
2718            crate::select::Ctx { session: at.session, check: at.check, channel: at.channel, lessons: at.lessons, summary, tokens: at.tokens, had: &had, alpha };
2719        let r = crate::select::graded_under(policy, applied, &ctx);
2720        let landed = record(&r, &Basis { summary, had: &had, why });
2721        (r, landed)
2722    };
2723    if policy != Policy::Learned {
2724        return on(stale, policy, "").0;
2725    }
2726    if !at.recorded {
2727        return on(stale, Policy::Fixed, outcome::DECIDED_NO_RECORD).0;
2728    }
2729    match lock_within(&decide_lock(at.lessons), wait) {
2730        Ok(_held) => {
2731            let summary = crate::select::summary::refresh(at.lessons);
2732            match on(&summary, policy, "") {
2733                (r, true) => r,
2734                // Still under the lock: no other check decided on the gate this one left as it was.
2735                (_, false) => on(&summary, Policy::Fixed, outcome::DECIDED_NO_RECORD).0,
2736            }
2737        }
2738        Err(why) => on(stale, Policy::Fixed, why).0,
2739    }
2740}
2741
2742/// Whether a check's `graded` record (`record_graded`) lands in the log its gate is folded from, the
2743/// outcomes beside `lessons` (select/summary.rs `refresh`): learning on, and `lessons` in the store the
2744/// hook records to. A `BORG_CONSCIENCE` elsewhere reads the gate of its own store, which no record of
2745/// this hook reaches.
2746fn records_reach_gate(lessons: &std::path::Path) -> bool {
2747    learning() && outcome::path_beside(lessons) == graded_log()
2748}
2749
2750/// Where a check's `graded` record goes: beside the store's other outcomes.
2751fn graded_log() -> PathBuf {
2752    learn::home().join("outcomes.jsonl")
2753}
2754
2755/// The seed of a check's tail sample (whole.rs `tail`): its own stream of the check's seed (select.rs
2756/// `seed`, which reads only the session, channel and check; the tag spells "tailpool").
2757fn tail_seed(session: &str, check: &str, summary: &crate::select::summary::Summary) -> u64 {
2758    let ctx = crate::select::Ctx {
2759        session,
2760        check,
2761        channel: "stop",
2762        lessons: std::path::Path::new(""),
2763        summary,
2764        tokens: 0,
2765        had: &[],
2766        alpha: crate::select::policy::ALPHA,
2767    };
2768    crate::select::seed(&ctx) ^ u64::from_be_bytes(*b"tailpool")
2769}
2770
2771/// The `graded` record of one check (outcome.rs `GRADED`, `graded_outcome`), appended beside the store's
2772/// other outcomes (`graded_log`); whether it landed. Nothing with learning off.
2773fn record_graded(
2774    session: &str,
2775    input: &Value,
2776    w: &Work,
2777    applied: &[(&conscience::Challenge, u8)],
2778    decided: Option<(&crate::select::Routed, &Basis)>,
2779    call: &Call,
2780) -> bool {
2781    if !learning() {
2782        return false;
2783    }
2784    let repo = repo_of(input.get("cwd").and_then(Value::as_str).unwrap_or_default());
2785    let mut o = graded_outcome(session, &repo, w.check, applied, decided, w.summary, call);
2786    // A detached check's decision is the live channel's (`Work::channel`).
2787    o.channel = w.channel().into();
2788    if w.detached.is_some() {
2789        o.run = crate::check_async::MODE.into();
2790    }
2791    outcome::append(&graded_log(), &o).is_ok()
2792}
2793
2794/// The `graded` record of one check: every check that called the model, one with no applies and a
2795/// failed one (`decided` none; its fit is that of `read`, the summary the check read at its start)
2796/// included, so each decision and each pool can be replayed. Per apply its grade, rank, fill position
2797/// (`Call::order`; a tail lesson's past the budget's, `Call::tail`), propensities to be asked (`p`) and
2798/// handed over free (`pf`); the routes (`chosen`, `free`, `placebo`, and ψ, each apply's chance of the
2799/// placebo draw among those routed nowhere, select.rs `Log::psi`); the applies that could not go free
2800/// because the session had them (`nofree`); the question-count test's k and s on a check it considered
2801/// (`xk`, `xs`, select.rs `Log::xk`); the tail sample (`tail`, `pt`, and after the applies' fill
2802/// positions each tail lesson's, outcome.rs `fill`); the policy, its fit, the summary the decision read
2803/// (`Basis`) and why it fell back to the fixed rule (`reason`); and the call.
2804#[allow(clippy::too_many_arguments)] // reason: one record of a check, from the parts the check holds apart
2805fn graded_outcome(
2806    session: &str,
2807    repo: &str,
2808    check: &str,
2809    applied: &[(&conscience::Challenge, u8)],
2810    decided: Option<(&crate::select::Routed, &Basis)>,
2811    read: &crate::select::summary::Summary,
2812    call: &Call,
2813) -> outcome::Outcome {
2814    let ids = |cs: &[&conscience::Challenge]| cs.iter().map(|c| c.id.clone()).collect::<Vec<_>>();
2815    let rounded = |p: &[f64]| p.iter().map(|x| outcome::round_p(*x)).collect::<Vec<_>>();
2816    let (routed, basis) = (decided.map(|d| d.0), decided.map(|d| d.1));
2817    let shown: u32 = call.order.len().try_into().unwrap_or(u32::MAX);
2818    let place =
2819        |id: &str| call.order.iter().position(|x| x.id == id).or_else(|| call.boundary.iter().chain(call.tail).find(|(_, x)| x.id == id).map(|(at, _)| *at));
2820    outcome::Outcome {
2821        kind: outcome::GRADED.into(),
2822        session: session.into(),
2823        lessons: applied.iter().map(|(c, _)| c.id.clone()).collect(),
2824        ms: learn::now_ms(),
2825        repo: repo.into(),
2826        channel: "stop".into(),
2827        grades: applied.iter().map(|(_, g)| *g).collect(),
2828        ranks: (0..applied.len()).map(|r| u16::try_from(r).unwrap_or(u16::MAX)).collect(),
2829        applied: u16::try_from(applied.len()).unwrap_or(u16::MAX),
2830        chosen: routed.map(|r| ids(&r.ask)).unwrap_or_default(),
2831        explore: routed.is_some_and(|r| r.log.source == "explore"),
2832        p: routed.map(|r| rounded(&r.p)).unwrap_or_default(),
2833        check: check.into(),
2834        policy: routed.map_or_else(|| crate::select::policy().name(), |r| r.log.policy).into(),
2835        fit: routed.map_or_else(|| read.fit.key.clone(), |r| r.log.fit.clone()),
2836        // The summary the decision's `Ctx` read (`decide`), not the log at this append.
2837        seen: basis.map(|b| b.summary.key()).unwrap_or_default(),
2838        source: routed.map(|r| r.log.source).unwrap_or_default().into(),
2839        free: routed.map(|r| ids(&r.free)).unwrap_or_default(),
2840        pf: routed.map(|r| rounded(&r.pf)).unwrap_or_default(),
2841        placebo: routed.map(|r| ids(&r.placebo)).unwrap_or_default(),
2842        psi: routed.map_or(0.0, |r| r.log.psi),
2843        nofree: basis.map(|b| crate::select::nofree(applied, b.had)).unwrap_or_default(),
2844        // The question-count test's k and s (select.rs `Log::xk`, `xs`): None outside it.
2845        xk: routed.and_then(|r| r.log.xk),
2846        xs: routed.and_then(|r| r.log.xs),
2847        // The tolerance it was decided under (select.rs `Log::alpha`): None at the default.
2848        alpha: routed.and_then(|r| r.log.alpha),
2849        tail: call.tail.iter().map(|(_, c)| c.id.clone()).collect(),
2850        pt: call.pt,
2851        reason: basis.map(|b| b.why).unwrap_or_default().into(),
2852        eligible: call.eligible.try_into().unwrap_or(u32::MAX),
2853        shown,
2854        tokens: crate::select::shown_tokens(call.order),
2855        budget: call.budget.try_into().unwrap_or(u32::MAX),
2856        // The applies' places, then each tail lesson's (aligned with `tail`, outcome.rs `fill`).
2857        fill: applied
2858            .iter()
2859            .map(|(c, _)| place(&c.id))
2860            .chain(call.tail.iter().map(|(at, _)| Some(*at)))
2861            .map(|i| i.map_or(u16::MAX, |i| u16::try_from(i).unwrap_or(u16::MAX)))
2862            .collect(),
2863        model: call.model.into(),
2864        prompt_hash: call.prompt_hash.clone(),
2865        harness: std::env::var("BORG_HARNESS").unwrap_or_default(),
2866        version: env!("CARGO_PKG_VERSION").into(),
2867        dur_ms: call.dur_ms,
2868        failed: routed.is_none(),
2869        usd: call.usd,
2870        // What the boundary and the tail showed past the pool's tokens (the latency fit reads both).
2871        xtokens: crate::select::shown_tokens(&call.boundary.iter().chain(call.tail).map(|(_, c)| *c).collect::<Vec<_>>()),
2872        line: call.line.into(),
2873        moves: call.moves.clone(),
2874        whys: call.whys.clone(),
2875        ..Default::default()
2876    }
2877}
2878
2879/// One hook event, canonical already (harness `normalise`), for the drone `h` runs: what borg answers
2880/// (Claude-shaped), or None. One function per event.
2881pub fn dispatch(h: &dyn Harness, event: &str, input: &Value) -> Option<Value> {
2882    let session = input.get("session_id").and_then(Value::as_str).unwrap_or("unknown");
2883    match event {
2884        "prompt" => on_prompt(h, session, input),
2885        "post_tool_failure" => {
2886            capture_test(session, input, true);
2887            if h.caps().deny_capture {
2888                capture_rejection(h, session, input);
2889            }
2890            None
2891        }
2892        "end" => {
2893            on_end(h, session, input);
2894            None
2895        }
2896        "post_tool" => {
2897            capture_test(session, input, false);
2898            on_post_tool(h, session, input)
2899        }
2900        "stop" => on_stop(h, session, input),
2901        _ => None,
2902    }
2903}
2904
2905fn on_prompt(h: &dyn Harness, session: &str, input: &Value) -> Option<Value> {
2906    let prompt = input.get("prompt").and_then(Value::as_str).unwrap_or_default();
2907    append(session, prompt);
2908    if learning() {
2909        // Where the session started, once: its commits later get reasoning notes (notes.rs).
2910        crate::notes::remember_start(&crate::notes::state_dir(), session, input.get("cwd").and_then(Value::as_str).unwrap_or_default());
2911        crate::notes::mark(&crate::notes::state_dir(), session, "prompt");
2912    }
2913    // The engineer's own words, separately from the drone's touches (the queen's criterion 3).
2914    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(prompts_file(session)) {
2915        let _ = writeln!(f, "{prompt}");
2916    }
2917    // "borg no <reason>": the engineer's veto (veto.rs), never the session's task nor a correction.
2918    let vetoed = (learning() && crate::veto::enabled()).then(|| crate::veto::reason_in(prompt)).flatten();
2919    match &vetoed {
2920        Some(reason) => veto(h, session, input, reason),
2921        None => capture_prompt(h, session, input, prompt),
2922    }
2923    note_where(session, &repo_of(input.get("cwd").and_then(Value::as_str).unwrap_or_default()));
2924    let _ = std::fs::write(state_file(session).with_extension("lastprompt"), prompt);
2925    // A newer borg (update.rs): the daily release check, started detached when `update.json` is stale
2926    // (one read of it when fresh), and what the drone hears of it below.
2927    let checked = crate::update::maybe_check(learn::now_ms());
2928    // An adapter that will not wait for the answer (`"context": false`), or a harness whose model never
2929    // sees a prompt's context, gets none: what is handed over here is marked delivered and would be lost.
2930    let out = (h.caps().prompt_context && input.get("context") != Some(&Value::Bool(false))).then(|| {
2931        let cwd = input.get("cwd").and_then(Value::as_str).unwrap_or_default();
2932        let ask = queen_ask(h, session, cwd);
2933        // Once per session per state (`<session>.update`): a newer release, a member nothing syncs with,
2934        // a member on a newer borg; the routed store's mesh/ is read only when it exists.
2935        let update = crate::update::notice(checked.as_ref(), &learn::home(), &state_file(session).with_extension("update"));
2936        let extra: Vec<String> = [ask, vetoed.as_deref().map(crate::veto::render), update].into_iter().flatten().collect();
2937        free_context_with(h, "UserPromptSubmit", session, input, prompt, (!extra.is_empty()).then(|| extra.join("\n\n")))
2938    });
2939    // A new turn's work starts, after a free delivery here took the last turn's as its evidence.
2940    if harness::transcript(h.caps(), input).is_none() {
2941        let _ = std::fs::write(turn_file(session), "");
2942    }
2943    out.flatten()
2944}
2945
2946/// A shell command that ran tests leaves `fail <ms>` or `pass <ms>` in `<state>/<session>.tests`: the
2947/// regress `tests` signal (`notes/regress_signals.rs`) and `borg learn`'s `tested` credit, whether or not
2948/// the harness reports denials. A call that did not fail and shows no output (an adapter that sends
2949/// none) leaves nothing: its result is unknown, not a pass.
2950fn capture_test(session: &str, input: &Value, failed: bool) {
2951    if !learning() || input.get("tool_name").and_then(Value::as_str) != Some("Bash") {
2952        return;
2953    }
2954    let command = input.pointer("/tool_input/command").and_then(Value::as_str).unwrap_or_default();
2955    if !crate::notes::regress_signals::runs_tests(command) {
2956        return;
2957    }
2958    let output = crate::notes::regress_signals::tool_output(input);
2959    if !failed && output.trim().is_empty() {
2960        return;
2961    }
2962    if let Some(what) = crate::notes::regress_signals::test_outcome(command, &output, failed)
2963        && let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(state_file(session).with_extension("tests"))
2964    {
2965        let _ = writeln!(f, "{what} {}", learn::now_ms());
2966    }
2967}
2968
2969fn on_end(h: &dyn Harness, session: &str, input: &Value) {
2970    let _ = std::fs::remove_file(denied_file(session));
2971    if !learning() {
2972        return;
2973    }
2974    // The session's landed commits get a redacted reasoning note, in the background.
2975    crate::notes::remember_proposal(&crate::notes::state_dir(), session, &full_proposal(h, input));
2976    crate::notes::mark(&crate::notes::state_dir(), session, "stop");
2977    // What still waits in the queue was never delivered (a resumed session's delivery supersedes it);
2978    // then the end itself, where the selection summary counts the session's deliveries (outcome.rs
2979    // `ENDED`), whatever it logged before.
2980    note_never_delivered(session, input);
2981    let end = outcome::Outcome {
2982        kind: outcome::ENDED.into(),
2983        session: session.into(),
2984        ms: learn::now_ms(),
2985        repo: repo_of(input.get("cwd").and_then(Value::as_str).unwrap_or_default()),
2986        ..Default::default()
2987    };
2988    let _ = outcome::append(&learn::home().join("outcomes.jsonl"), &end);
2989    // Generalise this session's corrections in the background; never delay the drone. The notes job
2990    // starts it once its links are recorded, so learning sees their regret episodes.
2991    if !crate::notes::on_end(session, input)
2992        && let Ok(exe) = std::env::current_exe()
2993    {
2994        let _ = std::process::Command::new(exe)
2995            .args(["learn", "--session", session])
2996            .stdin(std::process::Stdio::null())
2997            .stdout(std::process::Stdio::null())
2998            .stderr(std::process::Stdio::null())
2999            .spawn();
3000    }
3001}
3002
3003fn on_post_tool(h: &dyn Harness, session: &str, input: &Value) -> Option<Value> {
3004    // The drone moved on from a denial: the next prompt is no longer the engineer's reason.
3005    let _ = std::fs::remove_file(denied_file(session));
3006    let tool = input.get("tool_name").and_then(Value::as_str).unwrap_or_default();
3007    let tool_input = input.get("tool_input").unwrap_or(&Value::Null);
3008    let touch = touched(tool_input, tool);
3009    append(session, &touch);
3010    // A path the drone touched says where it works; a command's free text does not.
3011    note_where(
3012        session,
3013        &if tool == "Bash" { bash_read_paths(tool_input.get("command").and_then(Value::as_str).unwrap_or_default()).join(" ") } else { touch.clone() },
3014    );
3015    // BORG_STOP_GATE=edits: this turn changed something, so its Stop asks the model.
3016    if prefilter::edits_gate() && prefilter::changes(tool, tool_input) {
3017        let _ = std::fs::OpenOptions::new().create(true).append(true).open(changes_file(session)).and_then(|mut f| writeln!(f, "{tool}"));
3018    }
3019    let edit = edit_text(tool, tool_input);
3020    if let Some(edit) = &edit
3021        && let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(edits_file(session))
3022    {
3023        let _ = writeln!(f, "{edit}");
3024    }
3025    // No transcript (omp, opencode): the turn's tool calls are the work so far a free delivery reads.
3026    if harness::transcript(h.caps(), input).is_none()
3027        && let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(turn_file(session))
3028    {
3029        let _ = writeln!(f, "[{tool}] {touch}");
3030        if let Some(edit) = &edit {
3031            let _ = writeln!(f, "{edit}");
3032        }
3033    }
3034    // Context only where the harness's model reads it after a tool call; otherwise it waits for the next
3035    // prompt or the stop. Never at a subagent's tool call: its context goes to the subagent, so what the
3036    // queue hands over there would be recorded delivered to a drone that never saw it.
3037    (h.caps().tool_context && !subagent(input)).then(|| free_context(h, "PostToolUse", session, input, &touch)).flatten()
3038}
3039
3040/// Whether the event is a subagent's (`agent_id`, as msg.rs `deliver` reads it): the session's id, but
3041/// Claude gives the hook's context to the subagent, not to the session's drone.
3042fn subagent(input: &Value) -> bool {
3043    input.get("agent_id").and_then(Value::as_str).is_some_and(|a| !a.is_empty())
3044}
3045
3046/// The stop holds the session lock (live.rs) for its whole check, so the session-end judge of a session
3047/// resumed meanwhile never judges the same deliveries beside it (`judge_session`). It never waits for
3048/// it: when the session-end judge holds the lock, the stop still checks the work, but leaves the
3049/// pending lessons to that judge (`Work::judge`) and asks none of them (`open_in`).
3050struct StopLock {
3051    budget: Option<std::time::Duration>,
3052    guard: Option<live::Guard>,
3053}
3054
3055impl StopLock {
3056    fn take(h: &dyn Harness, session: &str) -> StopLock {
3057        StopLock { budget: crate::expert_route::stop_budget(h), guard: live::acquire(&state_file(session), learn::now_ms()) }
3058    }
3059}
3060
3061fn on_stop(h: &dyn Harness, session: &str, input: &Value) -> Option<Value> {
3062    let started = std::time::Instant::now();
3063    let started_ms = learn::now_ms();
3064    if learning() {
3065        // The drone's latest proposal, for the reasoning note of what it commits (notes.rs).
3066        crate::notes::remember_proposal(&crate::notes::state_dir(), session, &full_proposal(h, input));
3067        crate::notes::mark(&crate::notes::state_dir(), session, "stop");
3068        // What this turn added to the harness's own transcript: its corrections and answers (sweep.rs).
3069        crate::sweep::turn_end(h, session, input);
3070    }
3071    let second_round = input.get("stop_hook_active").and_then(Value::as_bool) == Some(true);
3072    if std::env::var("BORG_CRITIC").as_deref() == Ok("generic") {
3073        return (!second_round).then(|| json!({"decision": "block", "reason": conscience::GENERIC}));
3074    }
3075    let (all, withheld) = loaded(session);
3076    let judged = judged(&all, withheld);
3077    // R1: the drone's answer to the last block's questions, at the first turn end after it: evidence
3078    // for the judge at the next model call (`pending`), not a label.
3079    note_answer(h, session, input);
3080    // BORG_STOP_GATE=edits: whether this turn changed anything; every Stop consumes the marker, so
3081    // the rework after a block does not count for the next turn.
3082    let changed = std::fs::remove_file(changes_file(session)).is_ok();
3083    // One round of self-questioning per turn (the harness's stop_hook_active); never loop.
3084    if second_round {
3085        return None;
3086    }
3087    stat(session, input, crate::stats::TURN_END, |_| {});
3088    if detached() {
3089        return stop_detached(h, session, input, &all, &judged, changed, started_ms);
3090    }
3091    let lock = StopLock::take(h, session);
3092    let state = std::fs::read_to_string(state_file(session)).unwrap_or_default();
3093    let proposal = full_proposal(h, input);
3094    let (done, mut challenges) = askable(session, &all, &judged, &state, &proposal, lock.guard.is_some(), false, None);
3095    // The pool's lines show each lesson's correction (whole.rs `line`, `A2`).
3096    conscience::with_corrections(&lessons_path(), &mut challenges);
3097    let message = input.get("last_assistant_message").and_then(Value::as_str).unwrap_or_default();
3098    let context = format!("{state}{message}");
3099    // Scopes at Stop: where the session works plus this turn's own message, not the whole history.
3100    let scope_text = format!("{}\n{message}", where_now(session));
3101    let check_id = crate::select::check_id(session, "stop", started_ms, std::process::id());
3102    // Folds the log's new tail and saves it, so the next check reads only what came after.
3103    let summary = crate::select::summary::refresh(&lessons_path());
3104    let work = Work {
3105        all: &all,
3106        judged: &judged,
3107        challenges: &challenges,
3108        done: &done,
3109        state: &state,
3110        context: &context,
3111        scope_text: &scope_text,
3112        message: &proposal,
3113        changed,
3114        judge: lock.guard.is_some(),
3115        budget: lock.budget,
3116        started,
3117        detached: None,
3118        edits: None,
3119        check: &check_id,
3120        summary: &summary,
3121    };
3122    let out = match stop_select(h, session, input, &work) {
3123        Ok(selected) => stop_ask(h, session, input, selected, &check_id),
3124        Err(out) => out,
3125    };
3126    drop(lock);
3127    out
3128}
3129
3130/// The lessons a stop's check may ask (`Work::challenges`) of `all`, and those done in the session
3131/// (`Work::done`), from its state file `state` and the drone's message `proposal`. `judge`: whether the
3132/// check judges the pending lessons (it holds the session lock, or no one does); without, it leaves the
3133/// open deliveries to the holder. `regraded`: a detached check's, which grades waiting asks again
3134/// (`waiting_in`).
3135#[allow(clippy::too_many_arguments)] // reason: the stop's view of the session, as each caller holds it
3136fn askable<'s>(
3137    session: &str,
3138    all: &[conscience::Challenge],
3139    judged: &[conscience::Challenge],
3140    state: &'s str,
3141    proposal: &str,
3142    judge: bool,
3143    regraded: bool,
3144    edits: Option<usize>,
3145) -> (std::collections::HashSet<&'s str>, Vec<conscience::Challenge>) {
3146    // A lesson is asked once per session: the drone answered it at that turn end, and asking it again
3147    // at every later one is noise (and a model call spent on it). R3b: a lesson delivered early or
3148    // handed over free (`#borg-early`, `#borg-queued`) is done once the drone acted on it; one it did
3149    // not act on may still be asked (`done_in`).
3150    let acted = early_acted(session, judged, proposal, edits);
3151    let only = std::env::var_os("BORG_ONLY").is_some();
3152    let done = if only { std::collections::HashSet::new() } else { done_in(state, &acted) };
3153    // What still waits is decided already: the free lessons of a check whose line no delivery took yet
3154    // (an earlier stop's at a harness without prompt context; `queued` hands them over). Out of this
3155    // stop's pool (`waiting_in`), so one check decides each.
3156    let queue = if only { String::new() } else { std::fs::read_to_string(queued_file(session)).unwrap_or_default() };
3157    // An ask a wake reserved (`reserve_asks`) is on its way: never graded again.
3158    let reserved = if only { String::new() } else { std::fs::read_to_string(reserved_file(session)).unwrap_or_default() };
3159    let waiting: std::collections::HashSet<&str> =
3160        if only { Default::default() } else { waiting_in(&queue, regraded).chain(waiting_in(&reserved, false)).collect() };
3161    // Without the lock, the session-end judge of a resumed session holds it and judges every open
3162    // delivery: a lesson it may be judging is not delivered again here, or its verdict on the earlier
3163    // delivery (`#borg-judged`, `END_JUDGED`) would land on this stop's question and settle it unjudged.
3164    let ending: std::collections::HashSet<&str> = if judge { Default::default() } else { open_in(state) };
3165    // Lessons of other systems (systems.rs) are neither shortlisted nor asked.
3166    let challenges = all
3167        .iter()
3168        .filter(|c| !done.contains(c.id.as_str()) && !waiting.contains(c.id.as_str()) && !ending.contains(c.id.as_str()) && !c.off_system)
3169        .cloned()
3170        .collect();
3171    (done, challenges)
3172}
3173
3174/// Whether this stop checks detached (check_async.rs): `BORG_ASYNC` on, and the stop's check is the
3175/// consequence model's graded call (`stop_select`, `whole::graded`). The evaluation modes (BORG_ONLY,
3176/// BORG_MATCH=bm25, BORG_STOP_GRADED=off) keep the stop's own check.
3177fn detached() -> bool {
3178    crate::check_async::on() && std::env::var_os("BORG_ONLY").is_none() && std::env::var("BORG_MATCH").as_deref() != Ok("bm25") && crate::whole::graded()
3179}
3180
3181/// Protocol: the state line a detached check's stop writes, `#borg-checking @<check>`: the turn ends
3182/// at the stop (prefilter.rs `MARKERS`), never at the check's later lines.
3183const CHECKING: &str = "#borg-checking";
3184
3185/// The stop of a detached check (check_async.rs): no session lock, no model call, no wait. The free gate
3186/// runs here (`gate`; a skip as at any stop, `skipped`); past it, the turn goes to a job file
3187/// (check_async.rs `Job`: the state file's length, the drone's message, where it works), the turn ends
3188/// (`CHECKING`) and `borg check --job` starts detached. Its asks reach the drone with the next prompt
3189/// or tool call (`queued`). The stop answers as a skip does: no block, the queen may ask.
3190fn stop_detached(
3191    h: &dyn Harness,
3192    session: &str,
3193    input: &Value,
3194    all: &[conscience::Challenge],
3195    judged: &[conscience::Challenge],
3196    changed: bool,
3197    started_ms: i64,
3198) -> Option<Value> {
3199    crate::check_async::sweep(session);
3200    reclaim_reserved(session);
3201    let ctx = state_file(session);
3202    let state = std::fs::read_to_string(&ctx).unwrap_or_default();
3203    // The .edits file's length with it: the turn the check judges ends here in both.
3204    let edits = std::fs::metadata(edits_file(session)).ok().and_then(|m| usize::try_from(m.len()).ok());
3205    let proposal = full_proposal(h, input);
3206    // The session-end judge holds the lock: the check leaves the open deliveries to it (`askable`).
3207    let judge = live::lock_state(&ctx, learn::now_ms(), live::stale_ms(), live::alive) != live::Lock::Held;
3208    let (_, challenges) = askable(session, all, judged, &state, &proposal, judge, true, None);
3209    let message = input.get("last_assistant_message").and_then(Value::as_str).unwrap_or_default();
3210    if let Some(why) = gate(&challenges, &format!("{state}{message}"), changed) {
3211        return skipped(h, session, input, why);
3212    }
3213    let cwd = input.get("cwd").and_then(Value::as_str).unwrap_or_default();
3214    let job = crate::check_async::Job {
3215        v: crate::check_async::JOB_VERSION,
3216        session: session.into(),
3217        check: crate::select::check_id(session, "stop", started_ms, std::process::id()),
3218        stop_ms: started_ms,
3219        harness: h.name().into(),
3220        cwd: cwd.into(),
3221        offset: state.len(),
3222        message: message.into(),
3223        proposal,
3224        changed,
3225        repo: repo_of(cwd),
3226        scope: format!("{}\n{message}", where_now(session)),
3227        edits,
3228    };
3229    let written = crate::check_async::write_job(&job);
3230    append(session, &format!("{CHECKING}{}", tag(&job.check)));
3231    // A job that cannot be written or started is counted (`borg status`), never lost unseen.
3232    if !written.is_ok_and(|path| crate::check_async::spawn(&path)) {
3233        crate::check_async::not_run(&job, crate::check_async::WHY_UNSTARTED);
3234    }
3235    queen_at_stop(h, session, input)
3236}
3237
3238/// A stop that asks nothing: the queen's question (queen.rs `stop`), when it has one.
3239fn queen_at_stop(h: &dyn Harness, session: &str, input: &Value) -> Option<Value> {
3240    let prompts = std::fs::read_to_string(prompts_file(session)).unwrap_or_default();
3241    crate::queen::stop(session, input, &|| harness::plan(h.caps(), input), &prompts, &[])
3242}
3243
3244/// A stop the free gate (or an empty pool) skipped: `#borg-skip <why>`, its stat, the queen's question.
3245fn skipped(h: &dyn Harness, session: &str, input: &Value, why: String) -> Option<Value> {
3246    append(session, &format!("#borg-skip {why}"));
3247    stat(session, input, crate::stats::GATE_SKIP, |e| e.why.clone_from(&why));
3248    queen_at_stop(h, session, input)
3249}
3250
3251/// Whether `session` ended since a detached check's stop: its presence reads ended (presence.rs
3252/// `state`: the end hook ran, or the harness process is gone), or the session-end judge ran past `offset`
3253/// in its state file.
3254pub(crate) fn gone(session: &str, offset: usize) -> bool {
3255    let ended = crate::presence::read(&learn::home(), session)
3256        .is_some_and(|p| crate::presence::state(&p, learn::now_ms(), crate::presence::alive) == crate::presence::State::Ended);
3257    ended || {
3258        let state = std::fs::read_to_string(state_file(session)).unwrap_or_default();
3259        state.get(offset..).unwrap_or_default().lines().any(|l| l.starts_with(END_JUDGED))
3260    }
3261}
3262
3263/// How often a detached check tries to take the session lock for its write (`run_job`), each after the
3264/// lock read free or stale. Guard: a lock file that cannot be written would otherwise be tried forever.
3265const LOCK_TRIES: usize = 3;
3266
3267/// A detached check's job (check_async.rs), in the process it started:the stop's check as the stop
3268/// would have run it, on the turn the job holds (the state file up to its `offset`, the drone's message,
3269/// where it worked), decided on channel live within its own limit (check_async.rs `call_limit`), the
3270/// model called without any lock. Its answer is written under the session lock (live.rs), taken only
3271/// then (`settle`, `Late`): the pending lessons it judges were open at the stop, and the session-end
3272/// judge (holding the lock at its start) keeps them. Nothing when the session ended before the call;
3273/// what it did otherwise (`Checked`, written under the lock, with the session's check count).
3274pub(crate) fn run_job(job: &crate::check_async::Job) -> Option<crate::check_async::Checked> {
3275    let session = job.session.as_str();
3276    let input = json!({"session_id": session, "cwd": job.cwd});
3277    if gone(session, job.offset) {
3278        crate::check_async::not_run(job, crate::check_async::WHY_GONE);
3279        return None;
3280    }
3281    let ctx = state_file(session);
3282    let now = std::fs::read_to_string(&ctx).unwrap_or_default();
3283    let state = now.get(..job.offset).unwrap_or(&now);
3284    let (all, withheld) = loaded(session);
3285    let judged = judged(&all, withheld);
3286    let judge = live::lock_state(&ctx, learn::now_ms(), live::stale_ms(), live::alive) != live::Lock::Held;
3287    let (done, mut challenges) = askable(session, &all, &judged, state, &job.proposal, judge, true, job.edits);
3288    conscience::with_corrections(&lessons_path(), &mut challenges);
3289    let context = format!("{state}{}", job.message);
3290    let summary = crate::select::summary::refresh(&lessons_path());
3291    let model = std::env::var("BORG_PREDICT_MODEL").unwrap_or_default();
3292    let work = Work {
3293        all: &all,
3294        judged: &judged,
3295        challenges: &challenges,
3296        done: &done,
3297        state,
3298        context: &context,
3299        scope_text: &job.scope,
3300        message: &job.proposal,
3301        changed: job.changed,
3302        judge,
3303        budget: None,
3304        started: std::time::Instant::now(),
3305        detached: Some(crate::check_async::call_limit(&model)),
3306        edits: job.edits,
3307        check: &job.check,
3308        summary: &summary,
3309    };
3310    let text = conscience::work_text(&context);
3311    let pool = match pool(session, &work, &text) {
3312        Ok(pool) => pool,
3313        Err(_) => {
3314            // Nothing to ask, none to judge: no call; the turn ended at the stop already.
3315            stat(session, &input, crate::stats::GATE_SKIP, |e| {
3316                e.why = "empty".into();
3317                e.mode = crate::check_async::MODE.into();
3318            });
3319            let mut out = crate::check_async::Checked::of(job);
3320            crate::check_async::finish(session, &mut out, false);
3321            return Some(out);
3322        }
3323    };
3324    let called = call(session, &input, &work, &pool, &text);
3325    let dur_ms = called.dur_ms;
3326    let (shown, cost) = (pool.shortlist.len(), called.pred.as_ref().map(|p| p.cost));
3327    // The only lock this check holds, for its write alone; a holder that renews it (the session-end
3328    // judge's batches) is waited for, one that stopped goes stale. A lock that cannot be written at all
3329    // is no reason to drop the answer: written without it, as `live::locked` appends.
3330    let wait = u64::try_from(live::stale_ms()).unwrap_or(u64::MAX);
3331    let guard = (0..LOCK_TRIES).find_map(|_| {
3332        while live::wait_free(&ctx, wait) == live::Lock::Held {}
3333        live::acquire(&ctx, learn::now_ms())
3334    });
3335    let late = Late { offset: job.offset, gone: gone(session, job.offset), proposal: &job.proposal, stop_ms: job.stop_ms };
3336    let s = settle(session, &input, &work, pool, called, Some(&late));
3337    let mut out = crate::check_async::Checked::of(job);
3338    out.dur_ms = dur_ms;
3339    out.superseded = s.superseded;
3340    match s.verdict {
3341        Verdict::Failed => out.failed = true,
3342        Verdict::Ask(asks) => {
3343            out.asks = asks;
3344            out.free = s.free;
3345        }
3346        Verdict::Skip(_) => {}
3347    }
3348    crate::check_async::finish(session, &mut out, true);
3349    // The pool log's scores (pools.rs) read the store and may load the embedder: after the lock.
3350    drop(guard);
3351    if let Some(p) = &s.pool {
3352        log_pool(session, &work, p);
3353    }
3354    note_check(session, &input, shown, dur_ms, cost, work.detached);
3355    Some(out)
3356}
3357
3358/// The lessons this stop asks; `Err` with the hook's answer when it asks none (a gate skip still lets
3359/// the queen ask).
3360fn stop_select<'a>(h: &dyn Harness, session: &str, input: &Value, w: &Work<'a>) -> Result<Vec<&'a conscience::Challenge>, Option<Value>> {
3361    match std::env::var("BORG_ONLY") {
3362        // BORG_ONLY: eval oracle, perfect retrieval = exactly these challenge ids.
3363        Ok(ids) => Ok(w.challenges.iter().filter(|c| ids.split(',').any(|i| i == c.id)).collect()),
3364        // Word matching alone always finds *some* lesson, so it asked at every turn end; the
3365        // consequence model may abstain, and it is the mechanism the sealed eval measured.
3366        Err(_) if std::env::var("BORG_MATCH").as_deref() != Ok("bm25") => stop_check(h, session, input, w),
3367        // R3b (conscience.rs `budget`): block only for a lesson near the best match not yet done;
3368        // then ask everything over the low bar, in-scope first.
3369        Err(_) => {
3370            let pool = conscience::budget_pool(&conscience::rank_scoped(w.all, &conscience::work_text(w.context), w.scope_text), w.done);
3371            let asked: Vec<&conscience::Challenge> = conscience::in_scope_first(conscience::explored(pool.iter().copied()))
3372                .into_iter()
3373                .filter_map(|c| w.challenges.iter().find(|x| x.id == c.id))
3374                .collect();
3375            append_p_of(session, &pool, &asked, w.check);
3376            Ok(asked)
3377        }
3378    }
3379}
3380
3381/// The consequence model's verdict at the stop.
3382fn stop_check<'a>(h: &dyn Harness, session: &str, input: &Value, w: &Work<'a>) -> Result<Vec<&'a conscience::Challenge>, Option<Value>> {
3383    match check(session, input, w) {
3384        Verdict::Skip(why) => Err(skipped(h, session, input, why)),
3385        Verdict::Failed => Err(None),
3386        Verdict::Ask(ids) => Ok(ids.iter().filter_map(|i| w.challenges.iter().find(|c| &c.id == i)).collect()),
3387    }
3388}
3389
3390/// `#borg-graded n=<applied> <id>:<grade>:<rank>,…`: a graded check's applies in the model's order, so
3391/// the `asked` outcome can carry each asked lesson's grade and rank (`graded_of`).
3392fn graded_line(applied: &[(&conscience::Challenge, u8)]) -> String {
3393    let items: Vec<String> = applied.iter().enumerate().map(|(r, (c, g))| format!("{}:{g}:{r}", c.id)).collect();
3394    format!("#borg-graded n={} {}", applied.len(), items.join(","))
3395}
3396
3397/// `#borg-p <id>=<p>,… @<check>`: the propensity (outcome.rs `p`) of each lesson a check chose to ask,
3398/// so every entry is above 0 and the latest entry per id is the check that asked it; the `asked`
3399/// outcome carries them (`tagged_p`, `propensity_in`). A graded check's `#borg-pf` line is the same for
3400/// what it hands over free. `check` is the check's id (select.rs `check_id`); none on an older
3401/// binary's line. Nothing when no lesson was chosen.
3402fn p_line<'a>(prefix: &str, chosen: impl IntoIterator<Item = (&'a str, f64)>, check: Option<&str>) -> Option<String> {
3403    let items: Vec<String> = chosen.into_iter().map(|(id, p)| format!("{id}={}", outcome::round_p(p))).collect();
3404    (!items.is_empty()).then(|| format!("{prefix} {}{}", items.join(","), tag(check.unwrap_or_default())))
3405}
3406
3407/// `p_line` for lessons `budget`/`widen` asked from `pool`: their chance of a slot under the
3408/// session's posterior (conscience.rs `propensity`). Then the pool itself, `#borg-pool
3409/// <id>=<base>[:s],… @<check>` in ranking order (`:s` a scoped lesson, base conscience.rs `pool_base`,
3410/// unrounded), so the `asked` record can carry it (`pool_in`) and `borg debug ope` can replay the draw. Both
3411/// lines carry the check that drew (its id, one draw per check), which ties the p of an asked lesson
3412/// to the pool its draw was over.
3413fn append_p_of(session: &str, pool: &[(f64, &conscience::Challenge)], asked: &[&conscience::Challenge], check: &str) {
3414    if asked.is_empty() {
3415        return;
3416    }
3417    let p = conscience::propensity(pool, session);
3418    let chosen: Vec<(&str, f64)> = asked.iter().filter_map(|c| p.get(&c.id).map(|p| (c.id.as_str(), *p))).collect();
3419    let items: Vec<String> =
3420        pool.iter().map(|(s, c)| format!("{}={}{}", c.id, conscience::pool_base(*s, c), if c.scope.is_empty() { "" } else { ":s" })).collect();
3421    let pool = format!("#borg-pool {}{}", items.join(","), tag(check));
3422    append(session, &p_line("#borg-p", chosen, Some(check)).map_or(pool.clone(), |p| format!("{p}\n{pool}")));
3423}
3424
3425/// Splits `line` into its items and its ` @<check>` tag (None without one).
3426fn tagged(line: &str) -> (&str, Option<&str>) {
3427    match line.rsplit_once(" @") {
3428        Some((items, check)) if !check.is_empty() && check.bytes().all(|b| b.is_ascii_hexdigit()) => (items, Some(check)),
3429        _ => (line, None),
3430    }
3431}
3432
3433/// The pool behind `ids` (`append_p_of`): the `#borg-pool` line of the check that wrote the latest
3434/// `#borg-p` entry of every one of them, as (ids, bases, scoped); empty when their latest entries come
3435/// from different checks, from a check without a pool, or from a line before checks were tagged (never
3436/// a guess: the p of one draw is never paired with another draw's pool).
3437fn pool_in(state: &str, ids: &[String]) -> (Vec<String>, Vec<f64>, Vec<bool>) {
3438    let parse = |line: &str| -> Option<(Vec<String>, Vec<f64>, Vec<bool>)> {
3439        let mut out = (Vec::new(), Vec::new(), Vec::new());
3440        for item in line.split(',') {
3441            let (id, rest) = item.rsplit_once('=')?;
3442            let (base, scoped) = match rest.strip_suffix(":s") {
3443                Some(b) => (b, true),
3444                None => (rest, false),
3445            };
3446            out.0.push(id.trim().to_string());
3447            out.1.push(base.trim().parse::<f64>().ok()?);
3448            out.2.push(scoped);
3449        }
3450        Some(out)
3451    };
3452    // The check of each id's latest p entry.
3453    let mut check: std::collections::HashMap<&str, Option<&str>> = std::collections::HashMap::new();
3454    for line in state.lines().filter_map(|l| l.strip_prefix("#borg-p ")) {
3455        let (items, tag) = tagged(line);
3456        for (id, _) in items.split(',').filter_map(|it| it.rsplit_once('=')) {
3457            check.insert(id.trim(), tag);
3458        }
3459    }
3460    let Some(Some(tag)) = ids.first().and_then(|i| check.get(i.as_str())).copied() else { return Default::default() };
3461    if !ids.iter().all(|i| check.get(i.as_str()).copied().flatten() == Some(tag)) {
3462        return Default::default();
3463    }
3464    state
3465        .lines()
3466        .rev()
3467        .filter_map(|l| l.strip_prefix("#borg-pool "))
3468        .map(tagged)
3469        .filter(|(_, t)| *t == Some(tag))
3470        .find_map(|(items, _)| parse(items).filter(|(pool, _, _)| ids.iter().all(|i| pool.contains(i))))
3471        .unwrap_or_default()
3472}
3473
3474/// For each of `ids`, its propensity in the state file's latest `#borg-p` entry for it, whatever check
3475/// wrote it (a line of an older binary has no tag); empty when any of them has none (unknown, never a
3476/// guess).
3477fn propensity_in(state: &str, ids: &[String]) -> Vec<f64> {
3478    let mut seen: std::collections::HashMap<&str, f64> = std::collections::HashMap::new();
3479    for line in state.lines().filter_map(|l| l.strip_prefix("#borg-p ")) {
3480        for (id, p) in tagged(line).0.split(',').filter_map(|it| it.rsplit_once('=')) {
3481            if let Ok(p) = p.trim().parse::<f64>() {
3482                seen.insert(id.trim(), p);
3483            }
3484        }
3485    }
3486    ids.iter().map(|i| seen.get(i.as_str()).copied()).collect::<Option<Vec<f64>>>().unwrap_or_default()
3487}
3488
3489/// For each of `ids`: its grade and rank in the session's latest graded check that graded it, and that
3490/// check's number of applies (the largest when the ids come from several checks); (0, 0) and 0 when none.
3491fn graded_of(session: &str, ids: &[String]) -> (Vec<u8>, Vec<u16>, u16) {
3492    let state = std::fs::read_to_string(state_file(session)).unwrap_or_default();
3493    let mut seen: std::collections::HashMap<String, (u8, u16, u16)> = std::collections::HashMap::new();
3494    for line in state.lines().filter_map(|l| l.strip_prefix("#borg-graded n=")) {
3495        let Some((n, items)) = line.split_once(' ') else { continue };
3496        let n: u16 = n.parse().unwrap_or(0);
3497        for it in items.split(',') {
3498            let mut p = it.rsplitn(3, ':');
3499            let (Some(r), Some(g), Some(id)) = (p.next(), p.next(), p.next()) else { continue };
3500            seen.insert(id.to_string(), (g.parse().unwrap_or(0), r.parse().unwrap_or(0), n));
3501        }
3502    }
3503    let got: Vec<(u8, u16, u16)> = ids.iter().map(|i| seen.get(i).copied().unwrap_or((0, 0, 0))).collect();
3504    (got.iter().map(|g| g.0).collect(), got.iter().map(|g| g.1).collect(), got.iter().map(|g| g.2).max().unwrap_or(0))
3505}
3506
3507/// Outcome tracking (outcome.rs) of `lessons` asked on `channel` (stop; live, a check's ask from before
3508/// 0.7.6), routed by `check`: `borg learn` later sees whether a correction followed. The `asked` record
3509/// names the check and carries its p and pool, so a verdict and the settle land on the check that decided
3510/// the lesson; then its `delivery` record. Nothing with learning off or without lessons.
3511fn record_asked(session: &str, input: &Value, lessons: &[&conscience::Challenge], channel: &str, check: &str) {
3512    if !learning() || lessons.is_empty() {
3513        return;
3514    }
3515    let state = std::fs::read_to_string(state_file(session)).unwrap_or_default();
3516    let ids: Vec<String> = lessons.iter().map(|c| c.id.clone()).collect();
3517    let members: Vec<&str> = ids.iter().map(String::as_str).collect();
3518    let (grades, ranks, applied) = graded_of(session, &ids);
3519    let p = if check.is_empty() { propensity_in(&state, &ids) } else { tagged_p(&state, "#borg-p ", check, &members) };
3520    let (pool, base, scoped) = pool_in(&state, &ids);
3521    let asked = outcome::Outcome {
3522        kind: outcome::ASKED.into(),
3523        session: session.into(),
3524        lessons: ids.clone(),
3525        ms: learn::now_ms(),
3526        repo: repo_of(input.get("cwd").and_then(Value::as_str).unwrap_or_default()),
3527        // The asked lessons' disagreements (outcome.rs `contradicted` reads them across the session's records).
3528        groups: conscience::disagreement_ids(lessons),
3529        channel: channel.into(),
3530        grades,
3531        ranks,
3532        applied,
3533        p,
3534        pool,
3535        base,
3536        scoped,
3537        check: check.into(),
3538        ..Default::default()
3539    };
3540    let _ = outcome::append(&learn::home().join("outcomes.jsonl"), &asked);
3541    record_route(session, input, check, ids, channel, outcome::ROUTE_ASKED, "");
3542}
3543
3544/// The stop asks `selected`, routed by its `check`: one block (the queen's, when it asks too),
3545/// `#borg-asked <ids> @<check>`, its `asked` record and its `delivery` record.
3546fn stop_ask(h: &dyn Harness, session: &str, input: &Value, mut lessons: Vec<&conscience::Challenge>, check: &str) -> Option<Value> {
3547    // Shown in an order stable within the session but different across sessions: no lesson always
3548    // sits first (position bias; memory.rs `shuffle`).
3549    crate::memory::shuffle(&mut lessons, session);
3550    let log = lessons.iter().map(|c| c.id.as_str()).collect::<Vec<_>>().join(",");
3551    // An empty `#borg-asked` still ends the turn (prefilter.rs `turn`): it carries no tag.
3552    let marker = if lessons.is_empty() { "#borg-asked ".to_string() } else { format!("#borg-asked {log}{}", tag(check)) };
3553    append(session, &marker);
3554    if !lessons.is_empty()
3555        && let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(edits_file(session))
3556    {
3557        // The blocked message is the work before the questions (R1 `pending`).
3558        let _ = writeln!(f, "{}\n{marker}", one_line(&full_proposal(h, input), EVIDENCE_CHARS));
3559    }
3560    // L4 (queen.rs): a rare, budgeted question to the engineer when the conscience had nothing
3561    // or its lessons contradict each other (its block then carries the lessons too).
3562    let prompts = std::fs::read_to_string(prompts_file(session)).unwrap_or_default();
3563    let queen = crate::queen::stop(session, input, &|| harness::plan(h.caps(), input), &prompts, &lessons);
3564    if lessons.is_empty() {
3565        return queen;
3566    }
3567    record_asked(session, input, &lessons, "stop", check);
3568    let out = queen.or_else(|| Some(json!({"decision": "block", "reason": conscience::render(&lessons)})));
3569    let chars = out.as_ref().and_then(|o| o["reason"].as_str()).map_or(0, str::len);
3570    log_cost(session, &format!("block questions={} chars={chars}", lessons.len()));
3571    stat(session, input, crate::stats::BLOCK, |e| {
3572        e.n = lessons.len() as u64;
3573        e.lessons = lessons.iter().map(|c| c.id.clone()).collect();
3574        e.channel = "stop".into();
3575    });
3576    out
3577}
3578
3579/// `borg hook <harness> <event>`, the event on stdin: routes the repo to its collective (an off repo
3580/// leaves no trace), records presence, then answers through steering, the experts and [`dispatch`],
3581/// with the messages from other sessions (msg.rs).
3582/// An unknown harness, like any failure here, lets the drone continue untouched.
3583pub fn main(name: &str, event: &str) {
3584    let Some(h) = harness::by_name(name) else { return };
3585    let mut raw = String::new();
3586    let _ = std::io::stdin().read_to_string(&mut raw);
3587    let mut input = h.normalise(serde_json::from_str(&raw).unwrap_or(Value::Null));
3588    // A pid only from a harness that sends its own: presence reads a gone one as an ended drone.
3589    if !h.caps().sends_pid
3590        && let Some(m) = input.as_object_mut()
3591    {
3592        m.shift_remove("pid");
3593    }
3594    // Which mind is this repo part of? Decided before anything is read or written: an off repo
3595    // (or a broken config) leaves no trace at all.
3596    let cwd = input.get("cwd").and_then(Value::as_str).unwrap_or_default();
3597    let cfg = collective::load();
3598    let route = match &cfg {
3599        Ok(cfg) => collective::route(cfg.as_ref(), cwd, &learn::home()),
3600        Err(_) => collective::Route::Off,
3601    };
3602    let collective::Route::Collective { home, name, .. } = route else { return };
3603    // SAFETY: single-threaded at this point; everything below (and the background `borg learn` it
3604    // spawns) inherits the collective's own store and scratch state.
3605    unsafe {
3606        std::env::set_var("BORG_HOME", &home);
3607        // borg's own model calls go through the drone's harness (runner.rs); `borg learn` reads it too.
3608        std::env::set_var("BORG_HARNESS", h.name());
3609        if std::env::var_os("BORG_STATE").is_none() {
3610            std::env::set_var("BORG_STATE", home.join("state"));
3611        }
3612    }
3613    // Who is working where (presence.rs, for `borg top`): one small atomic write per hook.
3614    let now = learn::now_ms();
3615    let session = input.get("session_id").and_then(Value::as_str).unwrap_or("unknown");
3616    // The turn of a session that listens for wakes (listen.rs): one stat of its opt-in for any other, at
3617    // a prompt, a tool call's start or end, a halted turn and the end. With steering's stat below, an idle
3618    // tool call costs two failed stats: the opt-in is per session, the admissions are per store.
3619    crate::listen::touch(&home, event, session, now);
3620    // A session's start (a hook of an older install) and a halted turn read and write nothing more.
3621    if matches!(event, "start" | "halted") {
3622        return;
3623    }
3624    // Not at pre_tool (every tool call): there only steering reads, once someone steered (steer.rs
3625    // `active`, the second stat).
3626    let me = match event {
3627        "pre_tool" => {
3628            if !crate::steer::active(&home) {
3629                return;
3630            }
3631            crate::presence::read(&home, session)
3632        }
3633        _ => crate::presence::touch(&home, &name, h.name(), event, &input, now),
3634    };
3635    // Steering (steer.rs) wraps the answer: a pause refuses tools and lets the stop wait.
3636    // At a stop, the routed collective's experts join the conscience's block (expert_route.rs).
3637    let started = me.map(|p| p.started);
3638    // A listening session's stop holds its wakes off until its answer is out (listen.rs).
3639    let stopping = crate::listen::stopping(&home, event, session, now);
3640    let out = crate::steer::around(&home, event, &input, started, now, || crate::expert_route::checkpoint(h, event, &input, || dispatch(h, event, &input)));
3641    // A listening session's stop starts its waiter, or asks for it (listen.rs).
3642    let out = crate::listen::after_stop(&home, h, stopping.as_ref(), &input, out);
3643    // An adapter that asked learns its wait budgets from the answer (harness.rs `Budgets`).
3644    let out = harness::with_budgets(h, &input, out);
3645    // Messages from other sessions (msg.rs) join a prompt's or tool call's context, each once.
3646    if let Some(out) = crate::msg::deliver(&home, h.caps(), event, &input, started, now, out) {
3647        println!("{}", for_harness(h, out));
3648    }
3649}
3650
3651/// The answer as `h` reads it: `"shown_ids"` (which messages it showed) is for an adapter, which drops a
3652/// held wake (listen.rs) once every message it names was shown; a harness that parses the answer itself
3653/// (Claude Code, Codex) gets only the keys it defines.
3654fn for_harness(h: &dyn Harness, mut out: Value) -> Value {
3655    if matches!(h.wiring(), harness::Wiring::CommandHooks { .. })
3656        && let Some(o) = out.as_object_mut()
3657    {
3658        o.shift_remove("shown_ids");
3659    }
3660    out
3661}
3662
3663#[cfg(test)]
3664mod tests {
3665    use super::*;
3666
3667    #[test]
3668    fn every_file_of_an_edit_survives_the_cut_of_its_text() {
3669        let long = "x".repeat(700);
3670        let input = json!({"file_path": "/w/a.rs", "new_string": long, "edits": [{"file_path": "/w/b.rs", "new_string": "more"}]});
3671        let text = edit_text("Edit", &input).unwrap();
3672        assert!(text.starts_with("/w/a.rs /w/b.rs x"), "{}", &text[..40]);
3673        assert_eq!(text.chars().count(), 600);
3674        let multi = json!({"file_path": "/w/a.rs", "edits": [{"old_string": "a", "new_string": "b"}, {"old_string": "c", "new_string": "d"}]});
3675        assert_eq!(edit_text("MultiEdit", &multi).as_deref(), Some("/w/a.rs b d"), "Claude's MultiEdit reads as before");
3676        assert_eq!(touched(&input, "Edit"), "/w/a.rs /w/b.rs");
3677    }
3678
3679    #[test]
3680    fn a_waiting_ask_is_graded_again_and_a_waiting_hand_over_is_not() {
3681        let c = "0123456789ab";
3682        let queue = format!("A,B @{c}\ninstead-next C @{c}\nlive D @{c}\nlive-next E\nplacebo F @{c}\n");
3683        assert_eq!(waiting_in(&queue, true).collect::<Vec<_>>(), ["A", "B", "C"], "a detached check's");
3684        assert_eq!(waiting_in(&queue, false).collect::<Vec<_>>(), ["A", "B", "C", "D", "E"], "a stop's own check's, as before");
3685    }
3686
3687    #[test]
3688    fn a_detached_checks_questions_name_the_turn_it_judged() {
3689        let edits = "#borg-proposal 0123456789ab first try\nwork\n#borg-proposal 0123456789ab I drop the token.\n#borg-proposal 00000000000f other\n";
3690        assert_eq!(proposal_of(edits, "0123456789ab"), Some("I drop the token."), "the latest line of that check");
3691        assert_eq!(proposal_of(edits, "0123456789"), None, "a whole id only");
3692        assert_eq!(proposal_of(edits, ""), None, "a line of an older binary has no check");
3693        let l = conscience::Challenge { id: "L1".into(), question: "Is the budget drained?".into(), ..Default::default() };
3694        let text = conscience::render_late(&[&l], &one_line("I drop\n the token.", LATE_EXCERPT));
3695        assert!(text.contains("your turn that ended with \"I drop the token.…\"") && text.contains("1. [L1] Is the budget drained?"), "{text}");
3696        assert!(text.contains("`<id>: applies`") && text.contains("already settled"), "{text}");
3697        assert!(conscience::render_late(&[&l], "").contains("After your last turn"));
3698    }
3699
3700    #[test]
3701    fn propensity_in_takes_the_latest_entry_and_nothing_when_one_is_missing() {
3702        let state = "#borg-p A=1,B=0.25\n#borg-asked A,B\n#borg-p B=0.5,C=0.1\n";
3703        let ids = |v: &[&str]| v.iter().map(|x| x.to_string()).collect::<Vec<_>>();
3704        assert_eq!(propensity_in(state, &ids(&["B", "A"])), [0.5, 1.0]);
3705        assert_eq!(propensity_in(state, &ids(&["C"])), [0.1]);
3706        assert!(propensity_in(state, &ids(&["A", "D"])).is_empty(), "D unknown: no guess");
3707        assert!(propensity_in("", &ids(&["A"])).is_empty());
3708    }
3709
3710    #[test]
3711    fn pool_in_takes_the_pool_of_the_check_that_logged_each_p() {
3712        let state = "#borg-p A=0.6,B=0.4 @c1\n#borg-pool A=1.5:s,B=0.25,C=2 @c1\n#borg-asked A,B\n#borg-p B=0.5 @c2\n#borg-pool B=0.5,D=0.125:s @c2\n";
3713        let ids = |v: &[&str]| v.iter().map(|x| x.to_string()).collect::<Vec<_>>();
3714        assert_eq!(pool_in(state, &ids(&["B"])), (ids(&["B", "D"]), vec![0.5, 0.125], vec![false, true]), "B's latest p is check 2's");
3715        assert_eq!(pool_in(state, &ids(&["A"])), (ids(&["A", "B", "C"]), vec![1.5, 0.25, 2.0], vec![true, false, false]));
3716        assert_eq!(pool_in(state, &ids(&["A", "B"])), Default::default(), "A's p from check 1, B's from check 2: no pairing");
3717        assert_eq!(pool_in("", &ids(&["A"])), Default::default());
3718        assert_eq!(pool_in(state, &[]), Default::default());
3719        // A graded check's p has no pool: a later pool that happens to hold the lesson is not its.
3720        let graded = "#borg-p A=1\n#borg-p B=0.5 @c3\n#borg-pool A=1,B=1 @c3\n";
3721        assert_eq!(pool_in(graded, &ids(&["A"])), Default::default());
3722        assert_eq!(pool_in(graded, &ids(&["B"])).0, ids(&["A", "B"]));
3723        // A state file from before checks were tagged: no pool.
3724        assert_eq!(pool_in("#borg-p A=1\n#borg-pool A=1,B=2\n", &ids(&["A"])), Default::default());
3725    }
3726
3727    #[test]
3728    fn a_queue_line_names_its_route_its_ids_and_its_check() {
3729        let c = "0123456789ab";
3730        assert_eq!(queue_line(&format!("A,B @{c}")), (Queued::Free, vec!["A", "B"], c));
3731        assert_eq!(queue_line(&format!("placebo X,Y @{c}")), (Queued::Placebo, vec!["X", "Y"], c));
3732        assert_eq!(queue_line(&format!("instead G0,G1 @{c}")), (Queued::Instead, vec!["G0", "G1"], c));
3733        assert_eq!(queue_line(&format!("instead-next G2 @{c}")), (Queued::Instead, vec!["G2"], c));
3734        // An older binary's lines: no check.
3735        assert_eq!(queue_line("A,B"), (Queued::Free, vec!["A", "B"], ""));
3736        // A live check's lines, written before 0.7.6: handed over in place of a question.
3737        assert_eq!(queue_line(&format!("live A @{c}")), (Queued::Ask, vec!["A"], c));
3738        assert_eq!(queue_line("live A"), (Queued::Ask, vec!["A"], ""));
3739        assert_eq!(queue_line(&format!("live-next A,B @{c}")), (Queued::Ask, vec!["A", "B"], c));
3740        assert_eq!(listed(&format!("#borg-asked A,B @{c}\n#borg-asked C\n"), "#borg-asked "), ["A", "B", "C"].into_iter().collect());
3741    }
3742
3743    #[test]
3744    fn a_tagged_propensity_is_its_checks_never_another_checks() {
3745        let state = "#borg-pf A=1,B=0.5 @c1\n#borg-pf A=0.25 @c2\n#borg-p A=0.75 @c1\n#borg-pf B=0.9\n";
3746        assert_eq!(tagged_p(state, "#borg-pf ", "c1", &["A", "B"]), [1.0, 0.5]);
3747        assert_eq!(tagged_p(state, "#borg-pf ", "c2", &["A"]), [0.25]);
3748        assert!(tagged_p(state, "#borg-pf ", "c2", &["A", "B"]).is_empty(), "B has none at c2: no guess");
3749        assert_eq!(tagged_p(state, "#borg-p ", "c1", &["A"]), [0.75], "#borg-p and #borg-pf are separate lines");
3750        assert_eq!(p_line("#borg-pf", [("A", 1.0), ("B", 0.33333)], Some("c1")).as_deref(), Some("#borg-pf A=1,B=0.3333 @c1"));
3751        assert_eq!(p_line("#borg-p", [], Some("c1")), None);
3752    }
3753
3754    #[test]
3755    fn unjudged_says_why_each_delivered_lesson_has_no_verdict() {
3756        let c = "0123456789ab";
3757        let state = format!(
3758            "#borg-asked A,B,C,D @{c}\n#borg-judged A=pending\n#borg-unjudged B why=slots\n#borg-judged A=pending\n#borg-unjudged D why=failed\n\
3759             #borg-queued E @{c}\n#borg-judged B=pending\n#borg-asked D\n"
3760        );
3761        let why = unjudged(&state);
3762        assert_eq!(why["A"], outcome::UNJUDGED_TRIES, "two unclear tries");
3763        assert_eq!(why["B"], outcome::UNJUDGED_SLOTS, "one try is not all of them; the last call left it out");
3764        assert_eq!(why["C"], outcome::UNJUDGED_NO_CALL);
3765        assert_eq!(why["D"], outcome::UNJUDGED_NO_CALL, "delivered again after its failed call");
3766        assert_eq!(why["E"], outcome::UNJUDGED_NO_CALL);
3767    }
3768
3769    #[test]
3770    fn the_declaration_coin_is_fixed_per_check_and_lesson_and_fair() {
3771        assert_eq!(stripped("0123456789ab", "L1"), stripped("0123456789ab", "L1"));
3772        let heads = (0..1000).filter(|i| stripped("0123456789ab", &format!("L{i}"))).count();
3773        assert!((400..600).contains(&heads), "{heads} of 1000");
3774        let checks = (0..1000).filter(|i| stripped(&format!("{i:012x}"), "L1")).count();
3775        assert!((400..600).contains(&checks), "{checks} of 1000");
3776    }
3777
3778    /// Judge v2's evidence: a free delivery and a placebo get the work so far before them and the work
3779    /// since after, as a question does; each lesson carries the check that routed it (a stop's block from
3780    /// before 0.7.6 may carry a live check's waiting ask); a stripped declaration is gone from the shared
3781    /// evidence.
3782    #[test]
3783    fn pending_gives_every_route_the_same_evidence_and_its_own_check() {
3784        let session = format!("pending-v2-{}", std::process::id());
3785        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
3786        let all = [c("A"), c("B"), c("C"), c("D"), c("E")];
3787        let edits = "Plan: the blocked message\n#borg-asked A,B @0000000000c1\n#borg-routed B=0000000000c2\n#borg-answer A: applies. B: skip.\n\
3788                     edit one\n#borg-work the work so far\n#borg-queued C @0000000000c3\n#borg-placebo D @0000000000c4\nedit two\n\
3789                     #borg-work later work\n#borg-asked E @0000000000c5\n";
3790        let _ = std::fs::write(edits_file(&session), edits);
3791        let state: String = edits.lines().filter(|l| DELIVERED.iter().any(|m| l.starts_with(m))).map(|l| format!("{l}\n")).collect();
3792        let j = pending(&session, &state, &all, "the final message", When::Stop, "00000000cafe");
3793        let got: Vec<(&str, &str, &str)> = j.pending.iter().map(|p| (p.lesson.id.as_str(), p.channel, p.check.as_str())).collect();
3794        assert_eq!(
3795            got,
3796            [
3797                ("A", "stop", "0000000000c1"),
3798                ("B", "stop", "0000000000c2"),
3799                ("C", "queued", "0000000000c3"),
3800                ("D", outcome::PLACEBO, "0000000000c4"),
3801                ("E", "stop", "0000000000c5")
3802            ]
3803        );
3804        let p = |id: &str| j.pending.iter().find(|p| p.lesson.id == id).unwrap();
3805        assert_eq!(
3806            (p("A").before.as_str(), p("C").before.as_str(), p("D").before.as_str()),
3807            ("Plan: the blocked message", "the work so far", "the work so far")
3808        );
3809        assert_eq!(p("C").after, "work since: edit two\nthis turn: the final message");
3810        assert_eq!(p("D").after, p("C").after, "a placebo is judged as a free delivery");
3811        assert!(p("A").after.starts_with("work since: ") && p("A").after.contains("edit one | edit two"), "{}", p("A").after);
3812        assert_eq!(p("A").after, p("B").after, "delivered together, one shared evidence");
3813        for (id, word) in [("A", "applies"), ("B", "skip")] {
3814            let x = p(id);
3815            assert_eq!(x.stripped, stripped(&x.check, id));
3816            match x.stripped {
3817                true => assert!(x.declared.is_none() && !x.after.contains(&format!("{id}: {word}")), "{x:?}"),
3818                false => assert!(x.declared.is_some() && x.after.contains(&format!("{id}: {word}")), "{x:?}"),
3819            }
3820        }
3821        assert!(!p("C").stripped && !p("D").stripped, "only questions are stripped");
3822        assert!(j.question && j.crowded.is_empty());
3823        let _ = std::fs::remove_file(edits_file(&session));
3824    }
3825
3826    /// History (live mode, before 0.7.6): a lesson handed over free at check A, then asked live at check B,
3827    /// in a session that runs across the upgrade. The live question still reads as channel live and is
3828    /// the lesson's latest delivery: the lesson is judged once, on the question, answered or not.
3829    #[test]
3830    fn a_live_question_from_before_the_upgrade_is_still_its_lessons_latest_delivery() {
3831        let session = format!("pending-later-{}", std::process::id());
3832        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
3833        let all = [c("L"), c("M")];
3834        let (a, b) = ("00000000000a", "00000000000b");
3835        let asked = format!("#borg-work so far\n#borg-queued L,M @{a}\nedit one\n#borg-work more\n#borg-asked L @{b}\n#borg-live L ms=1\nedit two\n");
3836        let state_of = |edits: &str| edits.lines().filter(|l| DELIVERED.iter().any(|m| l.starts_with(m))).map(|l| format!("{l}\n")).collect::<String>();
3837        let answered = format!("{asked}#borg-answer L: applies\n");
3838        for edits in [&asked, &answered] {
3839            let _ = std::fs::write(edits_file(&session), edits);
3840            for when in [When::Stop, When::End] {
3841                let j = pending(&session, &state_of(edits), &all, "", when, "00000000cafe");
3842                let got: Vec<(String, &str, String)> = j.pending.iter().map(|p| (p.lesson.id.clone(), p.channel, p.check.clone())).collect();
3843                assert_eq!(got, [("M".to_string(), "queued", a.to_string()), ("L".to_string(), "live", b.to_string())], "{when:?}: {edits}");
3844            }
3845        }
3846        let _ = std::fs::remove_file(edits_file(&session));
3847    }
3848
3849    /// Judge timing (0.7.5): a stop's judge call judges a free delivery, an early one and a placebo, work
3850    /// after it or not, and so does the session-end judge.
3851    #[test]
3852    fn a_free_delivery_or_placebo_is_judged_at_the_turn_end() {
3853        let session = format!("pending-work-{}", std::process::id());
3854        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
3855        let all = [c("A"), c("C"), c("D"), c("E")];
3856        let state_of = |edits: &str| edits.lines().filter(|l| DELIVERED.iter().any(|m| l.starts_with(m))).map(|l| format!("{l}\n")).collect::<String>();
3857        let ids = |j: &Judging| j.pending.iter().map(|p| p.lesson.id.clone()).collect::<Vec<_>>();
3858        let fresh = "Plan: blocked\n#borg-asked A @0000000000c1\n#borg-answer A: applies\n#borg-work so far\n#borg-queued C @0000000000c3\n#borg-placebo D @0000000000c4\n";
3859        for edits in [fresh.to_string(), format!("{fresh}edit one\n"), format!("{fresh}edit one\n#borg-work later\n#borg-early E @0000000000c5\nedit two\n")] {
3860            let _ = std::fs::write(edits_file(&session), &edits);
3861            let state = state_of(&edits);
3862            let early = edits.contains("#borg-early ");
3863            for when in [When::Stop, When::End] {
3864                let want: &[&str] = if early { &["A", "C", "D", "E"] } else { &["A", "C", "D"] };
3865                assert_eq!(ids(&pending(&session, &state, &all, "", when, "00000000cafe")), want, "{when:?}: {edits}");
3866            }
3867        }
3868        let _ = std::fs::remove_file(edits_file(&session));
3869    }
3870
3871    /// The session-end judge (`When::End`): every delivery no verdict settled, a question not answered
3872    /// and a lesson whose check tries ran out included, never one a verdict settled; what it was shown
3873    /// (`END_JUDGED`) is settled for every later call, until the lesson is delivered again, and settles
3874    /// at `borg learn` as having used its tries.
3875    #[test]
3876    fn the_session_end_judge_never_outlives_its_lock() {
3877        let (end, stale) = (end_timeout(), live::stale_ms());
3878        assert!(i64::try_from(end.as_millis()).unwrap() < stale, "{end:?} against a lock stale after {stale} ms");
3879        assert!(end <= learn::learn_timeout());
3880    }
3881
3882    #[test]
3883    fn the_session_end_judge_takes_every_open_delivery_once() {
3884        let session = format!("pending-end-{}", std::process::id());
3885        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
3886        let all = [c("A"), c("B"), c("C"), c("D"), c("Q")];
3887        let ids = |j: &Judging| j.pending.iter().map(|p| p.lesson.id.clone()).collect::<Vec<_>>();
3888        let edits = "Plan\n#borg-asked A,B @0000000000c1\n#borg-answer A: applies\nedit\n#borg-work w\n#borg-queued C,D @0000000000c3\nedit two\nPlan two\n#borg-asked Q @0000000000c4\n";
3889        let _ = std::fs::write(edits_file(&session), edits);
3890        let delivered: String = edits.lines().filter(|l| DELIVERED.iter().any(|m| l.starts_with(m))).map(|l| format!("{l}\n")).collect();
3891        // A changed, B unclear twice (tries spent), C unclear once, D not judged, Q never answered.
3892        let state = format!("{delivered}#borg-judged A=changed\n#borg-judged B=pending\n#borg-judged C=pending\n#borg-judged B=pending\n");
3893        assert_eq!(ids(&pending(&session, &state, &all, "", When::Stop, "00000000cafe")), ["C", "D", "Q"], "a check: B's tries are spent");
3894        assert_eq!(ids(&pending(&session, &state, &all, "", When::End, "00000000cafe")), ["B", "C", "D", "Q"], "the end: every open one");
3895        let ended = format!("{state}#borg-judged B=pending\n#borg-judged C=changed\n{END_JUDGED}B,C,D,Q\n");
3896        for when in [When::Stop, When::End] {
3897            assert!(pending(&session, &ended, &all, "", when, "00000000cafe").pending.is_empty(), "{when:?}: judged at the end once");
3898        }
3899        let why = unjudged(&ended);
3900        assert_eq!((why["B"], why["D"], why["Q"]), (outcome::UNJUDGED_TRIES, outcome::UNJUDGED_TRIES, outcome::UNJUDGED_TRIES));
3901        // Delivered again after the end (a resumed session): open again.
3902        let again = format!("{ended}#borg-queued D @0000000000c5\n");
3903        let _ = std::fs::write(edits_file(&session), format!("{edits}#borg-queued D @0000000000c5\n"));
3904        assert_eq!(ids(&pending(&session, &again, &all, "", When::End, "00000000cafe")), ["D"]);
3905        assert_eq!(unjudged(&again)["D"], outcome::UNJUDGED_NO_CALL);
3906        let _ = std::fs::remove_file(edits_file(&session));
3907    }
3908
3909    /// A source of the session's work by its own text, `marks` where each delivery stood.
3910    fn source(label: &'static str, text: &str, marks: &[(usize, &str)]) -> Source {
3911        Source { label, text: text.chars().collect(), marks: marks.iter().map(|(at, id)| (*at, vec![id.to_string()])).collect() }
3912    }
3913
3914    /// The session's work for the session-end judge: each source labelled, whole when all fit; else the
3915    /// budget, framing included, shared so a short source leaves the rest to the longer ones, each cut
3916    /// its newest characters (`newest`, and wherever a source has no delivery in it).
3917    #[test]
3918    fn the_session_work_shares_its_budget_and_is_read_whole_when_it_fits() {
3919        let parts = [source("short", "abc", &[]), source("long", "0123456789", &[]), source("none", "", &[])];
3920        let none = [None, None, None];
3921        let whole = "### short\nabc\n### long\n0123456789\n";
3922        assert_eq!(within(&parts, &none, 100, Cut::After), (whole.to_string(), false));
3923        assert_eq!(within(&parts, &none, whole.chars().count(), Cut::After), (whole.to_string(), false), "exactly fits: whole");
3924        // Framing: `### short\n` + `\n` (11) and `### long\n` + `\n` (10), two marks each: 25 of 33 leave 8.
3925        assert_eq!(within(&parts, &none, 33, Cut::Newest), ("### short\nabc\n### long\n…56789\n".to_string(), true), "short keeps its 3, long the other 5");
3926        assert_eq!(within(&parts, &none, 29, Cut::After), ("### short\n…bc\n### long\n…89\n".to_string(), true), "two each when neither fits its share");
3927        for budget in 0..60 {
3928            for cut in [Cut::Newest, Cut::After, Cut::Ends] {
3929                let (text, _) = within(&parts, &[None, Some(3), None], budget, cut);
3930                assert!(text.chars().count() <= budget, "{budget} {cut:?}: {text:?}");
3931            }
3932        }
3933    }
3934
3935    /// A long session's work cut for a batch: `newest` its end; `after` from the batch's earliest
3936    /// delivery on, its start first; `ends` half from there, half the end; a source without the batch's
3937    /// delivery in it, and every cut when what follows the delivery fits, the newest.
3938    #[test]
3939    fn a_long_sessions_work_is_cut_at_the_batchs_earliest_delivery() {
3940        let text: Vec<char> = "0123456789abcdefghij".chars().collect();
3941        assert_eq!(Cut::Newest.window(&text, 6, Some(4)), "…efghij");
3942        assert_eq!(Cut::After.window(&text, 6, Some(4)), "…456789…", "from the delivery on, its start first");
3943        assert_eq!(Cut::After.window(&text, 6, Some(0)), "012345…");
3944        assert_eq!(Cut::Ends.window(&text, 6, Some(4)), "…456…hij", "half after the delivery, half the end");
3945        assert_eq!(Cut::After.window(&text, 6, Some(16)), "…efghij", "what follows fits: the newest");
3946        assert_eq!(Cut::Ends.window(&text, 6, Some(15)), "…efghij", "the halves meet: the newest");
3947        for cut in [Cut::Newest, Cut::After, Cut::Ends] {
3948            assert_eq!(cut.window(&text, 6, None), "…efghij", "{cut:?}: no delivery in it");
3949        }
3950        // Within a batch's budget: the edits part from the batch's delivery on, the messages their end.
3951        let parts = [source("edits", "0123456789abcdefghij", &[(4, "L")]), source("said", "ABCDEFGHIJKLMNOPQRST", &[])];
3952        let (work, cut) = within(&parts, &[parts[0].anchor(&["L"]), parts[1].anchor(&["L"])], 34, Cut::After);
3953        assert!(cut);
3954        assert_eq!(work, "### edits\n…4567…\n### said\n…PQRST\n");
3955    }
3956
3957    /// Where each delivery stood in a source: its latest marker naming the lesson (the delivery it is
3958    /// judged on), the earliest of a batch's; the markers and bookkeeping are no work, a stripped
3959    /// question's declaration is gone from it, an answer to a question is.
3960    #[test]
3961    fn a_source_marks_where_each_delivery_stood_without_stripped_declarations() {
3962        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
3963        let (a, f) = (c("A"), c("F"));
3964        let p = |lesson, stripped| predict::Pending {
3965            lesson,
3966            before: String::new(),
3967            after: String::new(),
3968            declared: None,
3969            check: String::new(),
3970            channel: "stop",
3971            stripped,
3972        };
3973        let pend = [p(&a, false), p(&f, true)];
3974        let edits = "one\n#borg-work w\n#borg-asked A,F @0000000000c1\n#borg-answer A: applies. F: applies.\ntwo\n#borg-queued A @0000000000c2\nthree\n";
3975        let s = Source::of("edits", edits, |l| l.strip_prefix("#borg-answer ").or_else(|| (!l.starts_with("#borg-")).then_some(l)), &pend);
3976        let text: String = s.text.iter().collect();
3977        assert_eq!(text, "one\nA: applies. .\ntwo\nthree");
3978        assert_eq!((s.anchor(&["F"]), s.anchor(&["A"]), s.anchor(&["A", "F"]), s.anchor(&["Z"])), (Some(4), Some(22), Some(4), None));
3979        assert!(Source::of("blank", "#borg-asked A\n \n", Some, &pend).text.is_empty(), "only whitespace: empty");
3980    }
3981
3982    /// A session-end pass's budget is the judge's measured length (judge_length.rs), or the step above
3983    /// on the chance the estimate gives while that step is open: drawn per pass from its seed, the
3984    /// chance and the draw logged (`draw`); the same seed draws the same.
3985    #[test]
3986    fn a_pass_explores_the_step_above_on_its_logged_chance() {
3987        let d = std::env::temp_dir().join(format!("borg-end-budget-{}", std::process::id()));
3988        let _ = std::fs::remove_dir_all(&d);
3989        std::fs::create_dir_all(&d).unwrap();
3990        let draws: Vec<EndBudget> = (0..500).map(|i| end_budget(&d, "m:x", &format!("check-{i}"))).collect();
3991        let psi = crate::judge_length::EXPLORE_SHARE;
3992        for b in &draws {
3993            let explored = b.draw.expect("drawn, logged").1;
3994            assert_eq!(*b, EndBudget { chars: if explored { 108_000 } else { END_START_BUDGET }, draw: Some((psi, explored)) });
3995        }
3996        let share = draws.iter().filter(|b| b.draw.is_some_and(|d| d.1)).count() as f64 / draws.len() as f64;
3997        assert!((share - psi).abs() < 0.06, "explored {share} of passes at chance {psi}");
3998        assert_eq!(end_budget(&d, "m:x", "check-7"), draws[7], "the same seed, the same draw");
3999        // The step above decided as failing: explored at the floor's chance, logged.
4000        let line = |chars: usize, n: u64, a: u64| {
4001            format!("judge call=1 pending={n} resolved={a} chars={chars} verdicts=changed:{a},unclear:{} model=m:x at=end\n", n - a).repeat(10)
4002        };
4003        std::fs::write(d.join("s.cost"), line(30_000, 40, 20) + &line(100_000, 40, 5)).unwrap();
4004        let floor = crate::judge_length::FAILED_EXPLORE_SHARE;
4005        let draws: Vec<EndBudget> = (0..2000).map(|i| end_budget(&d, "m:x", &format!("check-{i}"))).collect();
4006        for b in &draws {
4007            let explored = b.draw.expect("drawn, logged").1;
4008            assert_eq!(*b, EndBudget { chars: if explored { 108_000 } else { END_START_BUDGET }, draw: Some((floor, explored)) });
4009        }
4010        let share = draws.iter().filter(|b| b.draw.is_some_and(|d| d.1)).count() as f64 / draws.len() as f64;
4011        assert!((share - floor).abs() < 0.02, "explored {share} of passes at chance {floor}");
4012        let _ = std::fs::remove_dir_all(&d);
4013    }
4014
4015    /// The session-end judge's batches: every open lesson in exactly one, in delivery order; each batch's
4016    /// lessons within half the budget (one alone may exceed it), and its whole prompt, with as much of a
4017    /// long session's work as is left, within the budget.
4018    #[test]
4019    fn every_lesson_lands_in_one_batch_whose_prompt_stays_within_the_budget() {
4020        let lessons: Vec<conscience::Challenge> = (0..278)
4021            .map(|i| conscience::Challenge {
4022                id: format!("L{i:03}"),
4023                question: format!("Question {i} {}", "q".repeat(i * 7 % 400)),
4024                consequence: format!("consequence {i}"),
4025                ..Default::default()
4026            })
4027            .collect();
4028        let pend: Vec<predict::Pending> = lessons
4029            .iter()
4030            .enumerate()
4031            .map(|(i, l)| predict::Pending {
4032                lesson: l,
4033                // Lessons delivered together share their evidence.
4034                before: format!("before {} {}", i / 3, "b".repeat(i * 13 % 1300)),
4035                after: format!("after {} {}", i / 3, "a".repeat(i * 31 % 4000)),
4036                declared: None,
4037                check: String::new(),
4038                channel: "queued",
4039                stripped: false,
4040            })
4041            .collect();
4042        let long: String = (0..4000).map(|i| format!("edit {i:04} of the long session\n")).collect();
4043        let marks: Vec<(usize, &str)> = lessons.iter().enumerate().map(|(i, l)| (i * 100, l.id.as_str())).collect();
4044        let sources = [source("prompts", &"p".repeat(30_000), &[]), source("edits", &long, &marks), source("said", "", &[])];
4045        for budget in [END_START_BUDGET, 20_000, 200_000] {
4046            let groups = batches(&pend, budget);
4047            assert!(groups.len() > 1, "{budget}: {groups:?}");
4048            assert_eq!(groups.iter().flat_map(Clone::clone).collect::<Vec<_>>(), (0..pend.len()).collect::<Vec<_>>(), "each once, in order");
4049            for g in &groups {
4050                let batch = &pend[g.clone()];
4051                assert!(batch.len() == 1 || predict::judge_section(batch).chars().count() <= budget / END_LESSONS_SHARE, "{budget}: {g:?}");
4052                let fixed = predict::judge_prompt("", batch).chars().count();
4053                let ids: Vec<&str> = batch.iter().map(|p| p.lesson.id.as_str()).collect();
4054                let anchors: Vec<Option<usize>> = sources.iter().map(|s| s.anchor(&ids)).collect();
4055                let (work, cut) = within(&sources, &anchors, budget.saturating_sub(fixed), Cut::After);
4056                assert!(cut, "a long session is cut");
4057                assert!(predict::judge_prompt(&work, batch).chars().count() <= budget, "{budget}: {g:?}");
4058                // `after`: the edits part starts at the batch's first delivery (or at the newest share).
4059                let first = g.start * 100;
4060                let body = work.split("### edits\n").nth(1).unwrap().trim_end_matches('\n');
4061                let body = body.strip_prefix('…').unwrap_or(body).trim_end_matches('…');
4062                let at = long.find(body).unwrap();
4063                assert!(at == first || (at <= first && at + body.len() + 1 >= long.len()), "{budget}: {g:?} at {at}");
4064            }
4065            // A lesson whose evidence alone exceeds half the budget goes alone.
4066            assert!(batches(&pend[..3], 100).iter().all(|g| g.len() == 1));
4067        }
4068    }
4069
4070    /// The judge's cost line names the ids the answer returned and those that matched no pending lesson,
4071    /// after the fields `borg status` sums (prefilter.rs `tally`).
4072    #[test]
4073    fn the_judge_cost_line_names_the_returned_and_unmatched_ids() {
4074        fn p(lesson: &conscience::Challenge) -> predict::Pending<'_> {
4075            predict::Pending { lesson, before: String::new(), after: String::new(), declared: None, check: String::new(), channel: "stop", stripped: false }
4076        }
4077        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4078        let (a, b) = (c("La"), c("Lb"));
4079        let pend = [p(&a), p(&b)];
4080        let returned = ["La".to_string(), "Lz".to_string()];
4081        let j = |id: &str, verdict, reason: &str| predict::Judged { id: id.into(), verdict, reason: reason.into() };
4082        let one = [j("La", predict::Verdict::NotArisen, "only docs"), j("Lz", predict::Verdict::Changed, "not pending")];
4083        assert_eq!(
4084            judge_line(3, &pend, &one, 900, &returned),
4085            "judge call=3 pending=2 resolved=1 chars=900 returned=La,Lz unmatched=Lz \
4086             verdicts=changed:0,ignored:0,not_arisen:1,consequence_remains:0,unclear:0 reason_chars=9"
4087        );
4088        let pair = [j("La", predict::Verdict::Unclear, ""), j("Lb", predict::Verdict::ConsequenceRemains, "still awaits")];
4089        assert_eq!(
4090            judge_line(4, &pend, &pair, 10, &[]),
4091            "judge call=4 pending=2 resolved=0 chars=10 returned= unmatched= \
4092             verdicts=changed:0,ignored:0,not_arisen:0,consequence_remains:1,unclear:1 reason_chars=12"
4093        );
4094    }
4095
4096    /// A live question answered at the stop: the stop's own message is the answer, so a stripped
4097    /// question's declaration must leave "this turn" too; and when that stop blocks with a new question,
4098    /// its blocked message (that question's work before, and work since for the live one) as well. A
4099    /// question that keeps its declaration keeps it.
4100    #[test]
4101    fn a_stripped_declaration_leaves_every_evidence_of_the_call() {
4102        let session = format!("pending-strip-{}", std::process::id());
4103        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4104        let all = [c("A"), c("F"), c("G")];
4105        let check = |lesson: &str, strip: bool| (0u64..).map(|i| format!("{i:012x}")).find(|k| stripped(k, lesson) == strip).unwrap();
4106        let (ca, cf, cg) = (check("A", false), check("F", true), check("G", false));
4107        let answer = "Fixed the arc. F: applies";
4108        let edits = format!(
4109            "Plan: the blocked message\n#borg-asked A @{ca}\n#borg-answer A: applies\nedit one\n#borg-work the work so far\n\
4110             #borg-asked F @{cf}\n#borg-live F ms=1\n#borg-answer {answer}\n{answer}\n#borg-asked G @{cg}\n"
4111        );
4112        let _ = std::fs::write(edits_file(&session), &edits);
4113        let state: String = edits.lines().filter(|l| DELIVERED.iter().any(|m| l.starts_with(m))).map(|l| format!("{l}\n")).collect();
4114        let j = pending(&session, &state, &all, answer, When::Stop, "00000000cafe");
4115        let p = |id: &str| j.pending.iter().find(|p| p.lesson.id == id).unwrap();
4116        assert!(p("F").stripped && p("F").declared.is_none(), "{:?}", p("F"));
4117        for x in &j.pending {
4118            assert!(!format!("{} {}", x.before, x.after).to_lowercase().contains("f: applies"), "{x:?}");
4119        }
4120        assert!(p("F").after.contains("this turn: Fixed the arc."), "the rest of the message stays: {}", p("F").after);
4121        assert!(!p("A").stripped && p("A").declared == Some(true) && p("A").after.contains("A: applies"), "{:?}", p("A"));
4122        // The work a call shows loses it too.
4123        assert_eq!(without_declarations(&format!("work\n{answer}"), &j.pending), "work\nFixed the arc. ");
4124        assert_eq!(without_declarations("A: applies", &j.pending), "A: applies");
4125        let _ = std::fs::remove_file(edits_file(&session));
4126    }
4127
4128    /// Track R's merged ask: a live check drew A and B from one pool, a stop check drew A again from
4129    /// another, and the stop's block holds the waiting B beside its A. The block's `asked` records are
4130    /// one per routing check (`stop_ask`), each with its own draw's pool; ids of two draws together
4131    /// carry none (counted as no_pool by `borg debug ope`), never one draw's pool with both p.
4132    #[test]
4133    fn a_merged_ask_of_two_draws_carries_no_pool() {
4134        let session = format!("pool-merge-{}", std::process::id());
4135        let _ = std::fs::remove_file(state_file(&session));
4136        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4137        let (a, b, cc, d, e, f) = (c("A"), c("B"), c("C"), c("D"), c("E"), c("F"));
4138        let live: Vec<(f64, &conscience::Challenge)> = vec![(1.0, &a), (0.9, &b), (0.8, &cc), (0.7, &d)];
4139        append_p_of(&session, &live, &[&a, &b], "c1");
4140        let stop: Vec<(f64, &conscience::Challenge)> = vec![(1.0, &a), (0.9, &e), (0.8, &f)];
4141        append_p_of(&session, &stop, &[&a], "c2");
4142        let state = std::fs::read_to_string(state_file(&session)).unwrap_or_default();
4143        let merged = vec!["A".to_string(), "B".to_string()];
4144        assert_eq!(propensity_in(&state, &merged).len(), 2, "each id has its latest p");
4145        assert_eq!(pool_in(&state, &merged), Default::default());
4146        assert_eq!(pool_in(&state, &["A".to_string()]).0, ["A", "E", "F"], "A alone: the stop check's pool");
4147        assert_eq!(pool_in(&state, &["B".to_string()]).0, ["A", "B", "C", "D"], "B alone: the live check's");
4148        assert_eq!(tagged_p(&state, "#borg-p ", "c1", &["B"]).len(), 1, "B's record carries its own check's p");
4149        let _ = std::fs::remove_file(state_file(&session));
4150    }
4151
4152    /// An early delivery's relative scores (`early_rel`): each lesson's match as a share of the best
4153    /// on-topic match, which a lesson done already may hold; a standing lesson's on the same scale (0
4154    /// without a word in common); nothing to scale by without a plain lesson on the topic.
4155    #[test]
4156    fn an_early_lessons_relative_score_is_its_share_of_the_best_on_topic_match() {
4157        let c = |id: &str, q: &str| conscience::Challenge { id: id.into(), question: q.into(), scope: vec!["Orion".into()], ..Default::default() };
4158        let standing = conscience::Challenge { source: conscience::CONSTRAINT.into(), ..c("S", "harpsichord tuning") };
4159        let all = vec![c("A", "orion budget place reset arc drain"), c("B", "orion budget place"), c("C", "telescope glaciers"), standing];
4160        let work = "drain the orion budget place, never with a reset arc";
4161        let topics = vec!["Orion".to_string()];
4162        let ranked = conscience::rank(&all, work);
4163        let score = |id: &str| ranked.iter().find(|(_, x)| x.id == id).map_or(0.0, |(s, _)| *s);
4164        assert!(score("A") > score("B") && score("B") > 0.0, "{ranked:?}");
4165        // A, the best match, was delivered before (done): B's share is of A's score all the same.
4166        let rel = early_rel(&all, work, &topics, &[&all[1], &all[3]]);
4167        assert_eq!(rel, [outcome::round_p(score("B") / score("A")), 0.0]);
4168        assert!(rel[0] < 1.0);
4169        assert_eq!(early_rel(&all, work, &topics, &[&all[0]]), [1.0]);
4170        // Only a standing lesson on a topic no plain lesson matches: nothing to scale by.
4171        assert!(early_rel(&all, work, &["Beacon".to_string()], &[&all[3]]).is_empty());
4172    }
4173
4174    /// A store for `decide`: its lessons file (empty) and outcomes beside it, and a saved selection summary
4175    /// whose fit is sure that handing over beats asking, with a gate open wide: a learned live check
4176    /// deviates on evidence, and its record moves the gate's ledger.
4177    fn decide_store(name: &str) -> std::path::PathBuf {
4178        decide_store_with(name, true)
4179    }
4180
4181    /// `decide_store`, its fit `sure` that handing over beats asking, or (not `sure`) with no evidence
4182    /// either way, asking and handing over worth the same: a learned live check then keeps the fixed rule
4183    /// on evidence and runs the question-count test, which the open gate admits.
4184    fn decide_store_with(name: &str, sure: bool) -> std::path::PathBuf {
4185        use crate::select::{policy, summary};
4186        let d = std::env::temp_dir().join(format!("borg-decide-{name}-{}", std::process::id()));
4187        let _ = std::fs::remove_dir_all(&d);
4188        std::fs::create_dir_all(&d).unwrap();
4189        let lessons = d.join("lessons.jsonl");
4190        std::fs::write(&lessons, "").unwrap();
4191        std::fs::write(outcome::path_beside(&lessons), "").unwrap();
4192        let at = |name: &str| policy::NAMES.iter().position(|n| *n == name).unwrap();
4193        let mut beta = vec![0.0; policy::P];
4194        beta[at("intercept")] = -0.5;
4195        if sure {
4196            beta[at("ask")] = -3.0;
4197            beta[at("not")] = -1.0;
4198        }
4199        let mut chol = vec![0.0; policy::P * policy::P];
4200        for k in 0..policy::P {
4201            chol[k * policy::P + k] = 0.05;
4202        }
4203        let model = policy::Model { beta, chol, sd: vec![1.0; policy::P], spread: [0.1; 3], ..policy::Model::default() };
4204        let rates = policy::Rates { free_stop: 1.0, free_live: 1.0, later: 0.0, wait: 0.0, ask_live: 1.0 };
4205        let fit = policy::Fit { key: "9-abcdef01".into(), model: Some(model), rates, ..policy::Fit::default() };
4206        let gate = summary::Gate { gain: vec![0.0; policy::DRAWS], allowance: 5.0, ..summary::Gate::default() };
4207        summary::save(&lessons, &summary::Summary { fit, gate, ..summary::Summary::default() }).unwrap();
4208        lessons
4209    }
4210
4211    /// `record` for `decide` in a test: the hook's own graded record (`graded_outcome`), beside `lessons`;
4212    /// whether it landed, as `record_graded` answers.
4213    fn record_beside(lessons: &std::path::Path, applied: &[(&conscience::Challenge, u8)], check: &str, r: &crate::select::Routed, basis: &Basis) -> bool {
4214        let order: Vec<&conscience::Challenge> = applied.iter().map(|(c, _)| *c).collect();
4215        let call = Call {
4216            eligible: order.len(),
4217            order: &order,
4218            tail: &[],
4219            pt: 0.0,
4220            boundary: &[],
4221            budget: 0,
4222            model: "",
4223            prompt_hash: String::new(),
4224            dur_ms: 1,
4225            usd: 0.0,
4226            line: "",
4227            moves: Vec::new(),
4228            whys: Vec::new(),
4229        };
4230        let o = graded_outcome("decide-s", "r", check, applied, Some((r, basis)), basis.summary, &call);
4231        outcome::append(&outcome::path_beside(lessons), &o).is_ok()
4232    }
4233
4234    /// The question-count test's k and s reach the graded record from the decision's log, the s = 0
4235    /// draw included; a check outside the test and a failed one log neither.
4236    #[test]
4237    fn the_graded_record_carries_the_question_count_tests_draw() {
4238        use crate::select::{Log, Routed, policy, summary};
4239        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4240        let ls = [c("A"), c("B")];
4241        let applied: Vec<(&conscience::Challenge, u8)> = ls.iter().zip([2, 1]).collect();
4242        let order: Vec<&conscience::Challenge> = ls.iter().collect();
4243        let call = Call {
4244            eligible: 2,
4245            order: &order,
4246            tail: &[],
4247            pt: 0.0,
4248            boundary: &[],
4249            budget: 16_000,
4250            model: "",
4251            prompt_hash: String::new(),
4252            dur_ms: 1,
4253            usd: 0.0,
4254            line: "",
4255            moves: Vec::new(),
4256            whys: Vec::new(),
4257        };
4258        let s = summary::Summary::default();
4259        let basis = Basis { summary: &s, had: &[], why: "" };
4260        let routed = |source: &'static str, xk: Option<u32>, xs: Option<u32>| Routed {
4261            ask: Vec::new(),
4262            free: vec![&ls[0], &ls[1]],
4263            placebo: Vec::new(),
4264            p: vec![0.0, 0.0],
4265            pf: vec![1.0, 1.0],
4266            log: Log { policy: "learned", source, xk, xs, ..Log::default() },
4267        };
4268        let record = |r: Option<&Routed>| graded_outcome("s", "r", "0123456789ab", &applied, r.map(|r| (r, &basis)), &s, &call);
4269        let drawn = record(Some(&routed(policy::EXPERIMENT_CONTROL, Some(2), Some(0))));
4270        assert_eq!((drawn.source.as_str(), drawn.xk, drawn.xs), (policy::EXPERIMENT_CONTROL, Some(2), Some(0)));
4271        let outside = record(Some(&routed(policy::EVIDENCE, None, None)));
4272        assert_eq!((outside.source.as_str(), outside.xk, outside.xs), (policy::EVIDENCE, None, None));
4273        let failed = record(None);
4274        assert_eq!((failed.failed, failed.xk, failed.xs), (true, None, None));
4275        // The pool budget the check filled to, whatever it decided.
4276        assert_eq!((drawn.budget, failed.budget), (16_000, 16_000));
4277    }
4278
4279    fn graded_records(lessons: &std::path::Path) -> Vec<outcome::Outcome> {
4280        outcome::read(&outcome::path_beside(lessons)).into_iter().filter(|o| o.kind == outcome::GRADED).collect()
4281    }
4282
4283    /// A check decides on the summary it loads under the store's decision lock, never on the one it read
4284    /// at its start: B started before A decided, yet B's decision reads A's ledger entry and says so.
4285    #[test]
4286    fn a_check_decides_on_the_summary_under_the_lock_with_every_decision_before_it() {
4287        use crate::select::{Policy, policy, summary};
4288        let lessons = decide_store("fresh");
4289        // What both checks read at their start, before either decided.
4290        let stale = summary::refresh(&lessons);
4291        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4292        let ls = [c("A"), c("B"), c("C"), c("D")];
4293        let applied: Vec<(&conscience::Challenge, u8)> = ls.iter().zip([3, 1, 2, 1]).collect();
4294        let mut gates = Vec::new();
4295        // One after the other, with the hook's own wait: a process spawned by another test while the
4296        // first held the lock keeps a copy of its descriptor, so the flock, until it execs.
4297        for check in ["0000000000a1", "0000000000b2"] {
4298            let at = Decide { session: "decide-fresh", check, channel: "live", lessons: &lessons, tokens: 0, recorded: true };
4299            let r = decide(Policy::Learned, &applied, &at, &stale, DECIDE_WAIT, |r, b| record_beside(&lessons, &applied, check, r, b));
4300            gates.push(r.log.gate);
4301        }
4302        let g = graded_records(&lessons);
4303        assert_eq!(g.len(), 2);
4304        assert!(g.iter().all(|o| o.policy == "learned" && o.reason.is_empty()), "{g:?}");
4305        assert_ne!(g[0].free, Vec::<String>::new(), "the first deviates: everything free at a live check, {:?}", g[0]);
4306        assert_eq!(g[0].seen, stale.key(), "nothing was decided before the first");
4307        // The second's summary is the start's with the first's record folded in, its ledger entry too.
4308        let log = std::fs::read_to_string(outcome::path_beside(&lessons)).unwrap();
4309        let first = log.lines().next().unwrap();
4310        let after = summary::fold(&format!("{first}\n"), stale.clone());
4311        assert_eq!(after.gate.checks, stale.gate.checks + 1, "the first's decision is in the gate's ledger");
4312        assert_eq!(g[1].seen, after.key(), "decided on the summary loaded under the lock, not the one read at its start");
4313        assert_eq!(gates[1], policy::remaining(&after.gate).max(0.0), "against the gate with the first's entry");
4314        let _ = std::fs::remove_dir_all(lessons.parent().unwrap());
4315    }
4316
4317    /// Two checks deciding at once take turns: while one holds the store's decision lock (deciding and
4318    /// recording), the other, past its wait, takes the fixed rule, which reads no gate, and says why;
4319    /// never both against the same gate. So for a deviation on evidence and for the question-count test
4320    /// alike (its draw logged on the learned record only). A lock that cannot be taken at all is the
4321    /// fixed rule too.
4322    #[test]
4323    fn two_checks_deciding_at_once_never_both_take_the_gate() {
4324        use crate::select::{Policy, policy, summary};
4325        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4326        let ls = [c("A"), c("B"), c("C"), c("D")];
4327        let applied: Vec<(&conscience::Challenge, u8)> = ls.iter().zip([3, 1, 2, 1]).collect();
4328        let mut stores = Vec::new();
4329        for (name, sure, sources) in [("race", true, &[policy::EVIDENCE][..]), ("race-test", false, &[policy::EXPERIMENT, policy::EXPERIMENT_CONTROL][..])] {
4330            let lessons = decide_store_with(name, sure);
4331            let stale = summary::refresh(&lessons);
4332            // Each records only once both have decided: the first to take the lock holds it until the
4333            // other decided without it.
4334            let both = std::sync::Barrier::new(2);
4335            std::thread::scope(|s| {
4336                for check in ["0000000000c1", "0000000000c2"] {
4337                    let (applied, lessons, stale, both) = (&applied, &lessons, &stale, &both);
4338                    s.spawn(move || {
4339                        let at = Decide { session: "decide-race", check, channel: "live", lessons, tokens: 0, recorded: true };
4340                        decide(Policy::Learned, applied, &at, stale, std::time::Duration::ZERO, |r, b| {
4341                            both.wait();
4342                            record_beside(lessons, applied, check, r, b)
4343                        });
4344                    });
4345                }
4346            });
4347            let mut got: Vec<(String, String, String, bool)> =
4348                graded_records(&lessons).into_iter().map(|o| (o.policy, o.source, o.reason, o.xk.is_some())).collect();
4349            got.sort();
4350            assert_eq!(got.len(), 2, "{name}: {got:?}");
4351            assert_eq!(got[0], ("fixed".to_string(), "fixed".to_string(), outcome::DECIDED_BUSY.to_string(), false), "{name}: the other took the fixed rule");
4352            let (policy, source, reason, drawn) = &got[1];
4353            assert!(policy == "learned" && sources.contains(&source.as_str()) && reason.is_empty(), "{name}: one decided under the lock: {got:?}");
4354            assert_eq!(*drawn, !sure, "{name}: the test's draw on its record alone");
4355            assert!(policy::remaining(&summary::refresh(&lessons).gate) >= -1e-9, "{name}: the gate never overspent");
4356            stores.push(lessons);
4357        }
4358        // No lock to be had (the store's directory is gone): the fixed rule, and why.
4359        let stale = summary::refresh(&stores[0]);
4360        let gone = std::path::Path::new("/nonexistent/borg-decide/lessons.jsonl");
4361        let at = Decide { session: "decide-race", check: "0000000000c3", channel: "live", lessons: gone, tokens: 0, recorded: true };
4362        let mut why = "";
4363        let r = decide(Policy::Learned, &applied, &at, &stale, DECIDE_WAIT, |_, b| {
4364            why = b.why;
4365            true
4366        });
4367        assert_eq!((r.log.policy, why), ("fixed", outcome::DECIDED_NO_LOCK));
4368        for lessons in stores {
4369            let _ = std::fs::remove_dir_all(lessons.parent().unwrap());
4370        }
4371    }
4372
4373    /// The fixed rule and the evaluation switches read no gate: while another check holds the store's
4374    /// decision lock, each decides by itself at once, on the summary its check read at its start, and its
4375    /// record gives no reason; only a learned check waits for the lock or falls back.
4376    #[test]
4377    fn a_policy_that_reads_no_gate_decides_by_itself_whoever_holds_the_lock() {
4378        use crate::select::{Ctx, Policy, graded_under, summary};
4379        let lessons = decide_store("nolock");
4380        let stale = summary::refresh(&lessons);
4381        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4382        let ls = [c("A"), c("B"), c("C"), c("D")];
4383        let applied: Vec<(&conscience::Challenge, u8)> = ls.iter().zip([3, 1, 3, 2]).collect();
4384        let ids = |cs: &[&conscience::Challenge]| cs.iter().map(|c| c.id.clone()).collect::<Vec<_>>();
4385        let held = lock_within(&decide_lock(&lessons), std::time::Duration::ZERO).unwrap();
4386        let check = "0000000000e1";
4387        let at = Decide { session: "decide-nolock", check, channel: "stop", lessons: &lessons, tokens: 0, recorded: true };
4388        let ctx = Ctx {
4389            session: at.session,
4390            check,
4391            channel: at.channel,
4392            lessons: &lessons,
4393            summary: &stale,
4394            tokens: 0,
4395            had: &[],
4396            alpha: crate::select::alpha(&lessons),
4397        };
4398        let fixed = ids(&graded_under(Policy::Fixed, &applied, &ctx).ask);
4399        for policy in [Policy::Fixed, Policy::Cap(1), Policy::Attention] {
4400            let own = graded_under(policy, &applied, &ctx);
4401            let mut got = None;
4402            let r = decide(policy, &applied, &at, &stale, std::time::Duration::ZERO, |r, b| {
4403                got = Some((r.log.policy, b.why, b.summary.key()));
4404                true
4405            });
4406            assert_eq!(got, Some((policy.name(), "", stale.key())), "{policy:?}");
4407            assert_eq!((ids(&r.ask), ids(&r.free)), (ids(&own.ask), ids(&own.free)), "{policy:?}: its own rule");
4408        }
4409        // Not vacuous: the cap's block differs from the fixed rule's, which a fallback would have asked.
4410        assert_ne!(ids(&graded_under(Policy::Cap(1), &applied, &ctx).ask), fixed);
4411        let mut why = "";
4412        let r = decide(Policy::Learned, &applied, &at, &stale, std::time::Duration::ZERO, |_, b| {
4413            why = b.why;
4414            true
4415        });
4416        assert_eq!((r.log.policy, why), ("fixed", outcome::DECIDED_BUSY), "the learned policy alone waits for the lock");
4417        drop(held);
4418        let _ = std::fs::remove_dir_all(lessons.parent().unwrap());
4419    }
4420
4421    /// A learned decision is carried out only once its record landed in the log its gate is folded from,
4422    /// the gate's only debit. One whose record would not land there (learning off, the lessons of
4423    /// another store: `Decide::recorded`) or did not (a failed append) takes the fixed rule and says why,
4424    /// so checks one after another never each risk the whole allowance of a gate none of them debited.
4425    #[test]
4426    fn a_learned_decision_whose_record_cannot_land_takes_the_fixed_rule() {
4427        use crate::select::{Ctx, Policy, graded_under, policy, summary};
4428        let lessons = decide_store("unrecorded");
4429        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4430        let ls = [c("A"), c("B"), c("C"), c("D")];
4431        let applied: Vec<(&conscience::Challenge, u8)> = ls.iter().zip([3, 1, 2, 1]).collect();
4432        let ids = |cs: &[&conscience::Challenge]| cs.iter().map(|c| c.id.clone()).collect::<Vec<_>>();
4433        let at = |check: &'static str, recorded: bool| Decide { session: "decide-unrecorded", check, channel: "live", lessons: &lessons, tokens: 0, recorded };
4434        let fixed = |check: &str, s: &summary::Summary| {
4435            let ctx = Ctx {
4436                session: "decide-unrecorded",
4437                check,
4438                channel: "live",
4439                lessons: &lessons,
4440                summary: s,
4441                tokens: 0,
4442                had: &[],
4443                alpha: crate::select::alpha(&lessons),
4444            };
4445            let r = graded_under(Policy::Fixed, &applied, &ctx);
4446            (ids(&r.ask), ids(&r.free), ids(&r.placebo))
4447        };
4448        // Its record lands: the learned policy deviates (not vacuous).
4449        let stale = summary::refresh(&lessons);
4450        let r = decide(Policy::Learned, &applied, &at("0000000000f1", true), &stale, std::time::Duration::ZERO, |r, b| {
4451            record_beside(&lessons, &applied, "0000000000f1", r, b)
4452        });
4453        assert_eq!(r.log.source, policy::EVIDENCE, "{:?}", r.log);
4454        let (ask, free, _) = fixed("0000000000f1", &stale);
4455        assert_ne!((ids(&r.ask), ids(&r.free)), (ask, free), "it deviated");
4456        let gate = summary::refresh(&lessons).gate;
4457        // Two checks whose records would not land, one after the other: both the fixed rule, never tried.
4458        for check in ["0000000000f2", "0000000000f3"] {
4459            let stale = summary::refresh(&lessons);
4460            let mut tried = Vec::new();
4461            let r = decide(Policy::Learned, &applied, &at(check, false), &stale, std::time::Duration::ZERO, |r, b| {
4462                tried.push((r.log.policy, b.why));
4463                false
4464            });
4465            assert_eq!(tried, [("fixed", outcome::DECIDED_NO_RECORD)], "{check}");
4466            assert_eq!((ids(&r.ask), ids(&r.free), ids(&r.placebo)), fixed(check, &stale), "{check}");
4467        }
4468        // A check whose record failed to append: decided again by the fixed rule under the lock, and that
4469        // carried out; the store's log and its gate are as they were.
4470        let stale = summary::refresh(&lessons);
4471        let mut tried = Vec::new();
4472        // After the first check's lock, so with the hook's wait (as in the test above).
4473        let r = decide(Policy::Learned, &applied, &at("0000000000f4", true), &stale, DECIDE_WAIT, |r, b| {
4474            tried.push((r.log.policy, b.why));
4475            false
4476        });
4477        assert_eq!(tried, [("learned", ""), ("fixed", outcome::DECIDED_NO_RECORD)]);
4478        assert_eq!((ids(&r.ask), ids(&r.free), ids(&r.placebo)), fixed("0000000000f4", &stale));
4479        assert_eq!(graded_records(&lessons).len(), 1, "only the first check's record");
4480        assert_eq!(summary::refresh(&lessons).gate, gate, "no unrecorded decision moved the gate");
4481        let _ = std::fs::remove_dir_all(lessons.parent().unwrap());
4482    }
4483
4484    /// A lesson another check queued to be asked or handed over, not delivered yet, is had at a decision
4485    /// as a delivered one is (`waiting_in`): it never takes a free slot (the next lesson does) nor a
4486    /// placebo's, and the record names it (`nofree`), so delivery never drops a second copy as queued
4487    /// twice. A lesson waiting as a placebo was routed nowhere: it may still go free.
4488    #[test]
4489    fn a_lesson_waiting_in_the_queue_never_takes_a_free_or_placebo_slot() {
4490        use crate::select::{Policy, summary};
4491        let lessons = decide_store("queued");
4492        let stale = summary::refresh(&lessons);
4493        let session = format!("decide-queued-{}", std::process::id());
4494        let _ = std::fs::remove_file(state_file(&session));
4495        std::fs::write(queued_file(&session), "X @0000000000d0\nlive Y @0000000000d0\nplacebo Z @0000000000d0\n").unwrap();
4496        let c = |id: &str| conscience::Challenge { id: id.into(), ..Default::default() };
4497        let named: Vec<String> = (1..=10).map(|i| format!("L{i:02}")).collect();
4498        let ls: Vec<conscience::Challenge> = ["X", "Y", "Z"].iter().map(|i| c(i)).chain(named.iter().map(|i| c(i))).collect();
4499        let ids = |cs: &[&conscience::Challenge]| cs.iter().map(|c| c.id.clone()).collect::<Vec<_>>();
4500        // No 3, so the fixed rule hands over up to `CAP`: X and Y first in line (the slots they would take
4501        // go to the next), or last, past ten 2s, where a placebo is drawn among what is left.
4502        let first: Vec<(&conscience::Challenge, u8)> = ls.iter().map(|c| (c, if ["X", "Y"].contains(&c.id.as_str()) { 2 } else { 1 })).collect();
4503        let last: Vec<(&conscience::Challenge, u8)> = ls.iter().map(|c| (c, if c.id.starts_with('L') { 2 } else { 1 })).collect();
4504        let waits = |l: &[String]| l.iter().any(|i| i == "X" || i == "Y");
4505        for (check, applied) in [("0000000000d1", &first), ("0000000000d2", &last)] {
4506            let at = Decide { session: &session, check, channel: "stop", lessons: &lessons, tokens: 0, recorded: true };
4507            let r = decide(Policy::Fixed, applied, &at, &stale, std::time::Duration::ZERO, |r, b| record_beside(&lessons, applied, check, r, b));
4508            let (free, placebo) = (ids(&r.free), ids(&r.placebo));
4509            assert_eq!(free.len(), conscience::CAP, "{check}: every free slot taken: {free:?}");
4510            assert!(!waits(&free) && !waits(&placebo), "{check}: free {free:?}, placebo {placebo:?}");
4511        }
4512        assert_eq!(graded_records(&lessons)[1].placebo, ["Z"], "waiting as a placebo, Z is still drawn");
4513        assert!(graded_records(&lessons)[0].free.contains(&"Z".to_string()), "and still goes free");
4514        let g = graded_records(&lessons);
4515        assert_eq!(g.iter().map(|o| o.nofree.clone()).collect::<Vec<_>>(), [["X", "Y"], ["X", "Y"]], "{g:?}");
4516        let _ = std::fs::remove_file(queued_file(&session));
4517        let _ = std::fs::remove_dir_all(lessons.parent().unwrap());
4518    }
4519
4520    #[test]
4521    fn only_an_adapter_reads_which_messages_an_answer_showed() {
4522        let out = json!({"hookSpecificOutput": {"additionalContext": "[borg] Message from s1: hi"}, "shown_ids": ["m0123456789ab"]});
4523        for name in ["claude", "codex"] {
4524            let h = harness::by_name(name).unwrap();
4525            assert_eq!(for_harness(h, out.clone()), json!({"hookSpecificOutput": out["hookSpecificOutput"]}), "{name}");
4526        }
4527        for name in ["omp", "opencode"] {
4528            assert_eq!(for_harness(harness::by_name(name).unwrap(), out.clone()), out, "{name}");
4529        }
4530    }
4531}