1use std::collections::{BTreeMap, HashSet};
27use std::path::Path;
28
29use serde::Serialize;
30use serde_json::{Value, json};
31
32use crate::clock::{DAY_MS, ago};
33use crate::collective::{self, Route, Share};
34use crate::msg::{self, Address, Message};
35use crate::presence::{self, Presence, State};
36use crate::{install, learn, protocol, select, steer, update};
37
38#[derive(Debug, Clone, Serialize)]
39pub struct Drone {
40 pub designation: String,
42 pub number: usize,
44 pub node: String,
46 pub state: State,
47 pub paused: bool,
49 #[serde(flatten)]
50 pub p: Presence,
51}
52
53#[derive(Debug, Clone, Default, Serialize)]
55pub struct Inbox {
56 pub unread: BTreeMap<String, usize>,
58 pub recent: Vec<Message>,
60 pub day: usize,
62 #[serde(skip)]
64 pub here: String,
65 pub listen: BTreeMap<String, crate::listen::Status>,
67}
68
69pub const RECENT: usize = 5;
71
72const NUMBERS: [&str; 20] = [
73 "One",
74 "Two",
75 "Three",
76 "Four",
77 "Five",
78 "Six",
79 "Seven",
80 "Eight",
81 "Nine",
82 "Ten",
83 "Eleven",
84 "Twelve",
85 "Thirteen",
86 "Fourteen",
87 "Fifteen",
88 "Sixteen",
89 "Seventeen",
90 "Eighteen",
91 "Nineteen",
92 "Twenty",
93];
94
95pub fn word(n: usize) -> String {
96 NUMBERS.get(n.wrapping_sub(1)).map(|s| s.to_string()).unwrap_or_else(|| n.to_string())
97}
98
99pub fn assemble(local: Vec<Presence>, remote: Vec<presence::Remote>, now: i64, alive: impl Fn(u32) -> bool, paused: impl Fn(&Presence) -> bool) -> Vec<Drone> {
101 let mut all: Vec<(String, Presence, State, bool)> = local
102 .into_iter()
103 .map(|p| {
104 let s = presence::state(&p, now, &alive);
105 let z = s != State::Ended && paused(&p);
106 (String::new(), p, s, z)
107 })
108 .collect();
109 all.extend(remote.into_iter().map(|r| {
111 let s = presence::state(&r.p, now, |_| true);
112 (r.key, r.p, s, false)
113 }));
114 all.sort_by(|a, b| (a.1.started, &a.0, &a.1.session).cmp(&(b.1.started, &b.0, &b.1.session)));
115 let n = all.len();
116 all.into_iter()
117 .enumerate()
118 .map(|(i, (node, p, state, paused))| Drone { designation: format!("{} of {}", word(i + 1), word(n)), number: i + 1, node, state, paused, p })
119 .collect()
120}
121
122pub fn drones(home: &Path, may_sync: bool, now: i64) -> Vec<Drone> {
124 let remote = if may_sync { presence::remote(home, now) } else { Vec::new() };
125 assemble(presence::local(home, now), remote, now, presence::alive, |p| steer::paused(home, &p.session, Some(p.started), now))
126}
127
128pub fn inbox(home: &Path, drones: &[Drone], now: i64) -> Inbox {
131 let listen = listen_of(home, drones);
132 if std::fs::metadata(msg::path(home)).is_err() {
133 return Inbox { listen, ..Inbox::default() };
134 }
135 let all: Vec<Message> = msg::read(home).into_iter().filter(|m| now - m.ms <= msg::TTL_MS).collect();
136 let here = msg::node(home);
137 let unread = unread_of(home, drones, &here, &all, now);
138 let day = all.iter().filter(|m| now - m.ms <= DAY_MS).count();
139 let mut recent = all;
140 recent.sort_by(|a, b| (b.ms, &b.id).cmp(&(a.ms, &a.id)));
141 recent.truncate(RECENT);
142 Inbox { unread, recent, day, here, listen }
143}
144
145pub fn acknowledge(home: &Path, cwd: &str, shown: &[&Message], done: &mut HashSet<String>, now: i64) {
154 let to_engineer: Vec<&Message> =
155 shown.iter().copied().filter(|m| matches!(Address::parse(&m.to), Some(Address::Engineer { .. })) && !done.contains(&m.id)).collect();
156 if to_engineer.is_empty() {
157 return;
158 }
159 let _ = msg::ack_engineer(home, cwd, &to_engineer, now);
160 done.extend(to_engineer.iter().map(|m| m.id.clone()));
161}
162
163pub fn engineer_looks<'a>(caller: impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String>, mut drones: impl Iterator<Item = &'a [Drone]>) -> bool {
173 drones.all(|ds| matches!(caller(ds), Ok(None)))
174}
175
176pub(crate) fn engineer_dir(dir: Option<&str>) -> String {
179 let d = dir.map_or_else(|| std::env::current_dir().unwrap_or_default(), std::path::PathBuf::from);
180 std::path::absolute(&d).unwrap_or(d).display().to_string()
181}
182
183fn listen_of(home: &Path, drones: &[Drone]) -> BTreeMap<String, crate::listen::Status> {
185 drones
186 .iter()
187 .filter(|d| d.node.is_empty() && d.state != State::Ended)
188 .filter_map(|d| crate::listen::status(home, &d.p.session).map(|s| (d.p.session.clone(), s)))
189 .collect()
190}
191
192pub fn wake_cell(inbox: &Inbox, d: &Drone) -> &'static str {
195 inbox.listen.get(&d.p.session).map_or("-", |s| s.name())
196}
197
198pub fn unread_of(home: &Path, drones: &[Drone], here: &str, all: &[Message], now: i64) -> BTreeMap<String, usize> {
201 drones
202 .iter()
203 .filter(|d| d.node.is_empty() && d.state != State::Ended)
204 .map(|d| (d.p.session.clone(), msg::unread(home, all, here, &d.p.session, d.p.started, now)))
205 .collect()
206}
207
208fn drone_at<'a>(drones: &'a [Drone], here: &str, node: &str, session: &str) -> Option<&'a Drone> {
210 drones.iter().find(|d| d.p.session == session && if d.node.is_empty() { node == here } else { d.node == node })
211}
212
213pub fn sender(drones: &[Drone], here: &str, m: &Message) -> String {
218 let name = match Address::parse(&m.by) {
219 Some(Address::Engineer { email }) if email.is_empty() => msg::ENGINEER.into(),
220 Some(Address::Engineer { email }) => collective::aliases().display(&email),
221 _ => match drone_at(drones, here, &m.node, &m.by) {
222 Some(d) => d.designation.clone(),
223 None => msg::sender_label(m),
224 },
225 };
226 if m.node == here { name } else { format!("{name} @{}", msg::short_node(&m.node)) }
227}
228
229pub fn recipient(drones: &[Drone], here: &str, to: &str) -> String {
234 match Address::parse(to) {
235 Some(Address::Engineer { email }) if !email.is_empty() => collective::aliases().display(&email),
236 Some(Address::Local { node }) if node == here => msg::LOCAL.into(),
237 Some(Address::Local { node }) => format!("{} @{}", msg::LOCAL, msg::short_node(&node)),
238 Some(Address::All) => msg::COLLECTIVE.into(),
239 Some(Address::Session { id, node }) => {
240 let d = match node {
241 Some(n) => drone_at(drones, here, &n, &id),
242 None => drones.iter().find(|d| d.p.session == id),
243 };
244 d.map_or_else(|| msg::short_session(&id), |d| d.designation.clone())
245 }
246 Some(a) => a.to_string(),
247 None => msg::short_session(to),
248 }
249}
250
251pub fn message_line(drones: &[Drone], here: &str, m: &Message) -> String {
254 let ask = if m.asks() { " asks" } else { "" };
255 format!("{}{ask} -> {}: {}", sender(drones, here, m), recipient(drones, here, &m.to), m.text)
256}
257
258pub fn unread_cell(inbox: &Inbox, d: &Drone) -> String {
260 inbox.unread.get(&d.p.session).map_or_else(|| "-".into(), usize::to_string)
261}
262
263pub fn resolve<'a>(drones: &'a [Drone], target: &str) -> Result<Option<&'a Drone>, String> {
269 let t = target.trim();
270 if [msg::ALL, msg::COLLECTIVE, msg::LOCAL].iter().any(|w| t.eq_ignore_ascii_case(w)) {
271 return Ok(None);
272 }
273 let first = t.split_whitespace().next().unwrap_or_default();
274 let number = first.parse::<usize>().ok().or_else(|| NUMBERS.iter().position(|w| w.eq_ignore_ascii_case(first)).map(|i| i + 1));
275 if let Some(n) = number {
276 return drones.iter().find(|d| d.number == n).map(Some).ok_or_else(|| format!("no drone {target}: there are {} (`borg top`)", drones.len()));
277 }
278 let mut hits: Vec<&Drone> = drones.iter().filter(|d| d.p.session == t || (t.len() >= 4 && d.p.session.starts_with(t))).collect();
279 if hits.iter().any(|d| d.state != State::Ended) {
280 hits.retain(|d| d.state != State::Ended);
281 }
282 if hits.iter().any(|d| d.node.is_empty()) {
283 hits.retain(|d| d.node.is_empty());
284 }
285 match hits.as_slice() {
286 [d] => Ok(Some(d)),
287 [] => Err(format!("no drone or session {target:?} (`borg top`)")),
288 _ => Err(format!("{target:?} names {} sessions: give more of the id", hits.len())),
289 }
290}
291
292pub const COLUMN_NODE_CHARS: usize = 6;
296
297pub fn column_node(node: &str) -> String {
299 node.chars().take(COLUMN_NODE_CHARS).collect()
300}
301
302const MESSAGE_CHARS: usize = 120;
305
306fn cut(s: &str, n: usize) -> String {
307 if s.chars().count() <= n { s.to_string() } else { format!("{}…", s.chars().take(n - 1).collect::<String>()) }
308}
309
310pub fn shows_last(total: usize, last: usize, width: usize) -> bool {
315 total <= width || total.saturating_sub(last) < width.saturating_sub(1)
316}
317
318fn prints_text(drones: &[Drone], here: &str, m: &Message) -> bool {
322 shows_last(message_line(drones, here, m).chars().count(), m.text.chars().count(), MESSAGE_CHARS)
323}
324
325pub fn render(name: &str, share: Share, drones: &[Drone], inbox: &Inbox, now: i64) -> String {
327 let working = drones.iter().filter(|d| matches!(d.state, State::Working | State::Waiting)).count();
328 let remote = drones.iter().filter(|d| !d.node.is_empty()).count();
329 let share = match share {
330 Share::Team => "team",
331 Share::Local => "local, never leaves this machine",
332 };
333 let mut s = format!("{name} ({share}): {} drone(s), {working} active, {remote} remote\n", drones.len());
334 if !drones.is_empty() {
335 s.push_str(&drone_table(drones, inbox, now));
336 }
337 if !inbox.recent.is_empty() {
338 s.push_str(&format!(" messages: {} in the last day, {} unread\n", inbox.day, inbox.unread.values().sum::<usize>()));
339 for m in &inbox.recent {
340 s.push_str(&format!(" {:>6} {}\n", ago(now - m.ms), cut(&message_line(drones, &inbox.here, m), MESSAGE_CHARS)));
341 }
342 }
343 s
344}
345
346fn drone_table(drones: &[Drone], inbox: &Inbox, now: i64) -> String {
347 let mut s = format!(
348 " {:<18} {:<8} {:<24} {:<14} {:<8} {:<6} {:>6} {:>6} {:<40} {}\n",
349 "DRONE", "SESSION", "ENGINEER", "REPO", "STATE", "WAKE", "IDLE", "UNREAD", "DOING", "FILES"
350 );
351 for d in drones {
352 let state = if d.paused { "paused" } else { d.state.name() };
353 let engineer = if d.node.is_empty() { d.p.engineer.clone() } else { format!("{} @{}", d.p.engineer, column_node(&d.node)) };
354 let line = format!(
355 " {:<18} {:<8} {:<24} {:<14} {:<8} {:<6} {:>6} {:>6} {:<40} {}",
356 d.designation,
357 cut(&d.p.session, 8),
358 cut(if engineer.is_empty() { "?" } else { &engineer }, 24),
359 cut(&d.p.repo, 14),
360 state,
361 wake_cell(inbox, d),
362 ago(now - d.p.ms),
363 unread_cell(inbox, d),
364 cut(&d.p.prompt, 40),
365 d.p.files.join(", ")
366 );
367 s.push_str(line.trim_end());
368 s.push('\n');
369 }
370 s
371}
372
373pub fn to_json(name: &str, share: Share, drones: &[Drone]) -> Value {
374 json!({"collective": name, "share": if share == Share::Team { "team" } else { "local" }, "drones": drones})
375}
376
377pub fn to_json_with(name: &str, share: Share, drones: &[Drone], inbox: &Inbox) -> Value {
379 let mut v = to_json(name, share, drones);
380 if let Some(ds) = v["drones"].as_array_mut() {
381 for (o, d) in ds.iter_mut().zip(drones) {
382 if let Some(n) = inbox.unread.get(&d.p.session) {
383 o["unread"] = json!(n);
384 }
385 if let Some(l) = inbox.listen.get(&d.p.session) {
386 o["listen"] = json!(l);
387 }
388 }
389 }
390 v["messages"] = json!({"day": inbox.day, "recent": inbox.recent});
391 v
392}
393
394#[derive(Debug, Clone, Default, PartialEq, Serialize)]
404pub struct Selection {
405 pub policy: &'static str,
408 pub default: &'static str,
411 pub source: &'static str,
414 pub gate: f64,
419 pub allowance: f64,
424 pub ledger: f64,
425 pub fit: String,
427 pub at: u64,
429 pub checks: u64,
431 pub sessions: u64,
432 pub deviations: Vec<Count>,
437 pub fallbacks: Vec<Count>,
441 pub spend: Vec<Spend>,
446 pub budget: Option<Budget>,
448 pub effects: Vec<Estimate>,
451 pub half_life: Option<u32>,
453 pub labels: Vec<Count>,
457 pub declaration: Option<Estimate>,
459 pub delivery: Option<select::policy::Rates>,
462 pub evaluation: Option<Evaluation>,
464 pub measurement: Option<Measurement>,
467 pub pool: Vec<Yield>,
469 pub latency: Vec<Latency>,
471 pub test: Option<Test>,
474 pub judge: Vec<crate::judge_length::Length>,
478 pub boundary: f64,
483 pub misses: Option<crate::pools::Misses>,
484}
485
486#[derive(Debug, Clone, Default, PartialEq, Serialize)]
491pub struct Test {
492 pub risk: f64,
495 pub set: bool,
497 pub ran: u64,
500 pub control: u64,
501 pub refused: u64,
504 pub sessions: u64,
506 pub kept: u64,
509 pub by_k: Vec<Draws>,
511 pub due: bool,
518 pub due_checks: u64,
519 pub due_sessions: u64,
520 pub due_min_sessions: u64,
521 pub dilution: Vec<Estimate>,
522 pub ask_free: Option<Estimate>,
523}
524
525#[derive(Debug, Clone, Default, PartialEq, Serialize)]
527pub struct Draws {
528 pub k: u32,
529 pub draws: Vec<u64>,
530 pub refused: u64,
531}
532
533impl Test {
534 #[must_use]
537 pub fn of(summary: &select::summary::Summary, risk: Option<f64>) -> Test {
538 use select::replay::{TEST_CHECKS, TEST_MIN_SESSIONS, TEST_SESSIONS, test_readout};
539 let e = &summary.experiment;
540 let (ran, refused) = e.totals();
541 let r = test_readout(summary);
542 let estimate = |name: String, unit: &str, x: &crate::ope::Estimate| {
544 x.value.map(|v| Estimate { name, unit: unit.into(), mean: 100.0 * v, lo: 100.0 * x.lo, hi: 100.0 * x.hi, n: x.n as u64 })
545 };
546 let mut by_k: Vec<Draws> = Vec::new();
547 let at = |by_k: &mut Vec<Draws>, k: u32| match by_k.iter().position(|d| d.k == k) {
548 Some(i) => i,
549 None => {
550 by_k.push(Draws { k, draws: vec![0; k as usize + 1], refused: 0 });
551 by_k.len() - 1
552 }
553 };
554 for &(k, s, n) in &e.ran {
555 let i = at(&mut by_k, k);
556 if let Some(c) = by_k[i].draws.get_mut(s as usize) {
557 *c += n;
558 }
559 }
560 for &(k, n) in &e.refused {
561 let i = at(&mut by_k, k);
562 by_k[i].refused += n;
563 }
564 by_k.sort_by_key(|d| d.k);
565 Test {
566 risk: risk.unwrap_or(select::policy::ALPHA),
567 set: risk.is_some(),
568 ran,
569 control: e.control,
570 refused,
571 sessions: u64::from(e.sessions),
572 kept: r.kept,
573 by_k,
574 due: r.due,
575 due_checks: TEST_CHECKS,
576 due_sessions: u64::from(TEST_SESSIONS),
577 due_min_sessions: u64::from(TEST_MIN_SESSIONS),
578 dilution: r.by_asked.iter().filter_map(|(m, x)| estimate(format!("{m} asked − 1 asked"), "points changed", x)).collect(),
579 ask_free: r.ask_free.as_ref().and_then(|x| estimate("grade 2 asked − handed over".into(), "points changed", x)),
580 }
581 }
582}
583
584#[derive(Debug, Clone, Default, PartialEq, Serialize)]
586pub struct Count {
587 pub name: String,
588 pub part: u64,
589 pub whole: u64,
590}
591
592#[derive(Debug, Clone, Default, PartialEq, Serialize)]
594pub struct Spend {
595 pub source: String,
597 pub calls: u64,
598 pub rounds: u64,
600 pub usd: f64,
601}
602
603#[derive(Debug, Clone, Default, PartialEq, Serialize)]
607pub struct Budget {
608 pub tokens: f64,
609 pub spent: f64,
610 pub price: f64,
611}
612
613#[derive(Debug, Clone, Default, PartialEq, Serialize)]
615pub struct Estimate {
616 pub name: String,
617 pub unit: String,
622 pub mean: f64,
623 pub lo: f64,
624 pub hi: f64,
625 pub n: u64,
627}
628
629#[derive(Debug, Clone, Default, PartialEq, Serialize)]
632pub struct Evaluation {
633 pub sessions: u64,
635 pub mean: f64,
638 pub ucb: Option<f64>,
639 pub units: u64,
641 pub ess: f64,
642 pub max_weight: f64,
644 pub truncated: u64,
646}
647
648#[derive(Debug, Clone, Default, PartialEq, Serialize)]
653pub struct Measurement {
654 pub placebos: u64,
656 pub judged: u64,
658 pub tail_shown: u64,
661 pub tail_applied: u64,
662}
663
664#[derive(Debug, Clone, Default, PartialEq, Serialize)]
667pub struct Yield {
668 pub from: u32,
669 pub to: u32,
670 pub shown: u64,
671 pub applied: u64,
672 pub rate: f64,
673}
674
675#[derive(Debug, Clone, Default, PartialEq, Serialize)]
678pub struct Latency {
679 pub harness: String,
680 pub checks: u64,
681 pub p50_ms: u64,
683 pub p95_ms: u64,
684 pub budget_ms: Option<u64>,
686 pub over: u64,
689}
690
691const QUANTILES: [f64; 2] = [0.50, 0.95];
694
695fn quantile(sorted: &[u64], q: f64) -> u64 {
697 let rank = ((q * sorted.len() as f64).ceil() as usize).clamp(1, sorted.len().max(1));
698 sorted.get(rank - 1).copied().unwrap_or(0)
699}
700
701impl Selection {
702 #[must_use]
706 pub fn of(summary: &select::summary::Summary, policy: select::Policy, risk: Option<f64>) -> Selection {
707 use select::summary::{LIVE, Label, STOP};
708 let s = summary;
709 let learned = policy == select::Policy::Learned;
710 let reverted = learned && s.monitor.reverted;
711 let default = select::precedence(None, None, false).name();
712 let count = |name: &str, part: usize, whole: usize| Count { name: name.into(), part: part as u64, whole: whole as u64 };
713 let (sessions, checks) = s.checked();
715 let deviations = match s.gate.checks {
716 0 => Vec::new(),
717 n => vec![count("route", s.gate.deviated as usize, n as usize)],
718 };
719 let fallbacks = select::summary::FELL_BACK
721 .iter()
722 .zip(s.fell_back)
723 .filter(|_| checks > 0)
724 .map(|(reason, n)| Count { name: (*reason).into(), part: n, whole: checks })
725 .collect();
726 let spend = [("stop", STOP), ("live", LIVE)]
727 .into_iter()
728 .filter(|(_, i)| s.spend[*i].calls > 0)
729 .map(|(source, i)| Spend { source: source.into(), calls: s.spend[i].calls, rounds: s.spend[i].rounds, usd: s.spend[i].usd })
730 .collect();
731 let budget = s.fit.budget.map(|tokens| Budget { tokens, spent: s.pool.tokens as f64 / sessions.max(1) as f64, price: s.fit.lambda });
732 let judged = |l: Label| l.verdict();
734 let of = |route: select::Route| {
740 let of: Vec<_> = s.applies.iter().filter(|a| a.route == route && (route != select::Route::Free || !a.had)).collect();
741 (of.iter().filter(|a| judged(a.label)).count(), of.len())
742 };
743 let mut labels = Vec::new();
744 for (name, route) in [("judged, asked", select::Route::Ask), ("judged, free", select::Route::Free), ("judged, placebo", select::Route::Placebo)] {
745 let (part, whole) = of(route);
746 labels.push(count(name, part, whole));
747 }
748 let answers: Vec<_> = s.applies.iter().filter(|a| a.jv >= crate::predict::PURE_UNCLEAR_FROM && (judged(a.label) || a.label.pending())).collect();
753 let share = |name: &str, label: Label, route: Option<select::Route>| {
754 let of: Vec<_> = answers.iter().filter(|a| route.is_none_or(|r| a.route == r)).collect();
755 count(name, of.iter().filter(|a| a.label == label).count(), of.len())
756 };
757 for (name, label) in [("unclear", Label::Unclear), ("not arisen", Label::NotArisen), ("consequence remains", Label::Remains)] {
758 labels.push(share(name, label, None));
759 }
760 for (name, route) in [("unclear, asked", select::Route::Ask), ("unclear, free", select::Route::Free), ("unclear, placebo", select::Route::Placebo)] {
761 labels.push(share(name, Label::Unclear, Some(route)));
762 }
763 let later = s.applies.iter().filter(|a| matches!(a.route, select::Route::Not | select::Route::Placebo)).count();
764 let (placebos_judged, placebos) = of(select::Route::Placebo);
765 labels.push(count("placebo", placebos, later));
766 labels.retain(|c| c.whole > 0);
767 let measurement = (checks > 0).then_some(Measurement {
768 placebos: placebos as u64,
769 judged: placebos_judged as u64,
770 tail_shown: s.pool.tail.0,
771 tail_applied: s.pool.tail.1,
772 });
773 let share = |stripped: bool| {
775 let of: Vec<_> = s.applies.iter().filter(|a| a.route == select::Route::Ask && judged(a.label) && a.stripped == stripped).collect();
776 (of.iter().filter(|a| a.label == Label::Changed).count() as f64, of.len() as f64)
777 };
778 let ((c1, n1), (c0, n0)) = (share(false), share(true));
779 let declaration = (n1 > 0.0 && n0 > 0.0).then(|| {
780 let (r1, r0) = (c1 / n1, c0 / n0);
781 let se = (r1 * (1.0 - r1) / n1 + r0 * (1.0 - r0) / n0).sqrt();
782 let mean = 100.0 * (r1 - r0);
783 Estimate { name: "declaration".into(), unit: "points".into(), mean, lo: mean - 100.0 * Z95 * se, hi: mean + 100.0 * Z95 * se, n: (n1 + n0) as u64 }
784 });
785 let pool = s
786 .pool
787 .bins
788 .iter()
789 .enumerate()
790 .filter(|(_, (shown, _))| *shown > 0)
791 .map(|(b, &(shown, applied))| Yield {
792 from: (1u32 << b) - 1,
793 to: (1u32 << (b + 1)) - 1,
794 shown,
795 applied,
796 rate: s.fit.pool.yields.get(b).map_or(applied as f64 / shown as f64, |(a, b)| a / (a + b)),
797 })
798 .collect();
799 let (allowance, ledger) = if learned && !reverted { select::policy::gate_parts(&s.gate, risk.unwrap_or(select::policy::ALPHA)) } else { (0.0, 0.0) };
800 let mon = &s.monitor;
801 let evaluation = (mon.sessions > 0).then(|| Evaluation {
802 sessions: u64::from(mon.sessions),
803 mean: mon.mean,
804 ucb: mon.ucb,
805 units: u64::from(mon.labelled),
806 ess: mon.ess,
807 max_weight: mon.max_weight,
808 truncated: u64::from(mon.truncated),
809 });
810 let latency = s
811 .pool
812 .stops
813 .iter()
814 .filter(|(_, calls)| !calls.is_empty())
815 .map(|(harness, calls)| {
816 let mut ms: Vec<u64> = calls.iter().map(|&(ms, _)| u64::from(ms)).collect();
817 ms.sort_unstable();
818 let budget = crate::harness::ALL.iter().find(|h| h.name() == harness).and_then(|h| h.caps().stop_budget);
819 Latency {
820 harness: harness.clone(),
821 checks: ms.len() as u64,
822 p50_ms: quantile(&ms, QUANTILES[0]),
823 p95_ms: quantile(&ms, QUANTILES[1]),
824 budget_ms: budget.map(|b| u64::try_from(b.as_millis()).unwrap_or(u64::MAX)),
825 over: calls.iter().filter(|(_, failed)| *failed).count() as u64,
826 }
827 })
828 .collect();
829 Selection {
830 policy: policy.name(),
831 default,
832 source: if reverted { select::REVERTED_SOURCE } else { policy.name() },
833 gate: if learned && !reverted { select::policy::remaining_at(&s.gate, risk.unwrap_or(select::policy::ALPHA)) } else { 0.0 },
834 allowance,
835 ledger,
836 fit: s.fit.key.clone(),
837 at: s.at,
838 checks,
839 sessions,
840 deviations,
841 fallbacks,
842 spend,
843 budget,
844 effects: s.fit.model.as_ref().map(|m| effects(m, s.fit.labels)).unwrap_or_default(),
845 half_life: s.fit.model.as_ref().map(|_| s.fit.half_life.unwrap_or(0)),
846 labels,
847 declaration,
848 delivery: (!s.fit.key.is_empty()).then(|| s.fit.rates.clone()),
849 evaluation,
850 measurement,
851 pool,
852 latency,
853 test: (learned || s.experiment.totals() != (0, 0)).then(|| Test::of(s, risk)),
854 judge: Vec::new(),
855 boundary: 0.0,
856 misses: None,
857 }
858 }
859
860 #[must_use]
863 pub fn shown(&self) -> bool {
864 self.checks > 0 || self.source != self.default
865 }
866}
867
868const Z95: f64 = 1.96;
870
871fn effects(m: &select::policy::Model, n: u32) -> Vec<Estimate> {
875 let p = select::policy::P;
876 if m.beta.len() != p || m.chol.len() != p * p {
877 return Vec::new();
878 }
879 let sd = |k: usize| m.chol[k * p..=k * p + k].iter().map(|x| x * x).sum::<f64>().sqrt();
881 let mut out: Vec<Estimate> = (0..p)
882 .map(|k| Estimate {
883 name: select::policy::NAMES[k].into(),
884 unit: "coefficient".into(),
885 mean: m.beta[k],
886 lo: m.beta[k] - Z95 * sd(k),
887 hi: m.beta[k] + Z95 * sd(k),
888 n: u64::from(n),
889 })
890 .collect();
891 for (name, v) in ["session spread", "lesson spread", "judge spread"].into_iter().zip(m.spread) {
892 out.push(Estimate { name: name.into(), unit: "logit".into(), mean: v, lo: v, hi: v, n: u64::from(n) });
893 }
894 out
895}
896
897#[must_use]
899pub fn selection(lessons: &Path, policy: select::Policy) -> Selection {
900 let mut s = Selection::of(&select::summary::load(lessons), policy, select::risk_of(lessons));
901 let current = crate::hook::judge_model(&std::env::var("BORG_PREDICT_MODEL").unwrap_or_default());
903 s.judge = crate::judge_length::lengths(&lessons.with_file_name("state"), ¤t);
904 s.boundary = crate::whole::boundary_share();
905 s.misses = lessons.parent().map(crate::pools::misses).filter(|m| m.checks > 0);
907 s
908}
909
910fn grouped(n: usize) -> String {
912 let digits = n.to_string();
913 let mut out = String::new();
914 for (i, c) in digits.chars().enumerate() {
915 if i > 0 && (digits.len() - i).is_multiple_of(3) {
916 out.push(',');
917 }
918 out.push(c);
919 }
920 out
921}
922
923#[must_use]
926pub fn judge_length_line(l: &crate::judge_length::Length) -> String {
927 use crate::judge_length::Decision;
928 let start = crate::hook::END_START_BUDGET;
929 let how = if l.start {
930 "the start: no decided step moved it".to_string()
931 } else if l.chars > start {
932 format!("measured, up from the start {}", grouped(start))
933 } else {
934 format!("measured, down from the start {}", grouped(start))
935 };
936 let mut parts = vec![format!("{}: session-end budget {} chars ({how})", l.model, grouped(l.chars))];
937 let measured: Vec<String> = l
938 .steps
939 .iter()
940 .filter(|s| s.sent > 0)
941 .map(|s| {
942 let what = match s.decision {
943 Decision::Reference => format!("reference, line {:.0}%", 100.0 * l.line.unwrap_or(0.0)),
944 Decision::Below => "below the reference".into(),
945 Decision::Holds => "holds".into(),
946 Decision::Fails => "fails".into(),
947 Decision::Open => "open".into(),
948 };
949 format!("≤{} {:.0}% of {} [{:.0}–{:.0}%] {what}", grouped(s.upto), 100.0 * s.answered as f64 / s.sent as f64, s.sent, 100.0 * s.lo, 100.0 * s.hi)
950 })
951 .collect();
952 parts.push(if measured.is_empty() { "no judge call measured yet".into() } else { format!("answered by prompt size: {}", measured.join(", ")) });
953 if l.line.is_none() && !measured.is_empty() {
954 parts.push(format!(
955 "no reference yet (the smallest steps with {} lessons in {} calls)",
956 crate::judge_length::REFERENCE_LESSONS,
957 crate::judge_length::MIN_CALLS
958 ));
959 }
960 parts.push(match l.explore {
961 Some(up) if l.steps.iter().any(|s| s.upto == up && s.decision == Decision::Fails) => {
962 format!("exploring the failed ≤{} on {:.0}% of session-end passes", grouped(up), 100.0 * l.psi)
963 }
964 Some(up) => format!("exploring ≤{} on {:.0}% of session-end passes", grouped(up), 100.0 * l.psi),
965 None => "nothing explored".into(),
966 });
967 parts.join(" · ")
968}
969
970#[must_use]
979pub fn selection_line(s: &Selection) -> String {
980 let default = s.source == s.default;
981 let head = match s.source {
982 select::REVERTED_SOURCE => "REVERTED to the fixed rule: the monitor saw harm beyond the gate's share (sticky)".to_string(),
983 "learned" => {
984 let opt = if default { "" } else { " (opt-in, BORG_SELECT=learned)" };
985 if s.gate > 0.0 {
986 format!("learned{opt}, gate {:.3} (quality it may still risk below the fixed rule)", s.gate)
987 } else if s.fit.is_empty() || s.allowance <= 0.0 {
988 format!("learned{opt}, cold start: gate 0, every decision the fixed rule's")
989 } else {
990 format!("learned{opt}, gate spent (ledger q05 {:+.3} against allowance {:.3}), deviations need positive evidence", s.ledger, s.allowance)
993 }
994 }
995 "fixed" if default => "the fixed rule (the default; BORG_SELECT=learned opts in to the learned policy)".into(),
996 "fixed" => "the fixed rule (BORG_SELECT=fixed)".into(),
997 "cap" => "the fixed rule with an ask cap (BORG_STOP_ASK_CAP, evaluation only)".into(),
998 "attention" => "attention's cut (BORG_STOP_ATTENTION=on, evaluation only)".into(),
999 other => other.to_string(),
1000 };
1001 if s.checks == 0 {
1002 return format!("{head} · no graded check folded yet");
1003 }
1004 let mut parts = vec![head, format!("{} checks", s.checks)];
1005 if !s.deviations.is_empty() {
1006 parts.push(format!("deviated: {}", s.deviations.iter().map(|c| format!("{} {}/{}", c.name, c.part, c.whole)).collect::<Vec<_>>().join(", ")));
1007 }
1008 let fell: Vec<String> = s.fallbacks.iter().filter(|c| c.part > 0).map(|c| format!("{} {}", c.name, c.part)).collect();
1009 if !fell.is_empty() {
1010 parts.push(format!("learned decisions the fixed rule took over: {}", fell.join(", ")));
1011 }
1012 if let Some(t) = &s.test {
1013 parts.push(test_line(t));
1014 }
1015 if let Some(m) = &s.measurement {
1016 parts.push(format!("placebos {} drawn, {} judged", m.placebos, m.judged));
1017 if m.tail_shown > 0 {
1018 parts.push(format!("tail {} shown past the pool, {} applied", m.tail_shown, m.tail_applied));
1019 }
1020 }
1021 if let Some(m) = &s.misses {
1022 parts.push(misses_line(m, s.boundary));
1023 }
1024 if let Some(e) = &s.evaluation {
1025 let reading = match e.ucb {
1026 Some(u) => format!("monitor {:+.3}, bound {u:+.3} over {} sessions (below 0 reverts)", e.mean, e.sessions),
1027 None => format!("monitor {:+.3} over {} sessions, no bound yet", e.mean, e.sessions),
1028 };
1029 parts.push(format!("{reading}, ESS {:.0} of {} weighed, largest weight {:.1}× the mean", e.ess, e.units, e.max_weight));
1030 }
1031 if !s.latency.is_empty() {
1032 let secs = |ms: u64| format!("{:.1} s", ms as f64 / 1000.0);
1033 let each: Vec<String> = s
1034 .latency
1035 .iter()
1036 .map(|l| {
1037 let budget = l.budget_ms.map_or_else(String::new, |b| format!(" of {}", secs(b)));
1038 let failed = if l.over > 0 { format!(", {} failed", l.over) } else { String::new() };
1039 format!("{} {}{budget}{failed}", l.harness, secs(l.p95_ms))
1040 })
1041 .collect();
1042 parts.push(format!("stop p95 {}", each.join("; ")));
1043 }
1044 if s.sessions > 0 && !s.spend.is_empty() {
1045 let n = s.sessions as f64;
1046 let (usd, rounds) = s.spend.iter().fold((0.0, 0), |(u, r), x| (u + x.usd, r + x.rounds));
1047 parts.push(format!("${:.2} and {:.1} extra rounds per session ({} sessions)", usd / n, rounds as f64 / n, s.sessions));
1048 }
1049 if let Some(b) = &s.budget {
1050 parts.push(format!("budget {:.0} pool tokens per session: {:.0} read, price {:.2e} per token", b.tokens, b.spent, b.price));
1051 }
1052 parts.join(" · ")
1053}
1054
1055#[must_use]
1059pub fn misses_line(m: &crate::pools::Misses, share: f64) -> String {
1060 let size = format!("{:.1} MB", m.at as f64 / 1_048_576.0);
1061 let explore = if share > 0.0 {
1062 format!("exploration {:.0}% of the budget and of the dropped lessons", 100.0 * share)
1063 } else {
1064 "exploration off (BORG_BOUNDARY=off)".into()
1065 };
1066 let head = format!("pool log {} checks ({size}), {explore}", m.checks);
1067 if m.positives == 0 {
1068 return format!("{head}, no first apply at grade ≥ 2 yet");
1069 }
1070 let mut causes: Vec<&(String, u64)> = m.causes.iter().filter(|(_, n)| *n > 0).collect();
1071 causes.sort_by_key(|c| std::cmp::Reverse(c.1));
1072 let why = if causes.is_empty() { String::new() } else { format!(": {}", causes.iter().map(|(c, n)| format!("{c} {n}")).collect::<Vec<_>>().join(", ")) };
1073 format!("{head}, {} of {} first applies at grade ≥ 2 ranked low ({} shown by a draw){why}", m.low, m.positives, m.drawn)
1074}
1075
1076#[must_use]
1082pub fn async_line(t: &crate::stats::AsyncTally, on: bool, limit: std::time::Duration, line: &str) -> String {
1083 let head = if on { "detached, the stop never waits (BORG_ASYNC=off checks at the stop)" } else { "at the stop, which waits for it (BORG_ASYNC=off)" };
1084 let mut parts = vec![head.to_string()];
1085 if on {
1086 parts.push(format!("call limit now {} s", limit.as_secs()));
1087 }
1088 if t.checks + t.superseded + t.gone + t.unstarted > 0 {
1089 let at_limit = format!("{} at the limit, {} s last", t.timed_out, t.limit_ms / 1000);
1090 parts.push(format!("{} detached checks (${:.2}), {} failed ({at_limit})", t.checks, t.usd, t.failed));
1091 parts.push(format!("{} superseded by a newer stop, {} after the session ended", t.superseded, t.gone));
1092 if t.unstarted > 0 {
1093 parts.push(format!("{} never started (their job could not be written or run)", t.unstarted));
1094 }
1095 parts.push(format!("peak memory {} MB", t.rss_kb.div_ceil(1024)));
1096 parts.push(format!("{:.1} lessons shown per check (line {line})", t.lessons));
1097 }
1098 parts.join(" · ")
1099}
1100
1101fn test_line(t: &Test) -> String {
1106 let risk = if t.set { format!("risk {:.2} (the collective's)", t.risk) } else { format!("risk {:.2}", t.risk) };
1107 match (t.ran, t.risk <= 0.0) {
1108 (0, true) => return "question-count test off (risk 0)".into(),
1109 (0, false) => return format!("question-count test: no check yet, {risk}"),
1110 _ => {}
1111 }
1112 let kept = if t.kept < t.ran { format!(", {} still in the summary", t.kept) } else { String::new() };
1113 let head =
1114 format!("question-count test {} drawn ({} control, {} refused by the gate{kept}) in {} sessions, {risk}", t.ran, t.control, t.refused, t.sessions);
1115 match (&t.ask_free, t.due) {
1116 (Some(e), true) => format!("{head}; readout: grade 2 asked − handed over {:+.1} [{:+.1}, {:+.1}] points changed", e.mean, e.lo, e.hi),
1117 (None, true) => format!("{head}; readout due, no judged grade 2 on both routes yet"),
1118 (_, false) => format!("{head}; readout after {} checks in {} sessions, or {} sessions", t.due_checks, t.due_min_sessions, t.due_sessions),
1119 }
1120}
1121
1122#[must_use]
1124pub fn render_with(name: &str, share: Share, drones: &[Drone], inbox: &Inbox, select: Option<&Selection>, now: i64) -> String {
1125 let mut s = render(name, share, drones, inbox, now);
1126 if let Some(sel) = select.filter(|s| s.shown()) {
1127 s.push_str(&format!(" selection: {}\n", selection_line(sel)));
1128 }
1129 s
1130}
1131
1132pub fn version_line<'a>(stores: impl IntoIterator<Item = &'a Path>, checked: Option<&update::Checked>, check: bool) -> Option<String> {
1138 let mut parts: Vec<String> = Vec::new();
1139 if let Some(l) = checked.and_then(|c| c.latest.as_deref()).filter(|l| check && update::newer(l, update::VERSION)) {
1140 parts.push(format!("borg {l} is released, this is {}: borg update", update::VERSION));
1141 }
1142 let aliases = collective::aliases();
1143 let ours = protocol::current();
1144 for home in stores.into_iter().filter(|h| h.join("mesh").is_dir()) {
1145 let versions = protocol::versions(home);
1146 let mismatches = protocol::mismatches(home);
1147 let name = |key: &str| aliases.display(&protocol::member(key, versions.get(key)));
1148 for (key, m) in &mismatches {
1149 parts.push(format!("nothing syncs with {} ({}, this node {})", name(key), m.theirs, m.ours));
1150 }
1151 for (key, v) in versions.iter().filter(|(k, v)| v.protocol < ours && !mismatches.contains_key(*k)) {
1152 parts.push(format!("{} on /{} misses {}", name(key), v.protocol, protocol::misses(v.protocol, ours)));
1153 }
1154 }
1155 parts.dedup();
1156 (!parts.is_empty()).then(|| parts.join(" · "))
1157}
1158
1159fn routes(dir: Option<&str>) -> Result<Vec<Route>, String> {
1161 let cfg = collective::load()?;
1162 let base = learn::home();
1163 Ok(match dir {
1164 Some(d) => vec![collective::route(cfg.as_ref(), d, &base)].into_iter().filter(|r| *r != Route::Off).collect(),
1165 None => install::collectives(cfg.as_ref(), &base),
1166 })
1167}
1168
1169pub fn gather(dir: Option<&str>, now: i64) -> Result<(Vec<crate::top_live::Coll>, Option<String>), String> {
1172 let mut out = Vec::new();
1173 let routes = routes(dir)?;
1174 let stores: Vec<&Path> = routes
1175 .iter()
1176 .filter_map(|r| match r {
1177 Route::Collective { home, .. } => Some(home.as_path()),
1178 Route::Off => None,
1179 })
1180 .collect();
1181 let version = version_line(stores, update::checked().as_ref(), update::on());
1182 for r in &routes {
1183 let Route::Collective { name, home, share } = r else { continue };
1184 presence::prune(home, now);
1185 let mut ds = drones(home, collective::may_sync(r), now);
1186 for d in &mut ds {
1187 d.p.engineer = collective::aliases().display(&d.p.engineer);
1188 }
1189 let inbox = inbox(home, &ds, now);
1190 if ds.is_empty() && inbox.recent.is_empty() && dir.is_none() {
1191 continue;
1192 }
1193 let lessons = home.join("lessons.jsonl");
1196 let summary = select::summary::peek(&lessons);
1197 let select = Some(Selection::of(&summary, select::policy(), select::risk_of(&lessons))).filter(Selection::shown).map(|s| selection_line(&s));
1198 out.push(crate::top_live::Coll { name: name.clone(), home: home.clone(), share: *share, drones: ds, inbox, select });
1199 }
1200 Ok((out, version))
1201}
1202
1203fn screen(dir: Option<&str>, json_out: bool, now: i64) -> Result<String, String> {
1205 let routes = routes(dir)?;
1206 let text = screen_of(&routes, dir.is_some(), &engineer_dir(dir), msg::caller, json_out, now);
1207 let stores = routes.iter().filter_map(|r| match r {
1208 Route::Collective { home, .. } => Some(home.as_path()),
1209 Route::Off => None,
1210 });
1211 Ok(match version_line(stores, update::checked().as_ref(), update::on()).filter(|_| !json_out) {
1212 Some(v) => format!("{v}\n\n{text}"),
1213 None => text,
1214 })
1215}
1216
1217fn screen_of(
1222 routes: &[Route],
1223 one: bool,
1224 cwd: &str,
1225 caller: impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String>,
1226 json_out: bool,
1227 now: i64,
1228) -> String {
1229 let mut text = String::new();
1230 let mut all = Vec::new();
1231 let (mut gathered, mut printed) = (Vec::new(), Vec::new());
1234 for r in routes {
1235 let Route::Collective { name, home, share } = r else { continue };
1236 presence::prune(home, now);
1237 let mut ds = drones(home, collective::may_sync(r), now);
1238 for d in &mut ds {
1240 d.p.engineer = collective::aliases().display(&d.p.engineer);
1241 }
1242 let inbox = inbox(home, &ds, now);
1243 let sel = selection(&home.join("lessons.jsonl"), select::policy());
1244 if json_out {
1245 let mut v = to_json_with(name, *share, &ds, &inbox);
1246 v["select"] = json!(sel);
1247 all.push(v);
1248 } else if !ds.is_empty() || !inbox.recent.is_empty() || one {
1249 text.push_str(&render_with(name, *share, &ds, &inbox, Some(&sel), now));
1250 text.push('\n');
1251 printed.push((home, inbox.recent.into_iter().filter(|m| prints_text(&ds, &inbox.here, m)).collect::<Vec<_>>()));
1252 }
1253 gathered.push(ds);
1254 }
1255 if !printed.is_empty() && engineer_looks(caller, gathered.iter().map(Vec::as_slice)) {
1256 let mut done = HashSet::new();
1257 for (home, recent) in &printed {
1258 acknowledge(home, cwd, &recent.iter().collect::<Vec<_>>(), &mut done, now);
1259 }
1260 }
1261 if json_out {
1262 return serde_json::to_string_pretty(&all).unwrap_or_default() + "\n";
1263 }
1264 if text.is_empty() {
1265 text = "no drones (borg top lists sessions whose hooks ran in the last day)\n".into();
1266 }
1267 text
1268}
1269
1270pub fn main(args: &[&str]) -> i32 {
1271 let (mut dir, mut once, mut json_out) = (None, false, false);
1272 let mut it = args.iter();
1273 while let Some(a) = it.next() {
1274 match *a {
1275 "--dir" if dir.is_none() => match it.next() {
1276 Some(d) => dir = Some(*d),
1277 None => return usage("--dir needs a value"),
1278 },
1279 "--once" => once = true,
1280 "--json" => json_out = true,
1281 a => return usage(&format!("unknown flag {a}")),
1282 }
1283 }
1284 if !json_out && !once && unsafe { libc::isatty(1) } == 1 {
1286 return crate::top_live::run(dir);
1287 }
1288 match screen(dir, json_out, learn::now_ms()) {
1289 Ok(s) => {
1290 print!("{s}");
1291 0
1292 }
1293 Err(e) => {
1294 eprintln!("borg top: {e}");
1295 1
1296 }
1297 }
1298}
1299
1300fn usage(why: &str) -> i32 {
1301 eprintln!("borg top: {why}\nusage: borg top [--once] [--json] [--dir DIR]");
1302 2
1303}
1304
1305#[cfg(test)]
1306pub(crate) fn tests_fixture() -> Vec<Drone> {
1307 tests::fixture()
1308}
1309
1310#[cfg(test)]
1311pub(crate) fn tests_inbox() -> Inbox {
1312 tests::inbox_fixture()
1313}
1314
1315#[cfg(test)]
1316mod tests {
1317 #[test]
1318 fn the_async_line_says_where_the_check_runs_and_what_it_did() {
1319 use crate::stats::AsyncTally;
1320 let secs = std::time::Duration::from_secs(45);
1321 assert_eq!(super::async_line(&AsyncTally::default(), false, secs, "a2"), "at the stop, which waits for it (BORG_ASYNC=off)");
1322 let fresh = super::async_line(&AsyncTally::default(), true, secs, "a2");
1323 assert_eq!(fresh, "detached, the stop never waits (BORG_ASYNC=off checks at the stop) · call limit now 45 s");
1324 let t =
1325 AsyncTally { checks: 12, failed: 2, timed_out: 1, superseded: 3, gone: 1, unstarted: 0, limit_ms: 45_000, rss_kb: 38_000, lessons: 41.3, usd: 0.5 };
1326 let l = super::async_line(&t, true, secs, "a2");
1327 for part in [
1328 "12 detached checks ($0.50), 2 failed (1 at the limit, 45 s last)",
1329 "3 superseded by a newer stop, 1 after the session ended",
1330 "peak memory 38 MB",
1331 "41.3 lessons shown per check (line a2)",
1332 ] {
1333 assert!(l.contains(part), "{part}: {l}");
1334 }
1335 assert!(!l.contains("never started"), "{l}");
1336 let lost = super::async_line(&AsyncTally { unstarted: 2, ..t }, true, secs, "a2");
1337 assert!(lost.contains("2 never started (their job could not be written or run)"), "{lost}");
1338 }
1339
1340 use super::*;
1341
1342 fn p(session: &str, engineer: &str, started: i64, ms: i64, event: &str, prompt: &str, files: &[&str]) -> Presence {
1343 Presence {
1344 session: session.into(),
1345 harness: "claude".into(),
1346 engineer: engineer.into(),
1347 repo: "borg-ai".into(),
1348 collective: "corp".into(),
1349 prompt: prompt.into(),
1350 files: files.iter().map(|f| f.to_string()).collect(),
1351 event: event.into(),
1352 ms,
1353 started,
1354 pid: None,
1355 }
1356 }
1357
1358 const NOW: i64 = 10_000_000;
1359
1360 pub(crate) fn fixture() -> Vec<Drone> {
1361 let local = vec![
1362 p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "post_tool", "fix the flaky sync test", &["crates/borg-mesh/src/sync.rs", "src/recon.rs"]),
1363 p("cccc3333-x", "alice@x", 300, NOW - presence::IDLE_MS - 60_000, "stop", "write the release notes for the mesh daemon and more words", &[]),
1364 ];
1365 let remote = vec![presence::Remote {
1366 p: p("bbbb2222-y", "bob@y", 200, NOW - 120_000, "stop", "review PR 42", &["README.md"]),
1367 key: "beefcafe".repeat(8),
1368 sig: String::new(),
1369 }];
1370 assemble(local, remote, NOW, |_| true, |p| p.session.starts_with("aaaa"))
1371 }
1372
1373 fn message(by: &str, name: &str, to: &str, kind: msg::Kind, ms: i64, text: &str) -> Message {
1374 message_on("n1", by, name, to, kind, ms, text)
1375 }
1376
1377 fn message_on(node: &str, by: &str, name: &str, to: &str, kind: msg::Kind, ms: i64, text: &str) -> Message {
1378 let from = msg::Sender { by: by.into(), name: name.into(), harness: "claude".into(), node: node.into() };
1379 Message::new(kind, &from, to, None, ms, text)
1380 }
1381
1382 pub(crate) fn inbox_fixture() -> Inbox {
1383 Inbox {
1384 unread: [("aaaa1111-x".to_string(), 2), ("cccc3333-x".to_string(), 0)].into_iter().collect(),
1385 recent: vec![
1386 message("cccc3333-x", "Three of Three", "aaaa1111-x", msg::ASK, NOW - 60_000, "which test\nflakes?"),
1387 message("engineer:alice@x", "alice@x", msg::ALL, msg::TELL, NOW - 120_000, "release at five"),
1388 message_on(&"feed".repeat(16), "gone-session", "Six of Six", "zzzz9999-q", msg::TELL, NOW - 180_000, "done"),
1389 ],
1390 day: 3,
1391 here: "n1".into(),
1392 listen: [("aaaa1111-x".to_string(), crate::listen::Status::Waits), ("cccc3333-x".to_string(), crate::listen::Status::Listens)]
1393 .into_iter()
1394 .collect(),
1395 }
1396 }
1397
1398 #[test]
1399 fn renders_one_screen_per_collective() {
1400 let out = render("corp", Share::Team, &fixture(), &inbox_fixture(), NOW);
1401 let expected = "\
1402corp (team): 3 drone(s), 2 active, 1 remote
1403 DRONE SESSION ENGINEER REPO STATE WAKE IDLE UNREAD DOING FILES
1404 One of Three aaaa111… alice@x borg-ai paused waiter 5s 2 fix the flaky sync test crates/borg-mesh/src/sync.rs, src/recon.rs
1405 Two of Three bbbb222… bob@y @beefca borg-ai waiting - 2m - review PR 42 README.md
1406 Three of Three cccc333… alice@x borg-ai idle on 31m 0 write the release notes for the mesh da…
1407 messages: 3 in the last day, 2 unread
1408 1m Three of Three asks -> One of Three: which test flakes?
1409 2m alice@x -> collective: release at five
1410 3m Six of Six @feedfeedfeed -> zzzz9999: done
1411";
1412 assert_eq!(out, expected, "\n{out}");
1413 let none = Inbox::default();
1414 assert_eq!(render("private", Share::Local, &[], &none, NOW), "private (local, never leaves this machine): 0 drone(s), 0 active, 0 remote\n");
1415 }
1416
1417 fn learned() -> Selection {
1421 let spend = |source: &str, rounds, usd| Spend { source: source.into(), calls: 1, rounds, usd };
1422 let evaluation = Evaluation { sessions: 12, mean: 0.004, ucb: Some(0.031), units: 300, ess: 180.0, max_weight: 6.5, truncated: 3 };
1423 Selection {
1424 policy: "learned",
1425 default: "fixed",
1426 source: "learned",
1427 gate: 0.12,
1428 checks: 480,
1429 sessions: 12,
1430 deviations: vec![Count { name: "route".into(), part: 5, whole: 400 }],
1431 spend: vec![spend("stop", 4, 3.0), spend("live", 0, 2.04)],
1432 evaluation: Some(evaluation),
1433 measurement: Some(Measurement { placebos: 40, judged: 31, tail_shown: 12, tail_applied: 3 }),
1434 test: Some(Test {
1435 risk: 0.1,
1436 ran: 30,
1437 control: 9,
1438 refused: 4,
1439 sessions: 6,
1440 kept: 30,
1441 due_checks: 300,
1442 due_sessions: 30,
1443 due_min_sessions: 15,
1444 ..Test::default()
1445 }),
1446 ..Selection::default()
1447 }
1448 }
1449
1450 #[test]
1454 fn the_judges_effective_length_is_read_from_the_stores_cost_lines_and_shown() {
1455 let home = std::env::temp_dir().join(format!("borg-top-judge-length-{}", std::process::id()));
1456 let _ = std::fs::remove_dir_all(&home);
1457 std::fs::create_dir_all(home.join("state")).unwrap();
1458 let line = |chars: usize, n: u64, a: u64| {
1459 format!(
1460 "judge call=1 pending={n} resolved={a} chars={chars} returned= unmatched= verdicts=changed:{a},ignored:0,not_arisen:0,consequence_remains:0,unclear:{} reason_chars=1 model=x:big at=end\n",
1461 n - a
1462 )
1463 .repeat(10)
1464 };
1465 std::fs::write(home.join("state/s.cost"), line(30_000, 40, 20) + &line(70_000, 30, 15) + &line(100_000, 30, 15)).unwrap();
1466 let sel = selection(&home.join("lessons.jsonl"), select::policy());
1467 let big = sel.judge.iter().find(|l| l.model == "x:big").expect("the store's judge");
1468 assert_eq!((big.chars, big.start, big.explore), (108_000, false, Some(162_000)));
1469 assert_eq!(
1470 judge_length_line(big),
1471 "x:big: session-end budget 108,000 chars (measured, up from the start 72,000) · answered by prompt size: ≤32,000 50% of 400 [45–55%] reference, line 42%, ≤72,000 50% of 300 [44–56%] holds, ≤108,000 50% of 300 [44–56%] holds · exploring ≤162,000 on 20% of session-end passes"
1472 );
1473 let v = serde_json::to_value(&sel).unwrap();
1474 assert_eq!(v["judge"].as_array().map(|a| a.iter().any(|j| j["model"] == "x:big" && j["chars"] == 108_000)), Some(true), "{v}");
1475 let none = crate::judge_length::Length::of("y:new", &crate::judge_length::Steps::new());
1476 assert_eq!(
1477 judge_length_line(&none),
1478 "y:new: session-end budget 72,000 chars (the start: no decided step moved it) · no judge call measured yet · exploring ≤108,000 on 20% of session-end passes"
1479 );
1480 assert_eq!((grouped(0), grouped(999), grouped(1_000), grouped(1_234_567)), ("0".into(), "999".into(), "1,000".into(), "1,234,567".into()));
1481 let _ = std::fs::remove_dir_all(&home);
1482 }
1483
1484 #[test]
1487 fn the_misses_line_names_the_causes_of_low_first_applies() {
1488 let causes = |v: &[(&str, u64)]| v.iter().map(|(c, n)| (c.to_string(), *n)).collect();
1489 let m = crate::pools::Misses {
1490 at: 3 * 1_048_576 / 2,
1491 checks: 40,
1492 positives: 12,
1493 low: 5,
1494 drawn: 2,
1495 causes: causes(&[("scope", 0), ("no_word_match", 1), ("demoted", 3), ("other", 1)]),
1496 };
1497 assert_eq!(
1498 misses_line(&m, 0.15),
1499 "pool log 40 checks (1.5 MB), exploration 15% of the budget and of the dropped lessons, 5 of 12 first applies at grade ≥ 2 ranked low (2 shown by a draw): demoted 3, no_word_match 1, other 1"
1500 );
1501 let none = crate::pools::Misses { positives: 0, ..m };
1502 assert!(misses_line(&none, 0.0).ends_with("exploration off (BORG_BOUNDARY=off), no first apply at grade ≥ 2 yet"), "{}", misses_line(&none, 0.0));
1503 }
1504
1505 #[test]
1506 fn the_selection_line_says_what_decides_and_prices_spend_only_under_a_budget() {
1507 use select::Policy;
1508 use select::summary::Summary;
1509 let fresh = |policy, default| Selection { default, ..Selection::of(&Summary::default(), policy, None) };
1510 let fixed = fresh(Policy::Fixed, "fixed");
1512 assert_eq!(selection_line(&fixed), "the fixed rule (the default; BORG_SELECT=learned opts in to the learned policy) · no graded check folded yet");
1513 assert!(!fixed.shown(), "the default with nothing folded: nothing to show on the drones screen");
1514 let opted = fresh(Policy::Learned, "fixed");
1515 assert_eq!(
1516 selection_line(&opted),
1517 "learned (opt-in, BORG_SELECT=learned), cold start: gate 0, every decision the fixed rule's · no graded check folded yet"
1518 );
1519 assert!(opted.shown(), "a switch decides: shown");
1520 assert_eq!(selection_line(&fresh(Policy::Fixed, "learned")), "the fixed rule (BORG_SELECT=fixed) · no graded check folded yet");
1522 assert!(!fresh(Policy::Learned, "learned").shown() && fresh(Policy::Fixed, "learned").shown());
1523 let d = select::precedence(None, None, false);
1524 assert_eq!(Selection::of(&Summary::default(), Policy::Cap(2), None).default, d.name(), "the default is select.rs's, no switch set");
1525 assert!(!Selection::of(&Summary::default(), d, None).shown());
1526 let mut s = learned();
1527 let line = selection_line(&s);
1528 assert_eq!(
1529 line,
1530 "learned (opt-in, BORG_SELECT=learned), gate 0.120 (quality it may still risk below the fixed rule) · 480 checks · deviated: route 5/400 · question-count test 30 drawn (9 control, 4 refused by the gate) in 6 sessions, risk 0.10; readout after 300 checks in 15 sessions, or 30 sessions · placebos 40 drawn, 31 judged · tail 12 shown past the pool, 3 applied · monitor +0.004, bound +0.031 over 12 sessions (below 0 reverts), ESS 180 of 300 weighed, largest weight 6.5× the mean · $0.42 and 0.3 extra rounds per session (12 sessions)"
1531 );
1532 assert!(!line.contains("budget") && !line.contains("price"), "no budget set: spend measured, never priced: {line}");
1533 let mut early = learned();
1534 early.evaluation = Some(Evaluation { ucb: None, sessions: 3, units: 40, ess: 12.4, max_weight: 3.0, ..Evaluation::default() });
1535 let line = selection_line(&early);
1536 assert!(line.contains(" · monitor +0.000 over 3 sessions, no bound yet, ESS 12 of 40 weighed, largest weight 3.0× the mean · "), "{line}");
1537 early.latency = vec![
1538 Latency { harness: "codex".into(), checks: 20, p50_ms: 10_000, p95_ms: 19_000, budget_ms: Some(30_000), over: 1 },
1539 Latency { harness: "claude".into(), checks: 1, p50_ms: 4_000, p95_ms: 4_000, budget_ms: None, over: 0 },
1540 ];
1541 let line = selection_line(&early);
1542 assert!(line.contains(" · stop p95 codex 19.0 s of 30.0 s, 1 failed; claude 4.0 s · $"), "per harness, against its stop budget: {line}");
1543 s.budget = Some(Budget { tokens: 30_000.0, spent: 24_000.0, price: 0.0002 });
1544 assert!(selection_line(&s).ends_with(" · budget 30000 pool tokens per session: 24000 read, price 2.00e-4 per token"), "{}", selection_line(&s));
1545 s.source = select::REVERTED_SOURCE;
1546 assert!(selection_line(&s).starts_with("REVERTED to the fixed rule"), "{}", selection_line(&s));
1547 }
1548
1549 #[test]
1553 fn a_spent_gate_is_not_a_cold_start() {
1554 use select::Policy;
1555 use select::summary::{Gate, Summary};
1556 let line = |s: &Summary| selection_line(&Selection { default: "learned", ..Selection::of(s, Policy::Learned, None) });
1557 let mut s = Summary::default();
1558 s.fit.key = "3556601-bc6d6191".into();
1559 assert!(line(&s).starts_with("learned, cold start: gate 0, every decision the fixed rule's"), "{}", line(&s));
1561 s.gate = Gate { gain: vec![-0.1018; select::policy::DRAWS], allowance: 0.0760, ..Gate::default() };
1562 let spent = Selection::of(&s, Policy::Learned, None);
1563 assert_eq!((spent.gate, spent.allowance, spent.ledger), (0.0, 0.0760, -0.1018));
1564 assert!(line(&s).starts_with("learned, gate spent (ledger q05 -0.102 against allowance 0.076), deviations need positive evidence · "), "{}", line(&s));
1565 s.fit.key.clear();
1567 assert!(line(&s).starts_with("learned, cold start: gate 0"), "{}", line(&s));
1568 s.fit.key = "3556601-bc6d6191".into();
1570 s.gate.gain = vec![-0.05; select::policy::DRAWS];
1571 assert!(line(&s).starts_with("learned, gate 0.026 (quality it may still risk below the fixed rule)"), "{}", line(&s));
1572 let fixed = Selection::of(&s, Policy::Fixed, None);
1574 assert_eq!((fixed.allowance, fixed.ledger), (0.0, 0.0));
1575 }
1576
1577 #[test]
1580 fn a_store_says_how_its_lessons_are_selected() {
1581 use select::Policy;
1582 let home = std::env::temp_dir().join(format!("borg-top-select-{}", std::process::id()));
1583 let _ = std::fs::remove_dir_all(&home);
1584 std::fs::create_dir_all(&home).unwrap();
1585 let lessons = home.join("lessons.jsonl");
1586 std::fs::write(&lessons, "").unwrap();
1587 let line = |lessons: &Path, policy| selection_line(&selection(lessons, policy));
1588 assert!(line(&lessons, Policy::Learned).ends_with("cold start: gate 0, every decision the fixed rule's · no graded check folded yet"));
1589 assert!(line(&lessons, Policy::Fixed).starts_with("the fixed rule ("), "{}", line(&lessons, Policy::Fixed));
1590 assert!(line(&lessons, Policy::Cap(2)).contains("(BORG_STOP_ASK_CAP, evaluation only)"));
1591 let absent = home.join("absent").join("lessons.jsonl");
1592 assert!(line(&absent, Policy::Attention).contains("(BORG_STOP_ATTENTION=on, evaluation only)"), "no store: the prior");
1593 let _ = std::fs::remove_dir_all(&home);
1594 }
1595
1596 #[test]
1598 fn the_selection_line_follows_the_screen_when_shown() {
1599 let (ds, inbox) = (fixture(), inbox_fixture());
1600 let plain = render("corp", Share::Team, &ds, &inbox, NOW);
1601 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, None, NOW), plain);
1602 let cold = Selection::of(&select::summary::Summary::default(), select::precedence(None, None, false), None);
1603 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, Some(&cold), NOW), plain, "the default at cold start: no line");
1604 let s = learned();
1605 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, Some(&s), NOW), format!("{plain} selection: {}\n", selection_line(&s)));
1606 }
1607
1608 #[test]
1610 fn the_selection_serialises_every_section_for_the_page() {
1611 let v = serde_json::to_value(learned()).unwrap();
1612 let mut keys: Vec<&str> = v.as_object().unwrap().keys().map(String::as_str).collect();
1613 keys.sort_unstable();
1614 let want = "allowance at boundary budget checks declaration default delivery deviations effects evaluation fallbacks fit gate half_life judge labels latency ledger measurement misses policy pool sessions source spend test";
1615 assert_eq!(keys, want.split(' ').collect::<Vec<_>>());
1616 assert_eq!((&v["budget"], &v["spend"][0]["source"], &v["deviations"][0]["whole"]), (&Value::Null, &json!("stop"), &json!(400)));
1617 let mut keys: Vec<&str> = v["evaluation"].as_object().unwrap().keys().map(String::as_str).collect();
1618 keys.sort_unstable();
1619 assert_eq!(keys, ["ess", "max_weight", "mean", "sessions", "truncated", "ucb", "units"]);
1620 assert_eq!((&v["default"], &v["measurement"]), (&json!("fixed"), &json!({"placebos": 40, "judged": 31, "tail_shown": 12, "tail_applied": 3})));
1621 let mut keys: Vec<&str> = v["test"].as_object().unwrap().keys().map(String::as_str).collect();
1622 keys.sort_unstable();
1623 let want = "ask_free by_k control dilution due due_checks due_min_sessions due_sessions kept ran refused risk sessions set";
1624 assert_eq!(keys, want.split(' ').collect::<Vec<_>>());
1625 }
1626
1627 #[test]
1632 fn the_judged_share_leaves_out_what_its_session_already_had() {
1633 use crate::outcome::{self, Outcome};
1634 use select::summary::{Summary, fold};
1635 let rec = |kind: &str, check: &str, channel: &str, ms: i64| Outcome {
1636 kind: kind.into(),
1637 session: "s".into(),
1638 lessons: vec!["A".into()],
1639 check: check.into(),
1640 channel: channel.into(),
1641 ms,
1642 ..Default::default()
1643 };
1644 let graded = |check: &str, ms: i64| Outcome {
1645 grades: vec![2],
1646 ranks: vec![0],
1647 applied: 1,
1648 free: vec!["A".into()],
1649 pf: vec![1.0],
1650 policy: "fixed".into(),
1651 shown: 20,
1652 tokens: 6_000,
1653 ..rec(outcome::GRADED, check, "live", ms)
1654 };
1655 let log = [
1656 graded("c1", 10),
1657 Outcome { route: outcome::ROUTE_FREE.into(), ..rec(outcome::DELIVERY, "c1", "", 11) },
1658 Outcome { judge: 2, ..rec(outcome::CHANGED, "c1", "queued", 12) },
1659 graded("c2", 20),
1660 Outcome { route: outcome::ROUTE_DROPPED_DONE.into(), ..rec(outcome::DELIVERY, "c2", "", 21) },
1661 Outcome { lessons: vec!["B".into()], ..rec(outcome::ASKED, "", "early", 25) },
1663 Outcome { lessons: vec!["B".into()], grades: vec![3], chosen: vec!["B".into()], free: Vec::new(), p: vec![1.0], pf: vec![0.0], ..graded("c3", 30) },
1664 Outcome { lessons: vec!["B".into()], route: outcome::ROUTE_ASKED.into(), ..rec(outcome::DELIVERY, "c3", "", 31) },
1665 Outcome { lessons: vec!["B".into()], judge: 2, ..rec(outcome::CHANGED, "c3", "live", 32) },
1666 ];
1667 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1668 let s = fold(&text, Summary::default());
1669 assert_eq!(s.applies.iter().map(|a| a.had).collect::<Vec<_>>(), [false, true, true], "c2's A was handed over at c1, c3's B early");
1670 let sel = Selection::of(&s, select::Policy::Fixed, None);
1671 let row = |name: &str| sel.labels.iter().find(|c| c.name == name).map(|c| (c.part, c.whole));
1672 assert_eq!(row("judged, free"), Some((1, 1)), "c1's delivered and judged; c2's had, out: {:?}", sel.labels);
1673 assert_eq!(row("judged, asked"), Some((1, 1)), "c3's question of what B's session had, judged: {:?}", sel.labels);
1674 }
1675
1676 #[test]
1680 fn the_verdict_shares_read_judge_three_answers_per_route() {
1681 use crate::outcome::{self, Outcome};
1682 use select::summary::{Summary, fold};
1683 let rec = |kind: &str, l: &str, check: &str, channel: &str, ms: i64, judge: u8| Outcome {
1684 kind: kind.into(),
1685 session: "s".into(),
1686 lessons: vec![l.into()],
1687 check: check.into(),
1688 channel: channel.into(),
1689 ms,
1690 judge,
1691 ..Default::default()
1692 };
1693 let graded = |check: &str, ms: i64, chosen: &[&str], free: &[&str], placebo: &[&str]| {
1694 let lessons: Vec<String> = chosen.iter().chain(free).chain(placebo).map(|l| l.to_string()).collect();
1695 let n = lessons.len();
1696 Outcome {
1697 grades: vec![2; n],
1698 ranks: (0..n as u16).collect(),
1699 applied: n as u16,
1700 chosen: chosen.iter().map(|l| l.to_string()).collect(),
1701 free: free.iter().map(|l| l.to_string()).collect(),
1702 placebo: placebo.iter().map(|l| l.to_string()).collect(),
1703 p: lessons.iter().map(|l| f64::from(u8::from(chosen.contains(&l.as_str())))).collect(),
1704 pf: lessons.iter().map(|l| f64::from(u8::from(free.contains(&l.as_str())))).collect(),
1705 psi: if placebo.is_empty() { 0.0 } else { 0.5 },
1706 policy: "fixed".into(),
1707 lessons,
1708 ..rec(outcome::GRADED, "", check, "stop", ms, 0)
1709 }
1710 };
1711 let deliver = |l: &str, check: &str, route: &str, ms: i64| Outcome { route: route.into(), ..rec(outcome::DELIVERY, l, check, "", ms, 0) };
1712 let log = [
1713 graded("c0", 5, &[], &["O"], &[]),
1715 deliver("O", "c0", outcome::ROUTE_FREE, 6),
1716 rec(outcome::UNCLEAR, "O", "c0", "queued", 7, 2),
1717 graded("c1", 10, &["A", "B"], &["F", "G", "H"], &["P", "Q"]),
1718 deliver("A", "c1", outcome::ROUTE_ASKED, 11),
1719 deliver("B", "c1", outcome::ROUTE_ASKED, 11),
1720 deliver("F", "c1", outcome::ROUTE_FREE, 11),
1721 deliver("G", "c1", outcome::ROUTE_FREE, 11),
1722 deliver("H", "c1", outcome::ROUTE_FREE, 11),
1723 rec(outcome::CHANGED, "A", "c1", "stop", 20, 3),
1724 rec(outcome::UNCLEAR, "B", "c1", "stop", 20, 3),
1725 rec(outcome::UNCLEAR, "F", "c1", "queued", 20, 3),
1726 Outcome { at: outcome::AT_END.into(), ..rec(outcome::NOT_ARISEN, "F", "c1", "queued", 30, 3) },
1727 rec(outcome::REMAINS, "G", "c1", "queued", 20, 3),
1728 rec(outcome::UNCLEAR, "H", "c1", "queued", 20, 3),
1729 rec(outcome::UNCLEAR, "P", "c1", outcome::PLACEBO, 20, 3),
1730 rec(outcome::IGNORED, "Q", "c1", outcome::PLACEBO, 20, 3),
1731 ];
1732 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1733 let sel = Selection::of(&fold(&text, Summary::default()), select::Policy::Fixed, None);
1734 let row = |name: &str| sel.labels.iter().find(|c| c.name == name).map(|c| (c.part, c.whole));
1735 assert_eq!(row("unclear"), Some((3, 7)), "B, H and P of seven judge-3 answers, O's older one out: {:?}", sel.labels);
1736 assert_eq!(row("not arisen"), Some((1, 7)));
1737 assert_eq!(row("consequence remains"), Some((1, 7)));
1738 assert_eq!(row("unclear, asked"), Some((1, 2)));
1739 assert_eq!(row("unclear, free"), Some((1, 3)), "F's end verdict replaced its unclear: {:?}", sel.labels);
1740 assert_eq!(row("unclear, placebo"), Some((1, 2)));
1741 }
1742
1743 #[test]
1748 fn the_measurement_counts_the_placebos_the_fixed_rule_drew_and_judged() {
1749 use crate::outcome::{self, Outcome};
1750 use select::summary::{Summary, fold};
1751 let rec = |kind: &str, lessons: &[&str], check: &str, channel: &str, ms: i64| Outcome {
1752 kind: kind.into(),
1753 session: "s".into(),
1754 lessons: lessons.iter().map(|l| l.to_string()).collect(),
1755 check: check.into(),
1756 channel: channel.into(),
1757 ms,
1758 ..Default::default()
1759 };
1760 let graded = Outcome {
1761 grades: vec![1, 1, 2],
1762 ranks: vec![0, 1, 2],
1763 applied: 3,
1764 free: vec!["C".into()],
1765 placebo: vec!["A".into(), "B".into()],
1766 psi: 0.5,
1767 policy: "fixed".into(),
1768 shown: 20,
1769 tokens: 6_000,
1770 ..rec(outcome::GRADED, &["A", "B", "C"], "c1", "stop", 10)
1771 };
1772 let tailed = Outcome {
1774 grades: vec![1],
1775 ranks: vec![0],
1776 applied: 1,
1777 policy: "fixed".into(),
1778 shown: 20,
1779 tokens: 6_000,
1780 tail: vec!["T1".into(), "T2".into()],
1781 pt: 0.5,
1782 fill: vec![21, 25, 21],
1783 ..rec(outcome::GRADED, &["T2"], "c2", "stop", 14)
1784 };
1785 let log = [
1786 graded,
1787 tailed,
1788 Outcome { route: outcome::ROUTE_PLACEBO.into(), ..rec(outcome::DELIVERY, &["A"], "c1", "", 11) },
1789 Outcome { route: outcome::ROUTE_PLACEBO.into(), ..rec(outcome::DELIVERY, &["B"], "c1", "", 11) },
1790 Outcome { judge: 2, ..rec(outcome::IGNORED, &["A"], "c1", outcome::PLACEBO, 12) },
1791 Outcome { judge: 2, ..rec(outcome::CHANGED, &["C"], "c1", "queued", 13) },
1792 ];
1793 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1794 let sel = Selection::of(&fold(&text, Summary::default()), select::Policy::Fixed, None);
1795 assert_eq!(sel.measurement, Some(Measurement { placebos: 2, judged: 1, tail_shown: 2, tail_applied: 1 }), "{sel:?}");
1796 assert!(selection_line(&sel).contains(" · 2 checks · placebos 2 drawn, 1 judged · tail 2 shown past the pool, 1 applied"), "{}", selection_line(&sel));
1797 assert_eq!(Selection::of(&Summary::default(), select::Policy::Fixed, None).measurement, None, "nothing folded: not measured");
1798 }
1799
1800 #[test]
1804 fn the_selection_counts_the_learned_decisions_the_fixed_rule_took_over_by_reason() {
1805 use crate::outcome::{self, Outcome};
1806 use select::summary::{Summary, fold};
1807 let graded = |check: &str, policy: &str, reason: &str, ms: i64| Outcome {
1808 kind: outcome::GRADED.into(),
1809 session: "s".into(),
1810 lessons: vec!["A".into()],
1811 grades: vec![1],
1812 ranks: vec![0],
1813 applied: 1,
1814 free: vec!["A".into()],
1815 pf: vec![1.0],
1816 channel: "live".into(),
1817 check: check.into(),
1818 policy: policy.into(),
1819 reason: reason.into(),
1820 ms,
1821 ..Default::default()
1822 };
1823 let log = [
1824 graded("c1", "learned", "", 1),
1825 graded("c2", "fixed", outcome::DECIDED_BUSY, 2),
1826 graded("c3", "fixed", outcome::DECIDED_NO_RECORD, 3),
1827 graded("c4", "fixed", outcome::DECIDED_BUSY, 4),
1828 ];
1829 let lines = |log: &[Outcome]| log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect::<String>();
1830 let sel = Selection::of(&fold(&lines(&log), Summary::default()), select::Policy::Learned, None);
1831 let rows: Vec<(&str, u64, u64)> = sel.fallbacks.iter().map(|c| (c.name.as_str(), c.part, c.whole)).collect();
1832 assert_eq!(rows, [("lock_busy", 2, 4), ("lock_failed", 0, 4), ("unrecorded", 1, 4)]);
1833 let line = selection_line(&sel);
1834 assert!(line.contains(" · 4 checks · learned decisions the fixed rule took over: lock_busy 2, unrecorded 1 · "), "{line}");
1835 let none = Selection::of(&fold(&lines(&log[..1]), Summary::default()), select::Policy::Learned, None);
1836 assert!(none.fallbacks.iter().all(|c| c.part == 0) && none.fallbacks.len() == 3, "{none:?}");
1837 assert!(!selection_line(&none).contains("took over"), "{}", selection_line(&none));
1838 assert!(Selection::of(&Summary::default(), select::Policy::Learned, None).fallbacks.is_empty(), "nothing folded: not measured");
1839 }
1840
1841 #[test]
1846 fn the_selection_fills_latency_evaluation_and_delivery_from_the_summary() {
1847 use select::summary::{Summary, fold};
1848 let call = |harness: &str, channel: &str, dur_ms: u64, failed: bool| {
1849 let o = crate::outcome::Outcome {
1850 kind: "graded".into(),
1851 session: "s".into(),
1852 channel: channel.into(),
1853 harness: harness.into(),
1854 dur_ms,
1855 failed,
1856 tokens: 6_000,
1857 ..Default::default()
1858 };
1859 format!("{}\n", serde_json::to_string(&o).unwrap())
1860 };
1861 let mut log: String = (1..=20).map(|k| call("codex", "stop", k * 1_000, k == 20)).collect();
1862 log += &call("claude", "stop", 4_000, false);
1863 log += &call("codex", "live", 99_000, false);
1864 log += &call("nonesuch", "stop", 5_000, false);
1865 let mut s = fold(&log, Summary::default());
1866 s.monitor = select::policy::Monitor {
1867 sessions: 12,
1868 units: 400,
1869 mean: 0.004,
1870 ucb: Some(0.031),
1871 labelled: 300,
1872 ess: 180.0,
1873 max_weight: 6.5,
1874 truncated: 3,
1875 ..Default::default()
1876 };
1877 s.fit.key = "fit".into();
1878 s.fit.rates = select::policy::Rates { free_stop: 0.6, free_live: 0.8, later: 0.4, wait: 0.25, ask_live: 1.0 };
1879 let sel = Selection::of(&s, select::Policy::Learned, None);
1880 let budget = |h: &str| crate::harness::ALL.iter().find(|x| x.name() == h).and_then(|x| x.caps().stop_budget).map(|b| b.as_millis() as u64);
1881 assert!(budget("codex").is_some() && budget("claude").is_none(), "codex names a stop budget, claude none");
1882 let want = vec![
1883 Latency { harness: "claude".into(), checks: 1, p50_ms: 4_000, p95_ms: 4_000, budget_ms: None, over: 0 },
1884 Latency { harness: "codex".into(), checks: 20, p50_ms: 10_000, p95_ms: 19_000, budget_ms: budget("codex"), over: 1 },
1885 ];
1886 assert_eq!(sel.latency, want);
1887 let e = sel.evaluation.as_ref().expect("the monitor weighed labelled applies");
1888 assert_eq!(*e, Evaluation { sessions: 12, mean: 0.004, ucb: Some(0.031), units: 300, ess: 180.0, max_weight: 6.5, truncated: 3 });
1889 assert_eq!(sel.delivery.as_ref().map(|r| r.wait), Some(0.25));
1890 let keep = select::summary::LATENCY_KEEP as u64;
1892 let many: String = (1..=keep + 1).map(|k| call("claude", "stop", k, false)).collect();
1893 let kept = Selection::of(&fold(&many, Summary::default()), select::Policy::Learned, None);
1894 assert_eq!((kept.latency[0].checks, kept.latency[0].p50_ms), (keep, 1 + keep / 2), "the oldest call dropped");
1895 let cold = Selection::of(&Summary::default(), select::Policy::Learned, None);
1896 assert!(cold.latency.is_empty() && cold.evaluation.is_none() && cold.delivery.is_none(), "{cold:?}");
1897 assert_eq!((quantile(&[], 0.95), quantile(&[7], 0.5)), (0, 7));
1898 }
1899
1900 fn tested(check: &str, session: &str, ms: i64, source: &str, xk: Option<u32>, xs: Option<u32>) -> crate::outcome::Outcome {
1903 crate::outcome::Outcome {
1904 kind: crate::outcome::GRADED.into(),
1905 session: session.into(),
1906 lessons: vec!["A".into(), "B".into()],
1907 grades: vec![2, 1],
1908 ranks: vec![0, 1],
1909 applied: 2,
1910 check: check.into(),
1911 channel: "live".into(),
1912 policy: "learned".into(),
1913 source: source.into(),
1914 xk,
1915 xs,
1916 ms,
1917 ..Default::default()
1918 }
1919 }
1920
1921 #[test]
1927 fn the_test_section_is_the_summarys_fold_of_its_checks() {
1928 use select::policy::{ALPHA, EVIDENCE, EXPERIMENT, EXPERIMENT_CONTROL, GATE};
1929 use select::summary::{Summary, fold};
1930 let log = [
1931 tested("c1", "s1", 1, EXPERIMENT, Some(2), Some(1)),
1932 tested("c2", "s1", 2, EXPERIMENT_CONTROL, Some(2), Some(0)),
1933 tested("c3", "s2", 3, EXPERIMENT, Some(2), Some(1)),
1934 tested("c4", "s2", 4, GATE, Some(2), Some(1)),
1935 tested("c5", "s2", 5, GATE, Some(1), Some(0)),
1936 tested("c6", "s3", 6, EVIDENCE, None, None),
1937 tested("c7", "s3", 7, GATE, None, None),
1938 crate::outcome::Outcome { policy: "fixed".into(), ..tested("c8", "s3", 8, EXPERIMENT, Some(2), Some(2)) },
1939 ];
1940 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1941 let s = fold(&text, Summary::default());
1942 let t = Selection::of(&s, select::Policy::Learned, Some(0.2)).test.expect("the learned policy shows its test");
1943 let draws = |k: u32, draws: Vec<u64>, refused: u64| Draws { k, draws, refused };
1944 assert_eq!((t.ran, t.control, t.refused, t.sessions), (5, 1, 2, 2), "{t:?}");
1945 assert_eq!(t.by_k, vec![draws(1, vec![1, 0], 1), draws(2, vec![1, 3, 0], 1)]);
1946 assert_eq!((t.risk, t.set), (0.2, true), "the collective's risk");
1947 assert!(!t.due && t.dilution.is_empty() && t.ask_free.is_none(), "5 checks, 2 sessions: not due, nothing estimated");
1948 assert_eq!((t.due_checks, t.due_sessions), (select::replay::TEST_CHECKS, u64::from(select::replay::TEST_SESSIONS)));
1949 let line = selection_line(&Selection::of(&s, select::Policy::Learned, Some(0.2)));
1950 assert!(
1951 line.contains(" · question-count test 5 drawn (1 control, 2 refused by the gate) in 2 sessions, risk 0.20 (the collective's); readout after 300 checks in 15 sessions, or 30 sessions · "),
1952 "{line}"
1953 );
1954 assert_eq!(t.kept, t.ran, "every drawn check still held: {t:?}");
1955 let mut compacted = s.clone();
1957 compacted.checks.remove(0);
1958 let c = Selection::of(&compacted, select::Policy::Learned, Some(0.2)).test.unwrap();
1959 assert_eq!((c.ran, c.kept), (5, 4), "{c:?}");
1960 assert!(
1961 test_line(&c).starts_with("question-count test 5 drawn (1 control, 2 refused by the gate, 4 still in the summary) in 2 sessions"),
1962 "{}",
1963 test_line(&c)
1964 );
1965 let default = Selection::of(&s, select::Policy::Learned, None).test.unwrap();
1966 assert_eq!((default.risk, default.set), (ALPHA, false), "unset: the policy's ALPHA");
1967 let cold = Selection::of(&Summary::default(), select::Policy::Learned, Some(0.0));
1969 assert_eq!(cold.test.as_ref().map(|t| (t.ran, t.risk)), Some((0, 0.0)));
1970 assert_eq!(test_line(cold.test.as_ref().unwrap()), "question-count test off (risk 0)");
1971 let none = Selection::of(&Summary::default(), select::Policy::Learned, None).test.unwrap();
1972 assert_eq!(test_line(&none), "question-count test: no check yet, risk 0.10");
1973 assert!(Selection::of(&Summary::default(), select::Policy::Fixed, None).test.is_none());
1975 assert_eq!(Selection::of(&s, select::Policy::Fixed, None).test.map(|t| t.ran), Some(5), "an earlier opt-in's checks");
1976 }
1977
1978 #[test]
1982 fn the_gate_shown_is_the_one_in_force_at_the_collectives_risk() {
1983 use select::summary::{Gate, Summary};
1984 let s = Summary { gate: Gate { gain: vec![0.0; select::policy::DRAWS], allowance: 0.3, ..Gate::default() }, ..Summary::default() };
1985 let gate = |risk: Option<f64>| Selection::of(&s, select::Policy::Learned, risk).gate;
1986 assert!((gate(None) - 0.3).abs() < 1e-12, "{}", gate(None));
1987 assert!((gate(Some(0.2)) - 0.6).abs() < 1e-12, "twice ALPHA: twice the allowance, {}", gate(Some(0.2)));
1988 assert_eq!(gate(Some(0.0)), 0.0, "risk 0");
1989 }
1990
1991 #[test]
1994 fn the_test_readout_shows_the_replays_estimates_once_due() {
1995 use crate::outcome::{self, Outcome};
1996 use select::policy::EXPERIMENT;
1997 use select::summary::{Summary, fold};
1998 let sessions = u64::from(select::replay::TEST_SESSIONS);
1999 let mut text = String::new();
2000 let mut line = |o: &Outcome| text.push_str(&format!("{}\n", serde_json::to_string(o).unwrap()));
2001 for i in 0..sessions {
2002 let (session, check, ms) = (format!("s{i}"), format!("{i:012x}"), 10 * i as i64);
2003 let asked = i % 2 == 0;
2005 let g = Outcome {
2006 lessons: vec!["A".into()],
2007 grades: vec![2],
2008 ranks: vec![0],
2009 applied: 1,
2010 chosen: if asked { vec!["A".into()] } else { Vec::new() },
2011 free: if asked { Vec::new() } else { vec!["A".into()] },
2012 p: vec![0.5],
2013 pf: vec![0.5],
2014 ..tested(&check, &session, ms, EXPERIMENT, Some(1), Some(u32::from(asked)))
2015 };
2016 line(&g);
2017 let rec = |kind: &str, ms: i64| Outcome {
2018 kind: kind.into(),
2019 session: session.clone(),
2020 lessons: vec!["A".into()],
2021 check: check.clone(),
2022 ms,
2023 ..Default::default()
2024 };
2025 let route = if asked { outcome::ROUTE_ASKED } else { outcome::ROUTE_FREE };
2026 line(&Outcome { route: route.into(), ..rec(outcome::DELIVERY, ms + 1) });
2027 let verdict = if asked { outcome::CHANGED } else { outcome::IGNORED };
2029 line(&Outcome { judge: 2, channel: if asked { "live" } else { "queued" }.into(), ..rec(verdict, ms + 2) });
2030 }
2031 let s = fold(&text, Summary::default());
2032 let t = Selection::of(&s, select::Policy::Learned, None).test.unwrap();
2033 assert!(t.due && t.sessions == sessions, "{t:?}");
2034 let r = select::replay::test_readout(&s);
2035 let want = r.ask_free.as_ref().and_then(|x| x.value).expect("judged 2s on both routes");
2036 let e = t.ask_free.as_ref().expect("estimated once due");
2037 assert_eq!((e.mean, e.unit.as_str()), (100.0 * want, "points changed"));
2038 assert!(e.mean > 0.0, "asking changed the work, handing over did not: {e:?}");
2039 assert!(t.dilution.is_empty() && r.by_asked.is_empty(), "{t:?}");
2041 let line = selection_line(&Selection::of(&s, select::Policy::Learned, None));
2042 assert!(line.contains(&format!("; readout: grade 2 asked − handed over {:+.1} [", e.mean)), "{line}");
2043 }
2044
2045 #[test]
2048 fn unread_counts_follow_the_shown_once_claims() {
2049 let home = std::env::temp_dir().join(format!("borg-top-inbox-{}", std::process::id()));
2050 let _ = std::fs::remove_dir_all(&home);
2051 std::fs::create_dir_all(&home).unwrap();
2052 let ds = fixture();
2053 assert!(inbox(&home, &ds, NOW).recent.is_empty());
2054 let here = msg::node(&home);
2055 msg::send(&home, message_on(&here, "cccc3333-x", "Three of Three", "aaaa1111-x", msg::TELL, NOW - 3_000, "look at recon"), NOW).unwrap();
2056 msg::send(&home, message_on(&here, "cccc3333-x", "Three of Three", msg::ALL, msg::TELL, NOW - 2_000, "lunch"), NOW).unwrap();
2057 let i = inbox(&home, &ds, NOW);
2058 assert_eq!(i.unread.get("aaaa1111-x"), Some(&2), "{i:?}");
2059 assert_eq!(i.unread.get("cccc3333-x"), Some(&0), "its own broadcast");
2060 assert!(!i.unread.contains_key(&ds[1].p.session), "remote: claimed on its node");
2061 assert_eq!((i.day, i.recent[0].text.as_str()), (2, "lunch"));
2062 let caps = crate::harness::by_name("claude").expect("a harness").caps();
2063 let ev = json!({"session_id": "aaaa1111-x"});
2064 assert!(msg::deliver(&home, caps, "prompt", &ev, Some(100), NOW, None).is_some());
2065 assert_eq!(inbox(&home, &ds, NOW).unread.get("aaaa1111-x"), Some(&0));
2066 let _ = std::fs::remove_dir_all(&home);
2067 }
2068
2069 fn seen_lines(home: &Path) -> Vec<(String, Vec<String>)> {
2071 msg::lines(home)
2072 .into_iter()
2073 .filter_map(|l| match l {
2074 msg::Line::Seen(s) => Some((s.by, s.ids)),
2075 msg::Line::Message(_) => None,
2076 })
2077 .collect()
2078 }
2079
2080 fn shell(sessions: Vec<(String, &'static str)>, pids: Vec<(u32, &'static str)>) -> impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String> {
2083 move |ds| msg::pick(sessions.clone(), &pids, ds)
2084 }
2085
2086 fn acked_fixture(base: &Path) -> (std::path::PathBuf, String, Message) {
2089 let _ = std::fs::remove_dir_all(base);
2090 let (home, work) = (base.join("store"), base.join("work"));
2091 std::fs::create_dir_all(&home).unwrap();
2092 std::fs::create_dir_all(&work).unwrap();
2093 let git = |a: &[&str]| std::process::Command::new("git").arg("-C").arg(&work).args(a).output().is_ok_and(|o| o.status.success());
2094 assert!(git(&["init", "-q"]) && git(&["config", "user.email", "ann@x"]));
2095 let here = msg::node(&home);
2096 let to_me = message_on(&here, "aaaa1111-x", "One of Three", msg::ENGINEER, msg::TELL, NOW - 3_000, "the build is green");
2097 let to_drone = message_on(&here, "cccc3333-x", "Three of Three", "aaaa1111-x", msg::TELL, NOW - 2_000, "look at recon");
2098 msg::send(&home, to_me.clone(), NOW).unwrap();
2099 msg::send(&home, to_drone, NOW).unwrap();
2100 (home, work.display().to_string(), to_me)
2101 }
2102
2103 #[test]
2107 fn the_screen_acknowledges_the_messages_to_the_engineer_it_prints() {
2108 let base = std::env::temp_dir().join(format!("borg-top-ack-{}", std::process::id()));
2109 let (home, cwd, to_me) = acked_fixture(&base);
2110 let routes = [Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local }];
2111 let terminal = || shell(Vec::new(), Vec::new());
2112 let json = screen_of(&routes, true, &cwd, terminal(), true, NOW);
2113 assert!(json.contains("the build is green") && seen_lines(&home).is_empty(), "--json acknowledges nothing: {json}");
2114 let text = screen_of(&routes, true, &cwd, terminal(), false, NOW);
2115 assert!(text.contains("the build is green") && text.contains("look at recon"), "{text}");
2116 assert_eq!(seen_lines(&home), [("engineer:ann@x".to_string(), vec![to_me.id.clone()])], "the engineer's message only, as theirs");
2117 screen_of(&routes, true, &cwd, terminal(), false, NOW + 1_000);
2118 assert_eq!(seen_lines(&home).len(), 1, "acknowledged once: a later look writes no second line");
2119 let _ = std::fs::remove_dir_all(&base);
2120 }
2121
2122 #[test]
2127 fn a_session_running_the_screen_acknowledges_nothing_as_the_engineer() {
2128 let base = std::env::temp_dir().join(format!("borg-top-ack-session-{}", std::process::id()));
2129 let (home, cwd, to_me) = acked_fixture(&base);
2130 let other = base.join("other");
2131 std::fs::create_dir_all(&other).unwrap();
2132 let pid = std::process::id();
2133 let oc = Presence { harness: "opencode".into(), pid: Some(pid), ..p("dddd4444-x", "ann@x", NOW - 9_000, NOW - 1_000, "prompt", "docs", &[]) };
2134 presence::write(&other, &oc).unwrap();
2135 let routes = [
2136 Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local },
2137 Route::Collective { name: "side".into(), home: other.clone(), share: Share::Local },
2138 ];
2139 let shells = [
2140 ("Claude Code", shell(vec![("eeee5555-x".into(), "claude")], Vec::new())),
2141 ("Codex in Claude Code, neither with presence", shell(vec![("eeee5555-x".into(), "claude"), ("ffff6666-x".into(), "codex")], Vec::new())),
2142 ("opencode, its drone in the other collective", shell(Vec::new(), vec![(pid, "opencode")])),
2143 ];
2144 for (what, caller) in shells {
2145 let text = screen_of(&routes, false, &cwd, caller, false, NOW);
2146 assert!(text.contains("the build is green"), "{what}: shown {text}");
2147 assert!(seen_lines(&home).is_empty() && seen_lines(&other).is_empty(), "{what}: acknowledged nothing as the engineer");
2148 }
2149 screen_of(&routes, false, &cwd, shell(Vec::new(), vec![(pid + 1, "opencode")]), false, NOW);
2150 assert_eq!(seen_lines(&home), [("engineer:ann@x".to_string(), vec![to_me.id.clone()])], "a terminal (no drone of its pid): the engineer's");
2151 let _ = std::fs::remove_dir_all(&base);
2152 }
2153
2154 #[test]
2159 fn the_printed_screen_acknowledges_only_the_messages_whose_text_it_prints() {
2160 let base = std::env::temp_dir().join(format!("borg-top-ack-cut-{}", std::process::id()));
2161 let (home, cwd, to_me) = acked_fixture(&base);
2162 let email = format!("{}@x", "a".repeat(58));
2164 assert!(std::process::Command::new("git").args(["-C", &cwd, "config", "user.email", &email]).output().is_ok_and(|o| o.status.success()));
2165 let (here, to) = (msg::node(&home), format!("{}:{email}", msg::ENGINEER));
2166 let one = message_on(&here, "bbbb2222-x", &"n".repeat(52), &to, msg::TELL, NOW - 1_500, "one character shows");
2167 let none = message_on(&here, "dddd4444-x", &"n".repeat(53), &to, msg::TELL, NOW - 1_000, "none shows");
2168 msg::send(&home, one.clone(), NOW).unwrap();
2169 msg::send(&home, none.clone(), NOW).unwrap();
2170 let routes = [Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local }];
2171 let text = screen_of(&routes, true, &cwd, shell(Vec::new(), Vec::new()), false, NOW);
2172 assert!(text.contains(&format!("{} -> {email}: o…", one.name)) && text.contains(&format!("{} -> {email}: …", none.name)), "{text}");
2173 let mut acked: Vec<String> = seen_lines(&home).into_iter().flat_map(|(by, ids)| ids.into_iter().map(move |id| format!("{by} {id}"))).collect();
2174 acked.sort();
2175 let mut want = [format!("{to} {}", to_me.id), format!("{to} {}", one.id)];
2176 want.sort();
2177 assert_eq!(acked, want, "the line that shows one character of its text is read; the one cut before it is not");
2178 let _ = std::fs::remove_dir_all(&base);
2179 }
2180
2181 #[test]
2185 fn a_live_screen_acknowledges_each_message_once() {
2186 let base = std::env::temp_dir().join(format!("borg-top-ack-once-{}", std::process::id()));
2187 let _ = std::fs::remove_dir_all(&base);
2188 std::fs::create_dir_all(&base).unwrap();
2189 let home = base.clone();
2190 let here = msg::node(&home);
2191 let to_me = |by: &str, ms: i64, text: &str| message_on(&here, by, "a drone", msg::ENGINEER, msg::TELL, ms, text);
2192 let first = to_me("aaaa1111-x", NOW - 3_000, "the build is green");
2193 msg::send(&home, first.clone(), NOW).unwrap();
2194 let cwd = base.display().to_string();
2195 let mut done = HashSet::new();
2196 let recent = |now: i64| inbox(&home, &[], now).recent;
2197 acknowledge(&home, &cwd, &recent(NOW).iter().collect::<Vec<_>>(), &mut done, NOW);
2198 assert_eq!(seen_lines(&home).len(), 1, "{:?}", seen_lines(&home));
2199 assert!(done.contains(&first.id));
2200 let kept: String = std::fs::read_to_string(msg::path(&home)).unwrap().lines().filter(|l| !l.contains("\"seen\"")).map(|l| format!("{l}\n")).collect();
2203 std::fs::write(msg::path(&home), kept).unwrap();
2204 acknowledge(&home, &cwd, &recent(NOW + 1_000).iter().collect::<Vec<_>>(), &mut done, NOW + 1_000);
2205 assert!(seen_lines(&home).is_empty(), "skipped: no second ack of a done id");
2206 let second = to_me("cccc3333-x", NOW + 2_000, "and the docs");
2208 msg::send(&home, second.clone(), NOW + 2_000).unwrap();
2209 acknowledge(&home, &cwd, &recent(NOW + 3_000).iter().collect::<Vec<_>>(), &mut done, NOW + 3_000);
2210 assert_eq!(seen_lines(&home).into_iter().map(|(_, ids)| ids).collect::<Vec<_>>(), [vec![second.id.clone()]], "only the new one");
2211 let _ = std::fs::remove_dir_all(&base);
2212 }
2213
2214 #[test]
2217 fn a_sender_is_tied_to_the_node_its_line_names() {
2218 let ds = fixture();
2219 let own = message("aaaa1111-x", "anything", msg::ALL, msg::TELL, NOW, "hi");
2220 assert_eq!(sender(&ds, "n1", &own), "One of Three", "this node's drone");
2221 let forged = message_on("evil0123456789", "aaaa1111-x", "One of Three", msg::ALL, msg::TELL, NOW, "hi");
2222 assert_eq!(sender(&ds, "n1", &forged), "One of Three @evil01234567", "the claim, with the node it came from");
2223 let remote = message_on(&"beefcafe".repeat(8), &ds[1].p.session, "x", msg::ALL, msg::TELL, NOW, "hi");
2224 assert_eq!(sender(&ds, "n1", &remote), "Two of Three @beefcafebeef", "the remote drone on its own node");
2225 assert_eq!(recipient(&ds, "n1", "aaaa1111-x@n1"), "One of Three");
2226 assert_eq!(recipient(&ds, "n1", "aaaa1111-x@evil"), "aaaa1111", "not the local drone of that id");
2227 assert_eq!((recipient(&ds, "n1", "engineer"), recipient(&ds, "n1", msg::ALL)), ("engineer".into(), "collective".into()), "an `all` line");
2228 assert_eq!(recipient(&ds, "n1", "local@n1"), "local");
2229 assert_eq!(recipient(&ds, "n1", "local@beefcafebeefcafe00"), "local @beefcafebeef");
2230 }
2231
2232 #[test]
2236 fn the_line_on_borg_itself_names_a_release_and_the_protocols_that_differ() {
2237 let home = std::env::temp_dir().join(format!("borg-top-version-{}", std::process::id()));
2238 let _ = std::fs::remove_dir_all(&home);
2239 std::fs::create_dir_all(&home).unwrap();
2240 let newer = update::Checked { checked_ms: 1, latest: Some("99.0.0".into()), error: None };
2241 let current = update::Checked { latest: Some(update::VERSION.into()), ..newer.clone() };
2242 assert_eq!(version_line([home.as_path()], None, true), None);
2243 assert_eq!(version_line([home.as_path()], Some(¤t), true), None);
2244 let release = format!("borg 99.0.0 is released, this is {}: borg update", update::VERSION);
2245 assert_eq!(version_line([home.as_path()], Some(&newer), true), Some(release.clone()));
2246 assert_eq!(version_line([home.as_path()], Some(&newer), false), None, "BORG_UPDATE_CHECK=off");
2247
2248 let ours = protocol::current();
2249 let v = |protocol, engineer: Option<&str>| protocol::PeerVersion { borg: None, protocol, ms: 1, engineer: engineer.map(str::to_string) };
2250 protocol::note_version(&home, "k-same", &v(ours, Some("ann@x"))).unwrap();
2251 assert_eq!(version_line([home.as_path()], None, true), None, "a member on this node's protocol is no news");
2252 protocol::note_version(&home, "k-old", &v(ours - 1, Some("bob@x"))).unwrap();
2253 let mismatch = protocol::Mismatch { theirs: format!("borg/mesh/{}", ours + 1), ours: format!("borg/mesh/{ours}"), ms: 2 };
2254 protocol::note_mismatch(&home, "k-new", mismatch).unwrap();
2255 let line = version_line([home.as_path()], Some(&newer), true).unwrap();
2256 let old = format!("bob@x on /{} misses /{ours}: {}", ours - 1, protocol::note(ours).unwrap());
2257 let none = format!("nothing syncs with peer k-new (borg/mesh/{}, this node borg/mesh/{ours})", ours + 1);
2258 assert_eq!(line, format!("{release} · {none} · {old}"));
2259 let _ = std::fs::remove_dir_all(&home);
2260 }
2261
2262 #[test]
2263 fn json_for_scripts() {
2264 let v = to_json_with("corp", Share::Team, &fixture(), &inbox_fixture());
2265 assert_eq!((v["drones"][0]["unread"].as_u64(), v["drones"][1].get("unread")), (Some(2), None));
2266 assert_eq!((v["messages"]["day"].as_u64(), v["messages"]["recent"][0]["kind"].as_str()), (Some(3), Some("ask")));
2267 assert_eq!(v["collective"], "corp");
2268 let d = &v["drones"];
2269 assert_eq!(d.as_array().unwrap().len(), 3);
2270 assert_eq!(d[0]["designation"], "One of Three");
2271 assert_eq!(d[0]["session"], "aaaa1111-x");
2272 assert_eq!(d[0]["state"], "working");
2273 assert_eq!(d[0]["paused"], true);
2274 assert_eq!(d[1]["node"], "beefcafe".repeat(8));
2275 assert_eq!(d[1]["state"], "waiting");
2276 assert_eq!(d[2]["state"], "idle");
2277 assert_eq!(d[2]["files"], json!([]));
2278 }
2279
2280 #[test]
2282 fn a_name_matching_here_and_on_a_member_names_the_drone_here() {
2283 let local = vec![p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "post_tool", "a", &[])];
2284 let remote = ["aaaa1111-x", "aaaa2222-z"].map(|s| presence::Remote {
2285 p: p(s, "bob@y", 50, NOW - 5_000, "post_tool", "b", &[]),
2286 key: "beefcafe".repeat(8),
2287 sig: String::new(),
2288 });
2289 let ds = assemble(local, remote.into(), NOW, |_| true, |_| false);
2290 for t in ["aaaa", "aaaa1111", "aaaa1111-x"] {
2291 let d = resolve(&ds, t).unwrap().unwrap();
2292 assert!(d.node.is_empty() && d.p.session == "aaaa1111-x", "{t}");
2293 }
2294 assert_eq!(resolve(&ds, "aaaa2").unwrap().unwrap().node, "beefcafe".repeat(8), "only a member's matches");
2295 assert!(!resolve(&ds, "One").unwrap().unwrap().node.is_empty(), "a number names one drone of all, wherever it runs");
2296 }
2297
2298 #[test]
2300 fn a_name_matching_an_ended_drone_here_and_a_live_member_names_the_live_one() {
2301 let local = vec![p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "end", "a", &[])];
2302 let remote =
2303 vec![presence::Remote { p: p("aaaa2222-z", "bob@y", 50, NOW - 5_000, "post_tool", "b", &[]), key: "beefcafe".repeat(8), sig: String::new() }];
2304 let ds = assemble(local, remote, NOW, |_| true, |_| false);
2305 let d = resolve(&ds, "aaaa").unwrap().unwrap();
2306 assert_eq!((d.p.session.as_str(), d.node.as_str()), ("aaaa2222-z", "beefcafe".repeat(8).as_str()), "the live one, though remote");
2307 assert!(resolve(&ds, "aaaa1111").unwrap().unwrap().node.is_empty(), "an ended drone is still named by its own id");
2308 }
2309
2310 #[test]
2311 fn resolves_drones_by_designation_or_session() {
2312 let ds = fixture();
2313 assert_eq!(resolve(&ds, "Two").unwrap().unwrap().p.session, "bbbb2222-y");
2314 assert_eq!(resolve(&ds, "two of three").unwrap().unwrap().p.session, "bbbb2222-y");
2315 assert_eq!(resolve(&ds, "3").unwrap().unwrap().p.session, "cccc3333-x");
2316 assert_eq!(resolve(&ds, "aaaa").unwrap().unwrap().p.session, "aaaa1111-x");
2317 assert!(resolve(&ds, "all").unwrap().is_none());
2318 assert!(resolve(&ds, "Four").is_err());
2319 assert!(resolve(&ds, "zzzz").is_err());
2320 assert!(resolve(&ds, "aa").is_err(), "too short a prefix");
2321 assert!(["collective", "local"].iter().all(|w| resolve(&ds, w).unwrap().is_none()));
2322 assert_eq!(resolve(&ds, "bbbb").unwrap().unwrap().p.session, "bbbb2222-y", "nothing here matches: the member's");
2323 assert_eq!(word(21), "21");
2324 }
2325}