1use 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
26pub(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
32pub(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
40pub(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 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
65const COMMAND_CHARS: usize = borg_trace::dialogue::TOUCH_CHARS;
69
70pub(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
81fn 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 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
93fn prompts_file(session: &str) -> PathBuf {
95 state_file(session).with_extension("prompts")
96}
97
98fn where_file(session: &str) -> PathBuf {
102 state_file(session).with_extension("where")
103}
104
105fn 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
126fn denied_file(session: &str) -> PathBuf {
130 state_file(session).with_extension("denied")
131}
132
133fn 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
156fn 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
177fn 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
199fn 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
206fn 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
231fn lessons(session: &str) -> Vec<conscience::Challenge> {
236 loaded(session).0
237}
238
239fn 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
246fn 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 let mut all = conscience::load_store_for(&lessons_path(), before, Some(session));
251 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 all.retain(|c| crate::reach::deliverable(c, &repos));
260 let withheld = crate::absorb::withhold_beside(&lessons_path(), &mut all, before);
262 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
273fn lessons_path() -> PathBuf {
275 std::env::var_os("BORG_CONSCIENCE").map(PathBuf::from).unwrap_or_else(|| learn::home().join("lessons.jsonl"))
276}
277
278fn edits_file(session: &str) -> PathBuf {
283 state_file(session).with_extension("edits")
284}
285
286fn cost_file(session: &str) -> PathBuf {
290 state_file(session).with_extension("cost")
291}
292
293fn 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
305fn 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
328fn 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
338fn 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
348fn 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
359fn 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
368const WORK: &str = "#borg-work ";
373
374pub(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
397const EDIT_CHARS: usize = 600;
400
401const EVIDENCE_CHARS: usize = 4000;
407const _: () = assert!(EVIDENCE_CHARS >= 3 * predict::JUDGE_PART);
408
409pub(crate) fn one_line(text: &str, n: usize) -> String {
411 text.split_whitespace().collect::<Vec<_>>().join(" ").chars().take(n).collect()
412}
413
414fn 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
429const MAX_PENDING: usize = conscience::CAP;
435const MAX_JUDGE_TRIES: usize = 2;
441
442pub(crate) const END_START_BUDGET: usize = 72_000;
450
451const END_LESSONS_SHARE: usize = 2;
455
456#[derive(Debug, Clone, Copy, PartialEq)]
458pub(crate) struct EndBudget {
459 pub chars: usize,
461 pub draw: Option<(f64, bool)>,
464}
465
466pub(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
487pub(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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
501enum Cut {
502 Newest,
503 After,
504 Ends,
505}
506
507impl Cut {
508 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 fn word(self) -> &'static str {
519 match self {
520 Cut::Newest => "newest",
521 Cut::After => "after",
522 Cut::Ends => "ends",
523 }
524 }
525
526 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
546fn end_timeout() -> std::time::Duration {
552 learn::learn_timeout().min(predict::timeout().saturating_mul(2))
553}
554
555const END_JUDGED: &str = "#borg-end-judged ";
560
561#[derive(Debug, Clone, Copy, PartialEq, Eq)]
564enum When {
565 Stop,
566 End,
567}
568
569const DELIVERED: [&str; 4] = ["#borg-asked ", "#borg-early ", "#borg-queued ", "#borg-placebo "];
574
575const ROUTED: &str = "#borg-routed ";
579
580fn 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#[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
598fn 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
623fn 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
632struct Judging<'a> {
635 pending: Vec<predict::Pending<'a>>,
636 question: bool,
637 crowded: Vec<String>,
638}
639
640fn 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
661fn 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 struct Delivery<'l> {
677 channel: &'static str,
678 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 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 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 let mut out: Vec<(predict::Pending, bool)> = Vec::new();
727 for d in deliveries {
728 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 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
762fn 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
768fn 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
777pub(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 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
822fn record_verdicts(session: &str, repo: &str, at: &str, pending: &[predict::Pending], pred: &predict::Prediction, logged: &Logged) {
837 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 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 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 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
889struct Logged {
894 call: usize,
895 chars: usize,
896 model: String,
897 fields: String,
898}
899
900fn 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
922pub 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 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 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 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 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 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 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 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
1050fn 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
1071struct Batch {
1074 chars: usize,
1075 cut: bool,
1076 pred: Option<predict::Prediction>,
1077 ms: u64,
1078}
1079
1080fn 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
1094struct Source {
1098 label: &'static str,
1099 text: Vec<char>,
1100 marks: Vec<(usize, Vec<String>)>,
1101}
1102
1103impl Source {
1104 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 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 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 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
1145fn 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
1161fn 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 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
1196fn 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
1203pub(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
1208fn ids_in(list: &str) -> impl Iterator<Item = &str> {
1210 tagged(list).0.split(',').map(str::trim).filter(|i| !i.is_empty())
1211}
1212
1213fn tag(check: &str) -> String {
1215 if check.is_empty() { String::new() } else { format!(" @{check}") }
1216}
1217
1218const 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 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 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 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 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 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 let now_settled: Vec<&str> = topics
1290 .iter()
1291 .filter(|t| {
1292 !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 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
1325fn 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
1336fn 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
1345fn 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
1375fn 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
1396fn 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
1403fn 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
1415fn queued_file(session: &str) -> PathBuf {
1422 state_file(session).with_extension("queued")
1423}
1424
1425#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
1427enum Queued {
1428 Ask,
1431 Instead,
1433 Free,
1434 Placebo,
1435}
1436
1437const INSTEAD_NEXT: &str = "instead-next ";
1443
1444const ASK_NEXT: [&str; 2] = ["live ", "live-next "];
1452
1453const LEGACY_INSTEAD: &str = "instead ";
1456
1457fn 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
1470fn 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
1489fn 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
1500fn 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
1510fn supersede(session: &str, input: &Value, shown: &[&conscience::Challenge], asks: &[&conscience::Challenge]) -> Vec<String> {
1516 let q = queued_file(session);
1517 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
1560const PROPOSAL: &str = "#borg-proposal ";
1565
1566const 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
1576fn 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
1585fn 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
1609const WAKE_EVENT: &str = "wake";
1612
1613#[must_use]
1616pub(crate) fn lesson_wake_id(check: &str, lesson: &str) -> String {
1617 format!("L@{check}:{lesson}")
1618}
1619
1620#[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
1626fn reserved_file(session: &str) -> PathBuf {
1630 state_file(session).with_extension("queued.wake")
1631}
1632
1633#[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#[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 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 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
1732fn 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
1739fn 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
1776pub(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
1798pub(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 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
1812fn 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
1822fn note_never_delivered(session: &str, input: &Value) {
1826 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
1842fn 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
1860type Pairs<'a> = Vec<(&'a conscience::Challenge, &'a str)>;
1862
1863type Drop<'a> = (&'a str, &'static str, &'static str, &'static str);
1866
1867fn 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 lines.sort_by_key(|l| l.0);
1892 let placebos = lines.iter().position(|l| l.0 == Queued::Placebo).unwrap_or(lines.len());
1893 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 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 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 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 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 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 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 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
2012fn 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
2019struct SoFar<'a> {
2023 h: &'a dyn Harness,
2024 input: &'a Value,
2025 text: std::cell::OnceCell<String>,
2026 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
2041fn turn_file(session: &str) -> PathBuf {
2044 state_file(session).with_extension("turn")
2045}
2046
2047fn 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
2074fn 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
2079fn 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
2095enum Verdict {
2097 Skip(String),
2099 Failed,
2101 Ask(Vec<String>),
2103}
2104
2105struct Work<'a> {
2107 all: &'a [conscience::Challenge],
2109 judged: &'a [conscience::Challenge],
2112 challenges: &'a [conscience::Challenge],
2115 done: &'a std::collections::HashSet<&'a str>,
2116 state: &'a str,
2118 context: &'a str,
2121 scope_text: &'a str,
2122 message: &'a str,
2124 changed: bool,
2126 judge: bool,
2129 budget: Option<std::time::Duration>,
2132 started: std::time::Instant,
2133 detached: Option<std::time::Duration>,
2137 edits: Option<usize>,
2140 check: &'a str,
2143 summary: &'a crate::select::summary::Summary,
2147}
2148
2149impl Work<'_> {
2150 fn limit(&self) -> std::time::Duration {
2153 self.detached.unwrap_or_else(|| live::call_limit(self.budget, self.started.elapsed()))
2154 }
2155
2156 fn channel(&self) -> &'static str {
2159 if self.detached.is_some() { "live" } else { "stop" }
2160 }
2161}
2162
2163struct Call<'a> {
2165 eligible: usize,
2168 order: &'a [&'a conscience::Challenge],
2169 tail: &'a [(usize, &'a conscience::Challenge)],
2173 pt: f64,
2174 boundary: &'a [(usize, &'a conscience::Challenge)],
2177 budget: usize,
2179 model: &'a str,
2180 prompt_hash: String,
2181 dur_ms: u64,
2182 usd: f64,
2184 line: &'static str,
2188 moves: Vec<String>,
2189 whys: Vec<String>,
2190}
2191
2192fn 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 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
2214fn 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
2221struct Pool<'a> {
2223 shortlist: Vec<&'a conscience::Challenge>,
2225 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 whole: bool,
2235 judging: Judging<'a>,
2237 rows: Option<crate::pools::Rows<'a>>,
2240 text: String,
2241}
2242
2243fn pool<'a>(session: &str, w: &Work<'a>, text: &str) -> Result<Pool<'a>, Verdict> {
2246 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 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 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 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 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 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 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
2315struct Called {
2317 pred: Option<predict::Prediction>,
2318 dur_ms: u64,
2319 model: String,
2320 extra: usize,
2322 graded: bool,
2323 compact: bool,
2324}
2325
2326fn 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 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 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
2352fn 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
2370pub(crate) struct Late<'l> {
2374 pub offset: usize,
2375 pub gone: bool,
2376 pub proposal: &'l str,
2377 pub stop_ms: i64,
2379}
2380
2381struct Settled<'a> {
2385 verdict: Verdict,
2386 free: Vec<String>,
2387 superseded: Vec<String>,
2388 pool: Option<PoolLog<'a>>,
2389}
2390
2391struct 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 idx: u64,
2404 stop_ms: i64,
2405}
2406
2407fn 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
2434fn 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 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 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 append(session, &format!("#borg-predict-cost {:.4}", pred.cost));
2496 let gone = late.is_some_and(|l| l.gone);
2499 if !pend.is_empty() && !gone {
2500 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 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 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 let lessons = lessons_path();
2526 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 let mut queue: Vec<String> = Vec::new();
2551 if let Some(late) = late
2552 && !r.ask.is_empty()
2553 {
2554 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 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 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 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 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 none(Verdict::Ask(ids(&widened)))
2607 }
2608}
2609
2610fn 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
2625struct Decide<'x> {
2627 session: &'x str,
2628 check: &'x str,
2629 channel: &'static str,
2630 lessons: &'x std::path::Path,
2631 tokens: u32,
2633 recorded: bool,
2636}
2637
2638struct Basis<'s> {
2643 summary: &'s crate::select::summary::Summary,
2644 had: &'s [String],
2645 why: &'static str,
2646}
2647
2648const DECIDE_WAIT: std::time::Duration = std::time::Duration::from_millis(live::AFTER_CALL.as_millis() as u64 / 3);
2655
2656const DECIDE_POLL: std::time::Duration = std::time::Duration::from_millis(5);
2658
2659fn decide_lock(lessons: &std::path::Path) -> PathBuf {
2661 crate::select::summary::cache_path(lessons).with_extension("json.lock")
2662}
2663
2664fn 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
2680fn 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 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 let alpha = crate::select::alpha(at.lessons);
2715 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 (_, false) => on(&summary, Policy::Fixed, outcome::DECIDED_NO_RECORD).0,
2736 }
2737 }
2738 Err(why) => on(stale, Policy::Fixed, why).0,
2739 }
2740}
2741
2742fn records_reach_gate(lessons: &std::path::Path) -> bool {
2747 learning() && outcome::path_beside(lessons) == graded_log()
2748}
2749
2750fn graded_log() -> PathBuf {
2752 learn::home().join("outcomes.jsonl")
2753}
2754
2755fn 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
2771fn 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 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#[allow(clippy::too_many_arguments)] fn 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 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 xk: routed.and_then(|r| r.log.xk),
2846 xs: routed.and_then(|r| r.log.xs),
2847 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 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 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
2879pub 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 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 if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(prompts_file(session)) {
2915 let _ = writeln!(f, "{prompt}");
2916 }
2917 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 let checked = crate::update::maybe_check(learn::now_ms());
2928 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 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 if harness::transcript(h.caps(), input).is_none() {
2941 let _ = std::fs::write(turn_file(session), "");
2942 }
2943 out.flatten()
2944}
2945
2946fn 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 crate::notes::remember_proposal(&crate::notes::state_dir(), session, &full_proposal(h, input));
2976 crate::notes::mark(&crate::notes::state_dir(), session, "stop");
2977 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 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 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 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 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 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 (h.caps().tool_context && !subagent(input)).then(|| free_context(h, "PostToolUse", session, input, &touch)).flatten()
3038}
3039
3040fn subagent(input: &Value) -> bool {
3043 input.get("agent_id").and_then(Value::as_str).is_some_and(|a| !a.is_empty())
3044}
3045
3046struct 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 crate::notes::remember_proposal(&crate::notes::state_dir(), session, &full_proposal(h, input));
3067 crate::notes::mark(&crate::notes::state_dir(), session, "stop");
3068 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 note_answer(h, session, input);
3080 let changed = std::fs::remove_file(changes_file(session)).is_ok();
3083 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 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 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 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#[allow(clippy::too_many_arguments)] fn 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 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 let queue = if only { String::new() } else { std::fs::read_to_string(queued_file(session)).unwrap_or_default() };
3157 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 let ending: std::collections::HashSet<&str> = if judge { Default::default() } else { open_in(state) };
3165 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
3174fn 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
3181const CHECKING: &str = "#borg-checking";
3184
3185fn 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 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 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 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
3238fn 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
3244fn 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
3251pub(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
3263const LOCK_TRIES: usize = 3;
3266
3267pub(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 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 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 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
3358fn 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 Ok(ids) => Ok(w.challenges.iter().filter(|c| ids.split(',').any(|i| i == c.id)).collect()),
3364 Err(_) if std::env::var("BORG_MATCH").as_deref() != Ok("bm25") => stop_check(h, session, input, w),
3367 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
3381fn 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
3390fn 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
3397fn 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
3407fn 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
3425fn 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
3433fn 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 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
3474fn 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
3489fn 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
3507fn 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 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
3544fn stop_ask(h: &dyn Harness, session: &str, input: &Value, mut lessons: Vec<&conscience::Challenge>, check: &str) -> Option<Value> {
3547 crate::memory::shuffle(&mut lessons, session);
3550 let log = lessons.iter().map(|c| c.id.as_str()).collect::<Vec<_>>().join(",");
3551 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 let _ = writeln!(f, "{}\n{marker}", one_line(&full_proposal(h, input), EVIDENCE_CHARS));
3559 }
3560 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
3579pub 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 if !h.caps().sends_pid
3590 && let Some(m) = input.as_object_mut()
3591 {
3592 m.shift_remove("pid");
3593 }
3594 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 unsafe {
3606 std::env::set_var("BORG_HOME", &home);
3607 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 let now = learn::now_ms();
3615 let session = input.get("session_id").and_then(Value::as_str).unwrap_or("unknown");
3616 crate::listen::touch(&home, event, session, now);
3620 if matches!(event, "start" | "halted") {
3622 return;
3623 }
3624 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 let started = me.map(|p| p.started);
3638 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 let out = crate::listen::after_stop(&home, h, stopping.as_ref(), &input, out);
3643 let out = harness::with_budgets(h, &input, out);
3645 if let Some(out) = crate::msg::deliver(&home, h.caps(), event, &input, started, now, out) {
3647 println!("{}", for_harness(h, out));
3648 }
3649}
3650
3651fn 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 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 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 assert_eq!(queue_line("A,B"), (Queued::Free, vec!["A", "B"], ""));
3736 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 #[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 #[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 #[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 #[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 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 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 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 #[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 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 #[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 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 #[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 #[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 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 #[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 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 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 assert!(batches(&pend[..3], 100).iter().all(|g| g.len() == 1));
4067 }
4068 }
4069
4070 #[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 #[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 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 #[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 #[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 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 assert!(early_rel(&all, work, &["Beacon".to_string()], &[&all[3]]).is_empty());
4172 }
4173
4174 fn decide_store(name: &str) -> std::path::PathBuf {
4178 decide_store_with(name, true)
4179 }
4180
4181 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 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 #[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 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 #[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 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 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 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 #[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 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 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 #[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 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 #[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 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 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 let stale = summary::refresh(&lessons);
4471 let mut tried = Vec::new();
4472 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 #[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 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}