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 pub placebos_needed: u64,
664}
665
666#[derive(Debug, Clone, Default, PartialEq, Serialize)]
669pub struct Yield {
670 pub from: u32,
671 pub to: u32,
672 pub shown: u64,
673 pub applied: u64,
674 pub rate: f64,
675}
676
677#[derive(Debug, Clone, Default, PartialEq, Serialize)]
680pub struct Latency {
681 pub harness: String,
682 pub checks: u64,
683 pub p50_ms: u64,
685 pub p95_ms: u64,
686 pub budget_ms: Option<u64>,
688 pub over: u64,
691}
692
693const QUANTILES: [f64; 2] = [0.50, 0.95];
696
697fn quantile(sorted: &[u64], q: f64) -> u64 {
699 let rank = ((q * sorted.len() as f64).ceil() as usize).clamp(1, sorted.len().max(1));
700 sorted.get(rank - 1).copied().unwrap_or(0)
701}
702
703impl Selection {
704 #[must_use]
708 pub fn of(summary: &select::summary::Summary, policy: select::Policy, risk: Option<f64>) -> Selection {
709 use select::summary::{LIVE, Label, STOP};
710 let s = summary;
711 let learned = policy == select::Policy::Learned;
712 let reverted = learned && s.monitor.reverted;
713 let default = select::precedence(None, None, false).name();
714 let count = |name: &str, part: usize, whole: usize| Count { name: name.into(), part: part as u64, whole: whole as u64 };
715 let (sessions, checks) = s.checked();
717 let deviations = match s.gate.checks {
718 0 => Vec::new(),
719 n => vec![count("route", s.gate.deviated as usize, n as usize)],
720 };
721 let fallbacks = select::summary::FELL_BACK
723 .iter()
724 .zip(s.fell_back)
725 .filter(|_| checks > 0)
726 .map(|(reason, n)| Count { name: (*reason).into(), part: n, whole: checks })
727 .collect();
728 let spend = [("stop", STOP), ("live", LIVE)]
729 .into_iter()
730 .filter(|(_, i)| s.spend[*i].calls > 0)
731 .map(|(source, i)| Spend { source: source.into(), calls: s.spend[i].calls, rounds: s.spend[i].rounds, usd: s.spend[i].usd })
732 .collect();
733 let budget = s.fit.budget.map(|tokens| Budget { tokens, spent: s.pool.tokens as f64 / sessions.max(1) as f64, price: s.fit.lambda });
734 let judged = |l: Label| l.verdict();
736 let of = |route: select::Route| {
742 let of: Vec<_> = s.applies.iter().filter(|a| a.route == route && (route != select::Route::Free || !a.had)).collect();
743 (of.iter().filter(|a| judged(a.label)).count(), of.len())
744 };
745 let mut labels = Vec::new();
746 for (name, route) in [("judged, asked", select::Route::Ask), ("judged, free", select::Route::Free), ("judged, placebo", select::Route::Placebo)] {
747 let (part, whole) = of(route);
748 labels.push(count(name, part, whole));
749 }
750 let answers: Vec<_> = s.applies.iter().filter(|a| a.jv >= crate::predict::PURE_UNCLEAR_FROM && (judged(a.label) || a.label.pending())).collect();
755 let share = |name: &str, label: Label, route: Option<select::Route>| {
756 let of: Vec<_> = answers.iter().filter(|a| route.is_none_or(|r| a.route == r)).collect();
757 count(name, of.iter().filter(|a| a.label == label).count(), of.len())
758 };
759 for (name, label) in [("unclear", Label::Unclear), ("not arisen", Label::NotArisen), ("consequence remains", Label::Remains)] {
760 labels.push(share(name, label, None));
761 }
762 for (name, route) in [("unclear, asked", select::Route::Ask), ("unclear, free", select::Route::Free), ("unclear, placebo", select::Route::Placebo)] {
763 labels.push(share(name, Label::Unclear, Some(route)));
764 }
765 let later = s.applies.iter().filter(|a| matches!(a.route, select::Route::Not | select::Route::Placebo)).count();
766 let (placebos_judged, placebos) = of(select::Route::Placebo);
767 labels.push(count("placebo", placebos, later));
768 labels.retain(|c| c.whole > 0);
769 let measurement = (checks > 0).then_some(Measurement {
770 placebos: placebos as u64,
771 judged: placebos_judged as u64,
772 tail_shown: s.pool.tail.0,
773 tail_applied: s.pool.tail.1,
774 placebos_needed: select::policy::NOT_LEARNED_UNTIL,
775 });
776 let share = |stripped: bool| {
778 let of: Vec<_> = s.applies.iter().filter(|a| a.route == select::Route::Ask && judged(a.label) && a.stripped == stripped).collect();
779 (of.iter().filter(|a| a.label == Label::Changed).count() as f64, of.len() as f64)
780 };
781 let ((c1, n1), (c0, n0)) = (share(false), share(true));
782 let declaration = (n1 > 0.0 && n0 > 0.0).then(|| {
783 let (r1, r0) = (c1 / n1, c0 / n0);
784 let se = (r1 * (1.0 - r1) / n1 + r0 * (1.0 - r0) / n0).sqrt();
785 let mean = 100.0 * (r1 - r0);
786 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 }
787 });
788 let pool = s
789 .pool
790 .bins
791 .iter()
792 .enumerate()
793 .filter(|(_, (shown, _))| *shown > 0)
794 .map(|(b, &(shown, applied))| Yield {
795 from: (1u32 << b) - 1,
796 to: (1u32 << (b + 1)) - 1,
797 shown,
798 applied,
799 rate: s.fit.pool.yields.get(b).map_or(applied as f64 / shown as f64, |(a, b)| a / (a + b)),
800 })
801 .collect();
802 let (allowance, ledger) = if learned && !reverted { select::policy::gate_parts(&s.gate, risk.unwrap_or(select::policy::ALPHA)) } else { (0.0, 0.0) };
803 let mon = &s.monitor;
804 let evaluation = (mon.sessions > 0).then(|| Evaluation {
805 sessions: u64::from(mon.sessions),
806 mean: mon.mean,
807 ucb: mon.ucb,
808 units: u64::from(mon.labelled),
809 ess: mon.ess,
810 max_weight: mon.max_weight,
811 truncated: u64::from(mon.truncated),
812 });
813 let latency = s
814 .pool
815 .stops
816 .iter()
817 .filter(|(_, calls)| !calls.is_empty())
818 .map(|(harness, calls)| {
819 let mut ms: Vec<u64> = calls.iter().map(|&(ms, _)| u64::from(ms)).collect();
820 ms.sort_unstable();
821 let budget = crate::harness::ALL.iter().find(|h| h.name() == harness).and_then(|h| h.caps().stop_budget);
822 Latency {
823 harness: harness.clone(),
824 checks: ms.len() as u64,
825 p50_ms: quantile(&ms, QUANTILES[0]),
826 p95_ms: quantile(&ms, QUANTILES[1]),
827 budget_ms: budget.map(|b| u64::try_from(b.as_millis()).unwrap_or(u64::MAX)),
828 over: calls.iter().filter(|(_, failed)| *failed).count() as u64,
829 }
830 })
831 .collect();
832 Selection {
833 policy: policy.name(),
834 default,
835 source: if reverted { select::REVERTED_SOURCE } else { policy.name() },
836 gate: if learned && !reverted { select::policy::remaining_at(&s.gate, risk.unwrap_or(select::policy::ALPHA)) } else { 0.0 },
837 allowance,
838 ledger,
839 fit: s.fit.key.clone(),
840 at: s.at,
841 checks,
842 sessions,
843 deviations,
844 fallbacks,
845 spend,
846 budget,
847 effects: s.fit.model.as_ref().map(|m| effects(m, s.fit.labels)).unwrap_or_default(),
848 half_life: s.fit.model.as_ref().map(|_| s.fit.half_life.unwrap_or(0)),
849 labels,
850 declaration,
851 delivery: (!s.fit.key.is_empty()).then(|| s.fit.rates.clone()),
852 evaluation,
853 measurement,
854 pool,
855 latency,
856 test: (learned || s.experiment.totals() != (0, 0)).then(|| Test::of(s, risk)),
857 judge: Vec::new(),
858 boundary: 0.0,
859 misses: None,
860 }
861 }
862
863 #[must_use]
866 pub fn shown(&self) -> bool {
867 self.checks > 0 || self.source != self.default
868 }
869}
870
871const Z95: f64 = 1.96;
873
874fn effects(m: &select::policy::Model, n: u32) -> Vec<Estimate> {
878 let p = select::policy::P;
879 if m.beta.len() != p || m.chol.len() != p * p {
880 return Vec::new();
881 }
882 let sd = |k: usize| m.chol[k * p..=k * p + k].iter().map(|x| x * x).sum::<f64>().sqrt();
884 let mut out: Vec<Estimate> = (0..p)
885 .map(|k| Estimate {
886 name: select::policy::NAMES[k].into(),
887 unit: "coefficient".into(),
888 mean: m.beta[k],
889 lo: m.beta[k] - Z95 * sd(k),
890 hi: m.beta[k] + Z95 * sd(k),
891 n: u64::from(n),
892 })
893 .collect();
894 for (name, v) in ["session spread", "lesson spread", "judge spread"].into_iter().zip(m.spread) {
895 out.push(Estimate { name: name.into(), unit: "logit".into(), mean: v, lo: v, hi: v, n: u64::from(n) });
896 }
897 out
898}
899
900#[must_use]
902pub fn selection(lessons: &Path, policy: select::Policy) -> Selection {
903 let mut s = Selection::of(&select::summary::load(lessons), policy, select::risk_of(lessons));
904 let current = crate::hook::judge_model(&std::env::var("BORG_PREDICT_MODEL").unwrap_or_default());
906 s.judge = crate::judge_length::lengths(&lessons.with_file_name("state"), ¤t);
907 s.boundary = crate::whole::boundary_share();
908 s.misses = lessons.parent().map(crate::pools::misses).filter(|m| m.checks > 0);
910 s
911}
912
913fn grouped(n: usize) -> String {
915 let digits = n.to_string();
916 let mut out = String::new();
917 for (i, c) in digits.chars().enumerate() {
918 if i > 0 && (digits.len() - i).is_multiple_of(3) {
919 out.push(',');
920 }
921 out.push(c);
922 }
923 out
924}
925
926#[must_use]
929pub fn judge_length_line(l: &crate::judge_length::Length) -> String {
930 use crate::judge_length::Decision;
931 let start = crate::hook::END_START_BUDGET;
932 let how = if l.start {
933 "the start: no decided step moved it".to_string()
934 } else if l.chars > start {
935 format!("measured, up from the start {}", grouped(start))
936 } else {
937 format!("measured, down from the start {}", grouped(start))
938 };
939 let mut parts = vec![format!("{}: session-end budget {} chars ({how})", l.model, grouped(l.chars))];
940 let measured: Vec<String> = l
941 .steps
942 .iter()
943 .filter(|s| s.sent > 0)
944 .map(|s| {
945 let what = match s.decision {
946 Decision::Reference => format!("reference, line {:.0}%", 100.0 * l.line.unwrap_or(0.0)),
947 Decision::Below => "below the reference".into(),
948 Decision::Holds => "holds".into(),
949 Decision::Fails => "fails".into(),
950 Decision::Open => "open".into(),
951 };
952 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)
953 })
954 .collect();
955 parts.push(if measured.is_empty() { "no judge call measured yet".into() } else { format!("answered by prompt size: {}", measured.join(", ")) });
956 if l.line.is_none() && !measured.is_empty() {
957 parts.push(format!(
958 "no reference yet (the smallest steps with {} lessons in {} calls)",
959 crate::judge_length::REFERENCE_LESSONS,
960 crate::judge_length::MIN_CALLS
961 ));
962 }
963 parts.push(match l.explore {
964 Some(up) if l.steps.iter().any(|s| s.upto == up && s.decision == Decision::Fails) => {
965 format!("exploring the failed ≤{} on {:.0}% of session-end passes", grouped(up), 100.0 * l.psi)
966 }
967 Some(up) => format!("exploring ≤{} on {:.0}% of session-end passes", grouped(up), 100.0 * l.psi),
968 None => "nothing explored".into(),
969 });
970 parts.join(" · ")
971}
972
973#[must_use]
982pub fn selection_line(s: &Selection) -> String {
983 let default = s.source == s.default;
984 let head = match s.source {
985 select::REVERTED_SOURCE => "REVERTED to the fixed rule: the monitor saw harm beyond the gate's share (sticky)".to_string(),
986 "learned" => {
987 let opt = if default { "" } else { " (opt-in, BORG_SELECT=learned)" };
988 if s.gate > 0.0 {
989 format!("learned{opt}, gate {:.3} (quality it may still risk below the fixed rule)", s.gate)
990 } else if s.fit.is_empty() || s.allowance <= 0.0 {
991 format!("learned{opt}, cold start: gate 0, every decision the fixed rule's")
992 } else {
993 format!("learned{opt}, gate spent (ledger q05 {:+.3} against allowance {:.3}), deviations need positive evidence", s.ledger, s.allowance)
996 }
997 }
998 "fixed" if default => "the fixed rule (the default; BORG_SELECT=learned opts in to the learned policy)".into(),
999 "fixed" => "the fixed rule (BORG_SELECT=fixed)".into(),
1000 "cap" => "the fixed rule with an ask cap (BORG_STOP_ASK_CAP, evaluation only)".into(),
1001 "attention" => "attention's cut (BORG_STOP_ATTENTION=on, evaluation only)".into(),
1002 other => other.to_string(),
1003 };
1004 if s.checks == 0 {
1005 return format!("{head} · no graded check folded yet");
1006 }
1007 let mut parts = vec![head, format!("{} checks", s.checks)];
1008 if !s.deviations.is_empty() {
1009 parts.push(format!("deviated: {}", s.deviations.iter().map(|c| format!("{} {}/{}", c.name, c.part, c.whole)).collect::<Vec<_>>().join(", ")));
1010 }
1011 let fell: Vec<String> = s.fallbacks.iter().filter(|c| c.part > 0).map(|c| format!("{} {}", c.name, c.part)).collect();
1012 if !fell.is_empty() {
1013 parts.push(format!("learned decisions the fixed rule took over: {}", fell.join(", ")));
1014 }
1015 if let Some(t) = &s.test {
1016 parts.push(test_line(t));
1017 }
1018 if let Some(m) = &s.measurement {
1019 parts.push(format!("placebos {} drawn, {} judged", m.placebos, m.judged));
1020 if m.tail_shown > 0 {
1021 parts.push(format!("tail {} shown past the pool, {} applied", m.tail_shown, m.tail_applied));
1022 }
1023 }
1024 if let Some(m) = &s.misses {
1025 parts.push(misses_line(m, s.boundary));
1026 }
1027 if let Some(e) = &s.evaluation {
1028 let reading = match e.ucb {
1029 Some(u) => format!("monitor {:+.3}, bound {u:+.3} over {} sessions (below 0 reverts)", e.mean, e.sessions),
1030 None => format!("monitor {:+.3} over {} sessions, no bound yet", e.mean, e.sessions),
1031 };
1032 parts.push(format!("{reading}, ESS {:.0} of {} weighed, largest weight {:.1}× the mean", e.ess, e.units, e.max_weight));
1033 }
1034 if !s.latency.is_empty() {
1035 let secs = |ms: u64| format!("{:.1} s", ms as f64 / 1000.0);
1036 let each: Vec<String> = s
1037 .latency
1038 .iter()
1039 .map(|l| {
1040 let budget = l.budget_ms.map_or_else(String::new, |b| format!(" of {}", secs(b)));
1041 let failed = if l.over > 0 { format!(", {} failed", l.over) } else { String::new() };
1042 format!("{} {}{budget}{failed}", l.harness, secs(l.p95_ms))
1043 })
1044 .collect();
1045 parts.push(format!("stop p95 {}", each.join("; ")));
1046 }
1047 if s.sessions > 0 && !s.spend.is_empty() {
1048 let n = s.sessions as f64;
1049 let (usd, rounds) = s.spend.iter().fold((0.0, 0), |(u, r), x| (u + x.usd, r + x.rounds));
1050 parts.push(format!("${:.2} and {:.1} extra rounds per session ({} sessions)", usd / n, rounds as f64 / n, s.sessions));
1051 }
1052 if let Some(b) = &s.budget {
1053 parts.push(format!("budget {:.0} pool tokens per session: {:.0} read, price {:.2e} per token", b.tokens, b.spent, b.price));
1054 }
1055 parts.join(" · ")
1056}
1057
1058#[must_use]
1062pub fn misses_line(m: &crate::pools::Misses, share: f64) -> String {
1063 let size = format!("{:.1} MB", m.at as f64 / 1_048_576.0);
1064 let explore = if share > 0.0 {
1065 format!("exploration {:.0}% of the budget and of the dropped lessons", 100.0 * share)
1066 } else {
1067 "exploration off (BORG_BOUNDARY=off)".into()
1068 };
1069 let head = format!("pool log {} checks ({size}), {explore}", m.checks);
1070 if m.positives == 0 {
1071 return format!("{head}, no first apply at grade ≥ 2 yet");
1072 }
1073 let mut causes: Vec<&(String, u64)> = m.causes.iter().filter(|(_, n)| *n > 0).collect();
1074 causes.sort_by_key(|c| std::cmp::Reverse(c.1));
1075 let why = if causes.is_empty() { String::new() } else { format!(": {}", causes.iter().map(|(c, n)| format!("{c} {n}")).collect::<Vec<_>>().join(", ")) };
1076 format!("{head}, {} of {} first applies at grade ≥ 2 ranked low ({} shown by a draw){why}", m.low, m.positives, m.drawn)
1077}
1078
1079#[must_use]
1085pub fn async_line(t: &crate::stats::AsyncTally, on: bool, limit: std::time::Duration, line: &str) -> String {
1086 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)" };
1087 let mut parts = vec![head.to_string()];
1088 if on {
1089 parts.push(format!("call limit now {} s", limit.as_secs()));
1090 }
1091 if t.checks + t.superseded + t.gone + t.unstarted > 0 {
1092 let at_limit = format!("{} at the limit, {} s last", t.timed_out, t.limit_ms / 1000);
1093 parts.push(format!("{} detached checks (${:.2}), {} failed ({at_limit})", t.checks, t.usd, t.failed));
1094 parts.push(format!("{} superseded by a newer stop, {} after the session ended", t.superseded, t.gone));
1095 if t.unstarted > 0 {
1096 parts.push(format!("{} never started (their job could not be written or run)", t.unstarted));
1097 }
1098 parts.push(format!("peak memory {} MB", t.rss_kb.div_ceil(1024)));
1099 parts.push(format!("{:.1} lessons shown per check (line {line})", t.lessons));
1100 }
1101 parts.join(" · ")
1102}
1103
1104fn test_line(t: &Test) -> String {
1109 let risk = if t.set { format!("risk {:.2} (the collective's)", t.risk) } else { format!("risk {:.2}", t.risk) };
1110 match (t.ran, t.risk <= 0.0) {
1111 (0, true) => return "question-count test off (risk 0)".into(),
1112 (0, false) => return format!("question-count test: no check yet, {risk}"),
1113 _ => {}
1114 }
1115 let kept = if t.kept < t.ran { format!(", {} still in the summary", t.kept) } else { String::new() };
1116 let head =
1117 format!("question-count test {} drawn ({} control, {} refused by the gate{kept}) in {} sessions, {risk}", t.ran, t.control, t.refused, t.sessions);
1118 match (&t.ask_free, t.due) {
1119 (Some(e), true) => format!("{head}; readout: grade 2 asked − handed over {:+.1} [{:+.1}, {:+.1}] points changed", e.mean, e.lo, e.hi),
1120 (None, true) => format!("{head}; readout due, no judged grade 2 on both routes yet"),
1121 (_, false) => format!("{head}; readout after {} checks in {} sessions, or {} sessions", t.due_checks, t.due_min_sessions, t.due_sessions),
1122 }
1123}
1124
1125#[must_use]
1127pub fn render_with(name: &str, share: Share, drones: &[Drone], inbox: &Inbox, select: Option<&Selection>, now: i64) -> String {
1128 let mut s = render(name, share, drones, inbox, now);
1129 if let Some(sel) = select.filter(|s| s.shown()) {
1130 s.push_str(&format!(" selection: {}\n", selection_line(sel)));
1131 }
1132 s
1133}
1134
1135pub fn version_line<'a>(stores: impl IntoIterator<Item = &'a Path>, checked: Option<&update::Checked>, check: bool) -> Option<String> {
1141 let mut parts: Vec<String> = Vec::new();
1142 if let Some(l) = checked.and_then(|c| c.latest.as_deref()).filter(|l| check && update::newer(l, update::VERSION)) {
1143 parts.push(format!("borg {l} is released, this is {}: borg update", update::VERSION));
1144 }
1145 let aliases = collective::aliases();
1146 let ours = protocol::current();
1147 for home in stores.into_iter().filter(|h| h.join("mesh").is_dir()) {
1148 let versions = protocol::versions(home);
1149 let mismatches = protocol::mismatches(home);
1150 let name = |key: &str| aliases.display(&protocol::member(key, versions.get(key)));
1151 for (key, m) in &mismatches {
1152 parts.push(format!("nothing syncs with {} ({}, this node {})", name(key), m.theirs, m.ours));
1153 }
1154 for (key, v) in versions.iter().filter(|(k, v)| v.protocol < ours && !mismatches.contains_key(*k)) {
1155 parts.push(format!("{} on /{} misses {}", name(key), v.protocol, protocol::misses(v.protocol, ours)));
1156 }
1157 }
1158 parts.dedup();
1159 (!parts.is_empty()).then(|| parts.join(" · "))
1160}
1161
1162fn routes(dir: Option<&str>) -> Result<Vec<Route>, String> {
1164 let cfg = collective::load()?;
1165 let base = learn::home();
1166 Ok(match dir {
1167 Some(d) => vec![collective::route(cfg.as_ref(), d, &base)].into_iter().filter(|r| *r != Route::Off).collect(),
1168 None => install::collectives(cfg.as_ref(), &base),
1169 })
1170}
1171
1172pub fn gather(dir: Option<&str>, now: i64) -> Result<(Vec<crate::top_live::Coll>, Option<String>), String> {
1175 let mut out = Vec::new();
1176 let routes = routes(dir)?;
1177 let stores: Vec<&Path> = routes
1178 .iter()
1179 .filter_map(|r| match r {
1180 Route::Collective { home, .. } => Some(home.as_path()),
1181 Route::Off => None,
1182 })
1183 .collect();
1184 let version = version_line(stores, update::checked().as_ref(), update::on());
1185 for r in &routes {
1186 let Route::Collective { name, home, share } = r else { continue };
1187 presence::prune(home, now);
1188 let mut ds = drones(home, collective::may_sync(r), now);
1189 for d in &mut ds {
1190 d.p.engineer = collective::aliases().display(&d.p.engineer);
1191 }
1192 let inbox = inbox(home, &ds, now);
1193 if ds.is_empty() && inbox.recent.is_empty() && dir.is_none() {
1194 continue;
1195 }
1196 let lessons = home.join("lessons.jsonl");
1199 let summary = select::summary::peek(&lessons);
1200 let select = Some(Selection::of(&summary, select::policy(), select::risk_of(&lessons))).filter(Selection::shown).map(|s| selection_line(&s));
1201 out.push(crate::top_live::Coll { name: name.clone(), home: home.clone(), share: *share, drones: ds, inbox, select });
1202 }
1203 Ok((out, version))
1204}
1205
1206fn screen(dir: Option<&str>, json_out: bool, now: i64) -> Result<String, String> {
1208 let routes = routes(dir)?;
1209 let text = screen_of(&routes, dir.is_some(), &engineer_dir(dir), msg::caller, json_out, now);
1210 let stores = routes.iter().filter_map(|r| match r {
1211 Route::Collective { home, .. } => Some(home.as_path()),
1212 Route::Off => None,
1213 });
1214 Ok(match version_line(stores, update::checked().as_ref(), update::on()).filter(|_| !json_out) {
1215 Some(v) => format!("{v}\n\n{text}"),
1216 None => text,
1217 })
1218}
1219
1220fn screen_of(
1225 routes: &[Route],
1226 one: bool,
1227 cwd: &str,
1228 caller: impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String>,
1229 json_out: bool,
1230 now: i64,
1231) -> String {
1232 let mut text = String::new();
1233 let mut all = Vec::new();
1234 let (mut gathered, mut printed) = (Vec::new(), Vec::new());
1237 for r in routes {
1238 let Route::Collective { name, home, share } = r else { continue };
1239 presence::prune(home, now);
1240 let mut ds = drones(home, collective::may_sync(r), now);
1241 for d in &mut ds {
1243 d.p.engineer = collective::aliases().display(&d.p.engineer);
1244 }
1245 let inbox = inbox(home, &ds, now);
1246 let sel = selection(&home.join("lessons.jsonl"), select::policy());
1247 if json_out {
1248 let mut v = to_json_with(name, *share, &ds, &inbox);
1249 v["select"] = json!(sel);
1250 all.push(v);
1251 } else if !ds.is_empty() || !inbox.recent.is_empty() || one {
1252 text.push_str(&render_with(name, *share, &ds, &inbox, Some(&sel), now));
1253 text.push('\n');
1254 printed.push((home, inbox.recent.into_iter().filter(|m| prints_text(&ds, &inbox.here, m)).collect::<Vec<_>>()));
1255 }
1256 gathered.push(ds);
1257 }
1258 if !printed.is_empty() && engineer_looks(caller, gathered.iter().map(Vec::as_slice)) {
1259 let mut done = HashSet::new();
1260 for (home, recent) in &printed {
1261 acknowledge(home, cwd, &recent.iter().collect::<Vec<_>>(), &mut done, now);
1262 }
1263 }
1264 if json_out {
1265 return serde_json::to_string_pretty(&all).unwrap_or_default() + "\n";
1266 }
1267 if text.is_empty() {
1268 text = "no drones (borg top lists sessions whose hooks ran in the last day)\n".into();
1269 }
1270 text
1271}
1272
1273pub fn main(args: &[&str]) -> i32 {
1274 let (mut dir, mut once, mut json_out) = (None, false, false);
1275 let mut it = args.iter();
1276 while let Some(a) = it.next() {
1277 match *a {
1278 "--dir" if dir.is_none() => match it.next() {
1279 Some(d) => dir = Some(*d),
1280 None => return usage("--dir needs a value"),
1281 },
1282 "--once" => once = true,
1283 "--json" => json_out = true,
1284 a => return usage(&format!("unknown flag {a}")),
1285 }
1286 }
1287 if !json_out && !once && unsafe { libc::isatty(1) } == 1 {
1289 return crate::top_live::run(dir);
1290 }
1291 match screen(dir, json_out, learn::now_ms()) {
1292 Ok(s) => {
1293 print!("{s}");
1294 0
1295 }
1296 Err(e) => {
1297 eprintln!("borg top: {e}");
1298 1
1299 }
1300 }
1301}
1302
1303fn usage(why: &str) -> i32 {
1304 eprintln!("borg top: {why}\nusage: borg top [--once] [--json] [--dir DIR]");
1305 2
1306}
1307
1308#[cfg(test)]
1309pub(crate) fn tests_fixture() -> Vec<Drone> {
1310 tests::fixture()
1311}
1312
1313#[cfg(test)]
1314pub(crate) fn tests_inbox() -> Inbox {
1315 tests::inbox_fixture()
1316}
1317
1318#[cfg(test)]
1319mod tests {
1320 #[test]
1321 fn the_async_line_says_where_the_check_runs_and_what_it_did() {
1322 use crate::stats::AsyncTally;
1323 let secs = std::time::Duration::from_secs(45);
1324 assert_eq!(super::async_line(&AsyncTally::default(), false, secs, "a2"), "at the stop, which waits for it (BORG_ASYNC=off)");
1325 let fresh = super::async_line(&AsyncTally::default(), true, secs, "a2");
1326 assert_eq!(fresh, "detached, the stop never waits (BORG_ASYNC=off checks at the stop) · call limit now 45 s");
1327 let t =
1328 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 };
1329 let l = super::async_line(&t, true, secs, "a2");
1330 for part in [
1331 "12 detached checks ($0.50), 2 failed (1 at the limit, 45 s last)",
1332 "3 superseded by a newer stop, 1 after the session ended",
1333 "peak memory 38 MB",
1334 "41.3 lessons shown per check (line a2)",
1335 ] {
1336 assert!(l.contains(part), "{part}: {l}");
1337 }
1338 assert!(!l.contains("never started"), "{l}");
1339 let lost = super::async_line(&AsyncTally { unstarted: 2, ..t }, true, secs, "a2");
1340 assert!(lost.contains("2 never started (their job could not be written or run)"), "{lost}");
1341 }
1342
1343 use super::*;
1344
1345 fn p(session: &str, engineer: &str, started: i64, ms: i64, event: &str, prompt: &str, files: &[&str]) -> Presence {
1346 Presence {
1347 session: session.into(),
1348 harness: "claude".into(),
1349 engineer: engineer.into(),
1350 repo: "borg-ai".into(),
1351 collective: "corp".into(),
1352 prompt: prompt.into(),
1353 files: files.iter().map(|f| f.to_string()).collect(),
1354 event: event.into(),
1355 ms,
1356 started,
1357 pid: None,
1358 }
1359 }
1360
1361 const NOW: i64 = 10_000_000;
1362
1363 pub(crate) fn fixture() -> Vec<Drone> {
1364 let local = vec![
1365 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"]),
1366 p("cccc3333-x", "alice@x", 300, NOW - presence::IDLE_MS - 60_000, "stop", "write the release notes for the mesh daemon and more words", &[]),
1367 ];
1368 let remote = vec![presence::Remote {
1369 p: p("bbbb2222-y", "bob@y", 200, NOW - 120_000, "stop", "review PR 42", &["README.md"]),
1370 key: "beefcafe".repeat(8),
1371 sig: String::new(),
1372 }];
1373 assemble(local, remote, NOW, |_| true, |p| p.session.starts_with("aaaa"))
1374 }
1375
1376 fn message(by: &str, name: &str, to: &str, kind: msg::Kind, ms: i64, text: &str) -> Message {
1377 message_on("n1", by, name, to, kind, ms, text)
1378 }
1379
1380 fn message_on(node: &str, by: &str, name: &str, to: &str, kind: msg::Kind, ms: i64, text: &str) -> Message {
1381 let from = msg::Sender { by: by.into(), name: name.into(), harness: "claude".into(), node: node.into() };
1382 Message::new(kind, &from, to, None, ms, text)
1383 }
1384
1385 pub(crate) fn inbox_fixture() -> Inbox {
1386 Inbox {
1387 unread: [("aaaa1111-x".to_string(), 2), ("cccc3333-x".to_string(), 0)].into_iter().collect(),
1388 recent: vec![
1389 message("cccc3333-x", "Three of Three", "aaaa1111-x", msg::ASK, NOW - 60_000, "which test\nflakes?"),
1390 message("engineer:alice@x", "alice@x", msg::ALL, msg::TELL, NOW - 120_000, "release at five"),
1391 message_on(&"feed".repeat(16), "gone-session", "Six of Six", "zzzz9999-q", msg::TELL, NOW - 180_000, "done"),
1392 ],
1393 day: 3,
1394 here: "n1".into(),
1395 listen: [("aaaa1111-x".to_string(), crate::listen::Status::Waits), ("cccc3333-x".to_string(), crate::listen::Status::Listens)]
1396 .into_iter()
1397 .collect(),
1398 }
1399 }
1400
1401 #[test]
1402 fn renders_one_screen_per_collective() {
1403 let out = render("corp", Share::Team, &fixture(), &inbox_fixture(), NOW);
1404 let expected = "\
1405corp (team): 3 drone(s), 2 active, 1 remote
1406 DRONE SESSION ENGINEER REPO STATE WAKE IDLE UNREAD DOING FILES
1407 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
1408 Two of Three bbbb222… bob@y @beefca borg-ai waiting - 2m - review PR 42 README.md
1409 Three of Three cccc333… alice@x borg-ai idle on 31m 0 write the release notes for the mesh da…
1410 messages: 3 in the last day, 2 unread
1411 1m Three of Three asks -> One of Three: which test flakes?
1412 2m alice@x -> collective: release at five
1413 3m Six of Six @feedfeedfeed -> zzzz9999: done
1414";
1415 assert_eq!(out, expected, "\n{out}");
1416 let none = Inbox::default();
1417 assert_eq!(render("private", Share::Local, &[], &none, NOW), "private (local, never leaves this machine): 0 drone(s), 0 active, 0 remote\n");
1418 }
1419
1420 fn learned() -> Selection {
1424 let spend = |source: &str, rounds, usd| Spend { source: source.into(), calls: 1, rounds, usd };
1425 let evaluation = Evaluation { sessions: 12, mean: 0.004, ucb: Some(0.031), units: 300, ess: 180.0, max_weight: 6.5, truncated: 3 };
1426 Selection {
1427 policy: "learned",
1428 default: "fixed",
1429 source: "learned",
1430 gate: 0.12,
1431 checks: 480,
1432 sessions: 12,
1433 deviations: vec![Count { name: "route".into(), part: 5, whole: 400 }],
1434 spend: vec![spend("stop", 4, 3.0), spend("live", 0, 2.04)],
1435 evaluation: Some(evaluation),
1436 measurement: Some(Measurement { placebos: 40, judged: 31, tail_shown: 12, tail_applied: 3, placebos_needed: 100 }),
1437 test: Some(Test {
1438 risk: 0.1,
1439 ran: 30,
1440 control: 9,
1441 refused: 4,
1442 sessions: 6,
1443 kept: 30,
1444 due_checks: 300,
1445 due_sessions: 30,
1446 due_min_sessions: 15,
1447 ..Test::default()
1448 }),
1449 ..Selection::default()
1450 }
1451 }
1452
1453 #[test]
1457 fn the_judges_effective_length_is_read_from_the_stores_cost_lines_and_shown() {
1458 let home = std::env::temp_dir().join(format!("borg-top-judge-length-{}", std::process::id()));
1459 let _ = std::fs::remove_dir_all(&home);
1460 std::fs::create_dir_all(home.join("state")).unwrap();
1461 let line = |chars: usize, n: u64, a: u64| {
1462 format!(
1463 "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",
1464 n - a
1465 )
1466 .repeat(10)
1467 };
1468 std::fs::write(home.join("state/s.cost"), line(30_000, 40, 20) + &line(70_000, 30, 15) + &line(100_000, 30, 15)).unwrap();
1469 let sel = selection(&home.join("lessons.jsonl"), select::policy());
1470 let big = sel.judge.iter().find(|l| l.model == "x:big").expect("the store's judge");
1471 assert_eq!((big.chars, big.start, big.explore), (108_000, false, Some(162_000)));
1472 assert_eq!(
1473 judge_length_line(big),
1474 "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"
1475 );
1476 let v = serde_json::to_value(&sel).unwrap();
1477 assert_eq!(v["judge"].as_array().map(|a| a.iter().any(|j| j["model"] == "x:big" && j["chars"] == 108_000)), Some(true), "{v}");
1478 let none = crate::judge_length::Length::of("y:new", &crate::judge_length::Steps::new());
1479 assert_eq!(
1480 judge_length_line(&none),
1481 "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"
1482 );
1483 assert_eq!((grouped(0), grouped(999), grouped(1_000), grouped(1_234_567)), ("0".into(), "999".into(), "1,000".into(), "1,234,567".into()));
1484 let _ = std::fs::remove_dir_all(&home);
1485 }
1486
1487 #[test]
1490 fn the_misses_line_names_the_causes_of_low_first_applies() {
1491 let causes = |v: &[(&str, u64)]| v.iter().map(|(c, n)| (c.to_string(), *n)).collect();
1492 let m = crate::pools::Misses {
1493 at: 3 * 1_048_576 / 2,
1494 checks: 40,
1495 positives: 12,
1496 low: 5,
1497 drawn: 2,
1498 causes: causes(&[("scope", 0), ("no_word_match", 1), ("demoted", 3), ("other", 1)]),
1499 };
1500 assert_eq!(
1501 misses_line(&m, 0.15),
1502 "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"
1503 );
1504 let none = crate::pools::Misses { positives: 0, ..m };
1505 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));
1506 }
1507
1508 #[test]
1509 fn the_selection_line_says_what_decides_and_prices_spend_only_under_a_budget() {
1510 use select::Policy;
1511 use select::summary::Summary;
1512 let fresh = |policy, default| Selection { default, ..Selection::of(&Summary::default(), policy, None) };
1513 let fixed = fresh(Policy::Fixed, "fixed");
1515 assert_eq!(selection_line(&fixed), "the fixed rule (the default; BORG_SELECT=learned opts in to the learned policy) · no graded check folded yet");
1516 assert!(!fixed.shown(), "the default with nothing folded: nothing to show on the drones screen");
1517 let opted = fresh(Policy::Learned, "fixed");
1518 assert_eq!(
1519 selection_line(&opted),
1520 "learned (opt-in, BORG_SELECT=learned), cold start: gate 0, every decision the fixed rule's · no graded check folded yet"
1521 );
1522 assert!(opted.shown(), "a switch decides: shown");
1523 assert_eq!(selection_line(&fresh(Policy::Fixed, "learned")), "the fixed rule (BORG_SELECT=fixed) · no graded check folded yet");
1525 assert!(!fresh(Policy::Learned, "learned").shown() && fresh(Policy::Fixed, "learned").shown());
1526 let d = select::precedence(None, None, false);
1527 assert_eq!(Selection::of(&Summary::default(), Policy::Cap(2), None).default, d.name(), "the default is select.rs's, no switch set");
1528 assert!(!Selection::of(&Summary::default(), d, None).shown());
1529 let mut s = learned();
1530 let line = selection_line(&s);
1531 assert_eq!(
1532 line,
1533 "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)"
1534 );
1535 assert!(!line.contains("budget") && !line.contains("price"), "no budget set: spend measured, never priced: {line}");
1536 let mut early = learned();
1537 early.evaluation = Some(Evaluation { ucb: None, sessions: 3, units: 40, ess: 12.4, max_weight: 3.0, ..Evaluation::default() });
1538 let line = selection_line(&early);
1539 assert!(line.contains(" · monitor +0.000 over 3 sessions, no bound yet, ESS 12 of 40 weighed, largest weight 3.0× the mean · "), "{line}");
1540 early.latency = vec![
1541 Latency { harness: "codex".into(), checks: 20, p50_ms: 10_000, p95_ms: 19_000, budget_ms: Some(30_000), over: 1 },
1542 Latency { harness: "claude".into(), checks: 1, p50_ms: 4_000, p95_ms: 4_000, budget_ms: None, over: 0 },
1543 ];
1544 let line = selection_line(&early);
1545 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}");
1546 s.budget = Some(Budget { tokens: 30_000.0, spent: 24_000.0, price: 0.0002 });
1547 assert!(selection_line(&s).ends_with(" · budget 30000 pool tokens per session: 24000 read, price 2.00e-4 per token"), "{}", selection_line(&s));
1548 s.source = select::REVERTED_SOURCE;
1549 assert!(selection_line(&s).starts_with("REVERTED to the fixed rule"), "{}", selection_line(&s));
1550 }
1551
1552 #[test]
1556 fn a_spent_gate_is_not_a_cold_start() {
1557 use select::Policy;
1558 use select::summary::{Gate, Summary};
1559 let line = |s: &Summary| selection_line(&Selection { default: "learned", ..Selection::of(s, Policy::Learned, None) });
1560 let mut s = Summary::default();
1561 s.fit.key = "3556601-bc6d6191".into();
1562 assert!(line(&s).starts_with("learned, cold start: gate 0, every decision the fixed rule's"), "{}", line(&s));
1564 s.gate = Gate { gain: vec![-0.1018; select::policy::DRAWS], allowance: 0.0760, ..Gate::default() };
1565 let spent = Selection::of(&s, Policy::Learned, None);
1566 assert_eq!((spent.gate, spent.allowance, spent.ledger), (0.0, 0.0760, -0.1018));
1567 assert!(line(&s).starts_with("learned, gate spent (ledger q05 -0.102 against allowance 0.076), deviations need positive evidence · "), "{}", line(&s));
1568 s.fit.key.clear();
1570 assert!(line(&s).starts_with("learned, cold start: gate 0"), "{}", line(&s));
1571 s.fit.key = "3556601-bc6d6191".into();
1573 s.gate.gain = vec![-0.05; select::policy::DRAWS];
1574 assert!(line(&s).starts_with("learned, gate 0.026 (quality it may still risk below the fixed rule)"), "{}", line(&s));
1575 let fixed = Selection::of(&s, Policy::Fixed, None);
1577 assert_eq!((fixed.allowance, fixed.ledger), (0.0, 0.0));
1578 }
1579
1580 #[test]
1583 fn a_store_says_how_its_lessons_are_selected() {
1584 use select::Policy;
1585 let home = std::env::temp_dir().join(format!("borg-top-select-{}", std::process::id()));
1586 let _ = std::fs::remove_dir_all(&home);
1587 std::fs::create_dir_all(&home).unwrap();
1588 let lessons = home.join("lessons.jsonl");
1589 std::fs::write(&lessons, "").unwrap();
1590 let line = |lessons: &Path, policy| selection_line(&selection(lessons, policy));
1591 assert!(line(&lessons, Policy::Learned).ends_with("cold start: gate 0, every decision the fixed rule's · no graded check folded yet"));
1592 assert!(line(&lessons, Policy::Fixed).starts_with("the fixed rule ("), "{}", line(&lessons, Policy::Fixed));
1593 assert!(line(&lessons, Policy::Cap(2)).contains("(BORG_STOP_ASK_CAP, evaluation only)"));
1594 let absent = home.join("absent").join("lessons.jsonl");
1595 assert!(line(&absent, Policy::Attention).contains("(BORG_STOP_ATTENTION=on, evaluation only)"), "no store: the prior");
1596 let _ = std::fs::remove_dir_all(&home);
1597 }
1598
1599 #[test]
1601 fn the_selection_line_follows_the_screen_when_shown() {
1602 let (ds, inbox) = (fixture(), inbox_fixture());
1603 let plain = render("corp", Share::Team, &ds, &inbox, NOW);
1604 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, None, NOW), plain);
1605 let cold = Selection::of(&select::summary::Summary::default(), select::precedence(None, None, false), None);
1606 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, Some(&cold), NOW), plain, "the default at cold start: no line");
1607 let s = learned();
1608 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, Some(&s), NOW), format!("{plain} selection: {}\n", selection_line(&s)));
1609 }
1610
1611 #[test]
1613 fn the_selection_serialises_every_section_for_the_page() {
1614 let v = serde_json::to_value(learned()).unwrap();
1615 let mut keys: Vec<&str> = v.as_object().unwrap().keys().map(String::as_str).collect();
1616 keys.sort_unstable();
1617 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";
1618 assert_eq!(keys, want.split(' ').collect::<Vec<_>>());
1619 assert_eq!((&v["budget"], &v["spend"][0]["source"], &v["deviations"][0]["whole"]), (&Value::Null, &json!("stop"), &json!(400)));
1620 let mut keys: Vec<&str> = v["evaluation"].as_object().unwrap().keys().map(String::as_str).collect();
1621 keys.sort_unstable();
1622 assert_eq!(keys, ["ess", "max_weight", "mean", "sessions", "truncated", "ucb", "units"]);
1623 assert_eq!(
1624 (&v["default"], &v["measurement"]),
1625 (&json!("fixed"), &json!({"placebos": 40, "judged": 31, "tail_shown": 12, "tail_applied": 3, "placebos_needed": 100}))
1626 );
1627 let mut keys: Vec<&str> = v["test"].as_object().unwrap().keys().map(String::as_str).collect();
1628 keys.sort_unstable();
1629 let want = "ask_free by_k control dilution due due_checks due_min_sessions due_sessions kept ran refused risk sessions set";
1630 assert_eq!(keys, want.split(' ').collect::<Vec<_>>());
1631 }
1632
1633 #[test]
1638 fn the_judged_share_leaves_out_what_its_session_already_had() {
1639 use crate::outcome::{self, Outcome};
1640 use select::summary::{Summary, fold};
1641 let rec = |kind: &str, check: &str, channel: &str, ms: i64| Outcome {
1642 kind: kind.into(),
1643 session: "s".into(),
1644 lessons: vec!["A".into()],
1645 check: check.into(),
1646 channel: channel.into(),
1647 ms,
1648 ..Default::default()
1649 };
1650 let graded = |check: &str, ms: i64| Outcome {
1651 grades: vec![2],
1652 ranks: vec![0],
1653 applied: 1,
1654 free: vec!["A".into()],
1655 pf: vec![1.0],
1656 policy: "fixed".into(),
1657 shown: 20,
1658 tokens: 6_000,
1659 ..rec(outcome::GRADED, check, "live", ms)
1660 };
1661 let log = [
1662 graded("c1", 10),
1663 Outcome { route: outcome::ROUTE_FREE.into(), ..rec(outcome::DELIVERY, "c1", "", 11) },
1664 Outcome { judge: 2, ..rec(outcome::CHANGED, "c1", "queued", 12) },
1665 graded("c2", 20),
1666 Outcome { route: outcome::ROUTE_DROPPED_DONE.into(), ..rec(outcome::DELIVERY, "c2", "", 21) },
1667 Outcome { lessons: vec!["B".into()], ..rec(outcome::ASKED, "", "early", 25) },
1669 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) },
1670 Outcome { lessons: vec!["B".into()], route: outcome::ROUTE_ASKED.into(), ..rec(outcome::DELIVERY, "c3", "", 31) },
1671 Outcome { lessons: vec!["B".into()], judge: 2, ..rec(outcome::CHANGED, "c3", "live", 32) },
1672 ];
1673 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1674 let s = fold(&text, Summary::default());
1675 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");
1676 let sel = Selection::of(&s, select::Policy::Fixed, None);
1677 let row = |name: &str| sel.labels.iter().find(|c| c.name == name).map(|c| (c.part, c.whole));
1678 assert_eq!(row("judged, free"), Some((1, 1)), "c1's delivered and judged; c2's had, out: {:?}", sel.labels);
1679 assert_eq!(row("judged, asked"), Some((1, 1)), "c3's question of what B's session had, judged: {:?}", sel.labels);
1680 }
1681
1682 #[test]
1686 fn the_verdict_shares_read_judge_three_answers_per_route() {
1687 use crate::outcome::{self, Outcome};
1688 use select::summary::{Summary, fold};
1689 let rec = |kind: &str, l: &str, check: &str, channel: &str, ms: i64, judge: u8| Outcome {
1690 kind: kind.into(),
1691 session: "s".into(),
1692 lessons: vec![l.into()],
1693 check: check.into(),
1694 channel: channel.into(),
1695 ms,
1696 judge,
1697 ..Default::default()
1698 };
1699 let graded = |check: &str, ms: i64, chosen: &[&str], free: &[&str], placebo: &[&str]| {
1700 let lessons: Vec<String> = chosen.iter().chain(free).chain(placebo).map(|l| l.to_string()).collect();
1701 let n = lessons.len();
1702 Outcome {
1703 grades: vec![2; n],
1704 ranks: (0..n as u16).collect(),
1705 applied: n as u16,
1706 chosen: chosen.iter().map(|l| l.to_string()).collect(),
1707 free: free.iter().map(|l| l.to_string()).collect(),
1708 placebo: placebo.iter().map(|l| l.to_string()).collect(),
1709 p: lessons.iter().map(|l| f64::from(u8::from(chosen.contains(&l.as_str())))).collect(),
1710 pf: lessons.iter().map(|l| f64::from(u8::from(free.contains(&l.as_str())))).collect(),
1711 psi: if placebo.is_empty() { 0.0 } else { 0.5 },
1712 policy: "fixed".into(),
1713 lessons,
1714 ..rec(outcome::GRADED, "", check, "stop", ms, 0)
1715 }
1716 };
1717 let deliver = |l: &str, check: &str, route: &str, ms: i64| Outcome { route: route.into(), ..rec(outcome::DELIVERY, l, check, "", ms, 0) };
1718 let log = [
1719 graded("c0", 5, &[], &["O"], &[]),
1721 deliver("O", "c0", outcome::ROUTE_FREE, 6),
1722 rec(outcome::UNCLEAR, "O", "c0", "queued", 7, 2),
1723 graded("c1", 10, &["A", "B"], &["F", "G", "H"], &["P", "Q"]),
1724 deliver("A", "c1", outcome::ROUTE_ASKED, 11),
1725 deliver("B", "c1", outcome::ROUTE_ASKED, 11),
1726 deliver("F", "c1", outcome::ROUTE_FREE, 11),
1727 deliver("G", "c1", outcome::ROUTE_FREE, 11),
1728 deliver("H", "c1", outcome::ROUTE_FREE, 11),
1729 rec(outcome::CHANGED, "A", "c1", "stop", 20, 3),
1730 rec(outcome::UNCLEAR, "B", "c1", "stop", 20, 3),
1731 rec(outcome::UNCLEAR, "F", "c1", "queued", 20, 3),
1732 Outcome { at: outcome::AT_END.into(), ..rec(outcome::NOT_ARISEN, "F", "c1", "queued", 30, 3) },
1733 rec(outcome::REMAINS, "G", "c1", "queued", 20, 3),
1734 rec(outcome::UNCLEAR, "H", "c1", "queued", 20, 3),
1735 rec(outcome::UNCLEAR, "P", "c1", outcome::PLACEBO, 20, 3),
1736 rec(outcome::IGNORED, "Q", "c1", outcome::PLACEBO, 20, 3),
1737 ];
1738 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1739 let sel = Selection::of(&fold(&text, Summary::default()), select::Policy::Fixed, None);
1740 let row = |name: &str| sel.labels.iter().find(|c| c.name == name).map(|c| (c.part, c.whole));
1741 assert_eq!(row("unclear"), Some((3, 7)), "B, H and P of seven judge-3 answers, O's older one out: {:?}", sel.labels);
1742 assert_eq!(row("not arisen"), Some((1, 7)));
1743 assert_eq!(row("consequence remains"), Some((1, 7)));
1744 assert_eq!(row("unclear, asked"), Some((1, 2)));
1745 assert_eq!(row("unclear, free"), Some((1, 3)), "F's end verdict replaced its unclear: {:?}", sel.labels);
1746 assert_eq!(row("unclear, placebo"), Some((1, 2)));
1747 }
1748
1749 #[test]
1754 fn the_measurement_counts_the_placebos_the_fixed_rule_drew_and_judged() {
1755 use crate::outcome::{self, Outcome};
1756 use select::summary::{Summary, fold};
1757 let rec = |kind: &str, lessons: &[&str], check: &str, channel: &str, ms: i64| Outcome {
1758 kind: kind.into(),
1759 session: "s".into(),
1760 lessons: lessons.iter().map(|l| l.to_string()).collect(),
1761 check: check.into(),
1762 channel: channel.into(),
1763 ms,
1764 ..Default::default()
1765 };
1766 let graded = Outcome {
1767 grades: vec![1, 1, 2],
1768 ranks: vec![0, 1, 2],
1769 applied: 3,
1770 free: vec!["C".into()],
1771 placebo: vec!["A".into(), "B".into()],
1772 psi: 0.5,
1773 policy: "fixed".into(),
1774 shown: 20,
1775 tokens: 6_000,
1776 ..rec(outcome::GRADED, &["A", "B", "C"], "c1", "stop", 10)
1777 };
1778 let tailed = Outcome {
1780 grades: vec![1],
1781 ranks: vec![0],
1782 applied: 1,
1783 policy: "fixed".into(),
1784 shown: 20,
1785 tokens: 6_000,
1786 tail: vec!["T1".into(), "T2".into()],
1787 pt: 0.5,
1788 fill: vec![21, 25, 21],
1789 ..rec(outcome::GRADED, &["T2"], "c2", "stop", 14)
1790 };
1791 let log = [
1792 graded,
1793 tailed,
1794 Outcome { route: outcome::ROUTE_PLACEBO.into(), ..rec(outcome::DELIVERY, &["A"], "c1", "", 11) },
1795 Outcome { route: outcome::ROUTE_PLACEBO.into(), ..rec(outcome::DELIVERY, &["B"], "c1", "", 11) },
1796 Outcome { judge: 2, ..rec(outcome::IGNORED, &["A"], "c1", outcome::PLACEBO, 12) },
1797 Outcome { judge: 2, ..rec(outcome::CHANGED, &["C"], "c1", "queued", 13) },
1798 ];
1799 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1800 let sel = Selection::of(&fold(&text, Summary::default()), select::Policy::Fixed, None);
1801 assert_eq!(
1802 sel.measurement,
1803 Some(Measurement { placebos: 2, judged: 1, tail_shown: 2, tail_applied: 1, placebos_needed: select::policy::NOT_LEARNED_UNTIL }),
1804 "{sel:?}"
1805 );
1806 assert!(selection_line(&sel).contains(" · 2 checks · placebos 2 drawn, 1 judged · tail 2 shown past the pool, 1 applied"), "{}", selection_line(&sel));
1807 assert_eq!(Selection::of(&Summary::default(), select::Policy::Fixed, None).measurement, None, "nothing folded: not measured");
1808 }
1809
1810 #[test]
1814 fn the_selection_counts_the_learned_decisions_the_fixed_rule_took_over_by_reason() {
1815 use crate::outcome::{self, Outcome};
1816 use select::summary::{Summary, fold};
1817 let graded = |check: &str, policy: &str, reason: &str, ms: i64| Outcome {
1818 kind: outcome::GRADED.into(),
1819 session: "s".into(),
1820 lessons: vec!["A".into()],
1821 grades: vec![1],
1822 ranks: vec![0],
1823 applied: 1,
1824 free: vec!["A".into()],
1825 pf: vec![1.0],
1826 channel: "live".into(),
1827 check: check.into(),
1828 policy: policy.into(),
1829 reason: reason.into(),
1830 ms,
1831 ..Default::default()
1832 };
1833 let log = [
1834 graded("c1", "learned", "", 1),
1835 graded("c2", "fixed", outcome::DECIDED_BUSY, 2),
1836 graded("c3", "fixed", outcome::DECIDED_NO_RECORD, 3),
1837 graded("c4", "fixed", outcome::DECIDED_BUSY, 4),
1838 ];
1839 let lines = |log: &[Outcome]| log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect::<String>();
1840 let sel = Selection::of(&fold(&lines(&log), Summary::default()), select::Policy::Learned, None);
1841 let rows: Vec<(&str, u64, u64)> = sel.fallbacks.iter().map(|c| (c.name.as_str(), c.part, c.whole)).collect();
1842 assert_eq!(rows, [("lock_busy", 2, 4), ("lock_failed", 0, 4), ("unrecorded", 1, 4)]);
1843 let line = selection_line(&sel);
1844 assert!(line.contains(" · 4 checks · learned decisions the fixed rule took over: lock_busy 2, unrecorded 1 · "), "{line}");
1845 let none = Selection::of(&fold(&lines(&log[..1]), Summary::default()), select::Policy::Learned, None);
1846 assert!(none.fallbacks.iter().all(|c| c.part == 0) && none.fallbacks.len() == 3, "{none:?}");
1847 assert!(!selection_line(&none).contains("took over"), "{}", selection_line(&none));
1848 assert!(Selection::of(&Summary::default(), select::Policy::Learned, None).fallbacks.is_empty(), "nothing folded: not measured");
1849 }
1850
1851 #[test]
1856 fn the_selection_fills_latency_evaluation_and_delivery_from_the_summary() {
1857 use select::summary::{Summary, fold};
1858 let call = |harness: &str, channel: &str, dur_ms: u64, failed: bool| {
1859 let o = crate::outcome::Outcome {
1860 kind: "graded".into(),
1861 session: "s".into(),
1862 channel: channel.into(),
1863 harness: harness.into(),
1864 dur_ms,
1865 failed,
1866 tokens: 6_000,
1867 ..Default::default()
1868 };
1869 format!("{}\n", serde_json::to_string(&o).unwrap())
1870 };
1871 let mut log: String = (1..=20).map(|k| call("codex", "stop", k * 1_000, k == 20)).collect();
1872 log += &call("claude", "stop", 4_000, false);
1873 log += &call("codex", "live", 99_000, false);
1874 log += &call("nonesuch", "stop", 5_000, false);
1875 let mut s = fold(&log, Summary::default());
1876 s.monitor = select::policy::Monitor {
1877 sessions: 12,
1878 units: 400,
1879 mean: 0.004,
1880 ucb: Some(0.031),
1881 labelled: 300,
1882 ess: 180.0,
1883 max_weight: 6.5,
1884 truncated: 3,
1885 ..Default::default()
1886 };
1887 s.fit.key = "fit".into();
1888 s.fit.rates = select::policy::Rates { free_stop: 0.6, free_live: 0.8, later: 0.4, wait: 0.25, ask_live: 1.0 };
1889 let sel = Selection::of(&s, select::Policy::Learned, None);
1890 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);
1891 assert!(budget("codex").is_some() && budget("claude").is_none(), "codex names a stop budget, claude none");
1892 let want = vec![
1893 Latency { harness: "claude".into(), checks: 1, p50_ms: 4_000, p95_ms: 4_000, budget_ms: None, over: 0 },
1894 Latency { harness: "codex".into(), checks: 20, p50_ms: 10_000, p95_ms: 19_000, budget_ms: budget("codex"), over: 1 },
1895 ];
1896 assert_eq!(sel.latency, want);
1897 let e = sel.evaluation.as_ref().expect("the monitor weighed labelled applies");
1898 assert_eq!(*e, Evaluation { sessions: 12, mean: 0.004, ucb: Some(0.031), units: 300, ess: 180.0, max_weight: 6.5, truncated: 3 });
1899 assert_eq!(sel.delivery.as_ref().map(|r| r.wait), Some(0.25));
1900 let keep = select::summary::LATENCY_KEEP as u64;
1902 let many: String = (1..=keep + 1).map(|k| call("claude", "stop", k, false)).collect();
1903 let kept = Selection::of(&fold(&many, Summary::default()), select::Policy::Learned, None);
1904 assert_eq!((kept.latency[0].checks, kept.latency[0].p50_ms), (keep, 1 + keep / 2), "the oldest call dropped");
1905 let cold = Selection::of(&Summary::default(), select::Policy::Learned, None);
1906 assert!(cold.latency.is_empty() && cold.evaluation.is_none() && cold.delivery.is_none(), "{cold:?}");
1907 assert_eq!((quantile(&[], 0.95), quantile(&[7], 0.5)), (0, 7));
1908 }
1909
1910 fn tested(check: &str, session: &str, ms: i64, source: &str, xk: Option<u32>, xs: Option<u32>) -> crate::outcome::Outcome {
1913 crate::outcome::Outcome {
1914 kind: crate::outcome::GRADED.into(),
1915 session: session.into(),
1916 lessons: vec!["A".into(), "B".into()],
1917 grades: vec![2, 1],
1918 ranks: vec![0, 1],
1919 applied: 2,
1920 check: check.into(),
1921 channel: "live".into(),
1922 policy: "learned".into(),
1923 source: source.into(),
1924 xk,
1925 xs,
1926 ms,
1927 ..Default::default()
1928 }
1929 }
1930
1931 #[test]
1937 fn the_test_section_is_the_summarys_fold_of_its_checks() {
1938 use select::policy::{ALPHA, EVIDENCE, EXPERIMENT, EXPERIMENT_CONTROL, GATE};
1939 use select::summary::{Summary, fold};
1940 let log = [
1941 tested("c1", "s1", 1, EXPERIMENT, Some(2), Some(1)),
1942 tested("c2", "s1", 2, EXPERIMENT_CONTROL, Some(2), Some(0)),
1943 tested("c3", "s2", 3, EXPERIMENT, Some(2), Some(1)),
1944 tested("c4", "s2", 4, GATE, Some(2), Some(1)),
1945 tested("c5", "s2", 5, GATE, Some(1), Some(0)),
1946 tested("c6", "s3", 6, EVIDENCE, None, None),
1947 tested("c7", "s3", 7, GATE, None, None),
1948 crate::outcome::Outcome { policy: "fixed".into(), ..tested("c8", "s3", 8, EXPERIMENT, Some(2), Some(2)) },
1949 ];
1950 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1951 let s = fold(&text, Summary::default());
1952 let t = Selection::of(&s, select::Policy::Learned, Some(0.2)).test.expect("the learned policy shows its test");
1953 let draws = |k: u32, draws: Vec<u64>, refused: u64| Draws { k, draws, refused };
1954 assert_eq!((t.ran, t.control, t.refused, t.sessions), (5, 1, 2, 2), "{t:?}");
1955 assert_eq!(t.by_k, vec![draws(1, vec![1, 0], 1), draws(2, vec![1, 3, 0], 1)]);
1956 assert_eq!((t.risk, t.set), (0.2, true), "the collective's risk");
1957 assert!(!t.due && t.dilution.is_empty() && t.ask_free.is_none(), "5 checks, 2 sessions: not due, nothing estimated");
1958 assert_eq!((t.due_checks, t.due_sessions), (select::replay::TEST_CHECKS, u64::from(select::replay::TEST_SESSIONS)));
1959 let line = selection_line(&Selection::of(&s, select::Policy::Learned, Some(0.2)));
1960 assert!(
1961 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 · "),
1962 "{line}"
1963 );
1964 assert_eq!(t.kept, t.ran, "every drawn check still held: {t:?}");
1965 let mut compacted = s.clone();
1967 compacted.checks.remove(0);
1968 let c = Selection::of(&compacted, select::Policy::Learned, Some(0.2)).test.unwrap();
1969 assert_eq!((c.ran, c.kept), (5, 4), "{c:?}");
1970 assert!(
1971 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"),
1972 "{}",
1973 test_line(&c)
1974 );
1975 let default = Selection::of(&s, select::Policy::Learned, None).test.unwrap();
1976 assert_eq!((default.risk, default.set), (ALPHA, false), "unset: the policy's ALPHA");
1977 let cold = Selection::of(&Summary::default(), select::Policy::Learned, Some(0.0));
1979 assert_eq!(cold.test.as_ref().map(|t| (t.ran, t.risk)), Some((0, 0.0)));
1980 assert_eq!(test_line(cold.test.as_ref().unwrap()), "question-count test off (risk 0)");
1981 let none = Selection::of(&Summary::default(), select::Policy::Learned, None).test.unwrap();
1982 assert_eq!(test_line(&none), "question-count test: no check yet, risk 0.10");
1983 assert!(Selection::of(&Summary::default(), select::Policy::Fixed, None).test.is_none());
1985 assert_eq!(Selection::of(&s, select::Policy::Fixed, None).test.map(|t| t.ran), Some(5), "an earlier opt-in's checks");
1986 }
1987
1988 #[test]
1992 fn the_gate_shown_is_the_one_in_force_at_the_collectives_risk() {
1993 use select::summary::{Gate, Summary};
1994 let s = Summary { gate: Gate { gain: vec![0.0; select::policy::DRAWS], allowance: 0.3, ..Gate::default() }, ..Summary::default() };
1995 let gate = |risk: Option<f64>| Selection::of(&s, select::Policy::Learned, risk).gate;
1996 assert!((gate(None) - 0.3).abs() < 1e-12, "{}", gate(None));
1997 assert!((gate(Some(0.2)) - 0.6).abs() < 1e-12, "twice ALPHA: twice the allowance, {}", gate(Some(0.2)));
1998 assert_eq!(gate(Some(0.0)), 0.0, "risk 0");
1999 }
2000
2001 #[test]
2004 fn the_test_readout_shows_the_replays_estimates_once_due() {
2005 use crate::outcome::{self, Outcome};
2006 use select::policy::EXPERIMENT;
2007 use select::summary::{Summary, fold};
2008 let sessions = u64::from(select::replay::TEST_SESSIONS);
2009 let mut text = String::new();
2010 let mut line = |o: &Outcome| text.push_str(&format!("{}\n", serde_json::to_string(o).unwrap()));
2011 for i in 0..sessions {
2012 let (session, check, ms) = (format!("s{i}"), format!("{i:012x}"), 10 * i as i64);
2013 let asked = i % 2 == 0;
2015 let g = Outcome {
2016 lessons: vec!["A".into()],
2017 grades: vec![2],
2018 ranks: vec![0],
2019 applied: 1,
2020 chosen: if asked { vec!["A".into()] } else { Vec::new() },
2021 free: if asked { Vec::new() } else { vec!["A".into()] },
2022 p: vec![0.5],
2023 pf: vec![0.5],
2024 ..tested(&check, &session, ms, EXPERIMENT, Some(1), Some(u32::from(asked)))
2025 };
2026 line(&g);
2027 let rec = |kind: &str, ms: i64| Outcome {
2028 kind: kind.into(),
2029 session: session.clone(),
2030 lessons: vec!["A".into()],
2031 check: check.clone(),
2032 ms,
2033 ..Default::default()
2034 };
2035 let route = if asked { outcome::ROUTE_ASKED } else { outcome::ROUTE_FREE };
2036 line(&Outcome { route: route.into(), ..rec(outcome::DELIVERY, ms + 1) });
2037 let verdict = if asked { outcome::CHANGED } else { outcome::IGNORED };
2039 line(&Outcome { judge: 2, channel: if asked { "live" } else { "queued" }.into(), ..rec(verdict, ms + 2) });
2040 }
2041 let s = fold(&text, Summary::default());
2042 let t = Selection::of(&s, select::Policy::Learned, None).test.unwrap();
2043 assert!(t.due && t.sessions == sessions, "{t:?}");
2044 let r = select::replay::test_readout(&s);
2045 let want = r.ask_free.as_ref().and_then(|x| x.value).expect("judged 2s on both routes");
2046 let e = t.ask_free.as_ref().expect("estimated once due");
2047 assert_eq!((e.mean, e.unit.as_str()), (100.0 * want, "points changed"));
2048 assert!(e.mean > 0.0, "asking changed the work, handing over did not: {e:?}");
2049 assert!(t.dilution.is_empty() && r.by_asked.is_empty(), "{t:?}");
2051 let line = selection_line(&Selection::of(&s, select::Policy::Learned, None));
2052 assert!(line.contains(&format!("; readout: grade 2 asked − handed over {:+.1} [", e.mean)), "{line}");
2053 }
2054
2055 #[test]
2058 fn unread_counts_follow_the_shown_once_claims() {
2059 let home = std::env::temp_dir().join(format!("borg-top-inbox-{}", std::process::id()));
2060 let _ = std::fs::remove_dir_all(&home);
2061 std::fs::create_dir_all(&home).unwrap();
2062 let ds = fixture();
2063 assert!(inbox(&home, &ds, NOW).recent.is_empty());
2064 let here = msg::node(&home);
2065 msg::send(&home, message_on(&here, "cccc3333-x", "Three of Three", "aaaa1111-x", msg::TELL, NOW - 3_000, "look at recon"), NOW).unwrap();
2066 msg::send(&home, message_on(&here, "cccc3333-x", "Three of Three", msg::ALL, msg::TELL, NOW - 2_000, "lunch"), NOW).unwrap();
2067 let i = inbox(&home, &ds, NOW);
2068 assert_eq!(i.unread.get("aaaa1111-x"), Some(&2), "{i:?}");
2069 assert_eq!(i.unread.get("cccc3333-x"), Some(&0), "its own broadcast");
2070 assert!(!i.unread.contains_key(&ds[1].p.session), "remote: claimed on its node");
2071 assert_eq!((i.day, i.recent[0].text.as_str()), (2, "lunch"));
2072 let caps = crate::harness::by_name("claude").expect("a harness").caps();
2073 let ev = json!({"session_id": "aaaa1111-x"});
2074 assert!(msg::deliver(&home, caps, "prompt", &ev, Some(100), NOW, None).is_some());
2075 assert_eq!(inbox(&home, &ds, NOW).unread.get("aaaa1111-x"), Some(&0));
2076 let _ = std::fs::remove_dir_all(&home);
2077 }
2078
2079 fn seen_lines(home: &Path) -> Vec<(String, Vec<String>)> {
2081 msg::lines(home)
2082 .into_iter()
2083 .filter_map(|l| match l {
2084 msg::Line::Seen(s) => Some((s.by, s.ids)),
2085 msg::Line::Message(_) => None,
2086 })
2087 .collect()
2088 }
2089
2090 fn shell(sessions: Vec<(String, &'static str)>, pids: Vec<(u32, &'static str)>) -> impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String> {
2093 move |ds| msg::pick(sessions.clone(), &pids, ds)
2094 }
2095
2096 fn acked_fixture(base: &Path) -> (std::path::PathBuf, String, Message) {
2099 let _ = std::fs::remove_dir_all(base);
2100 let (home, work) = (base.join("store"), base.join("work"));
2101 std::fs::create_dir_all(&home).unwrap();
2102 std::fs::create_dir_all(&work).unwrap();
2103 let git = |a: &[&str]| std::process::Command::new("git").arg("-C").arg(&work).args(a).output().is_ok_and(|o| o.status.success());
2104 assert!(git(&["init", "-q"]) && git(&["config", "user.email", "ann@x"]));
2105 let here = msg::node(&home);
2106 let to_me = message_on(&here, "aaaa1111-x", "One of Three", msg::ENGINEER, msg::TELL, NOW - 3_000, "the build is green");
2107 let to_drone = message_on(&here, "cccc3333-x", "Three of Three", "aaaa1111-x", msg::TELL, NOW - 2_000, "look at recon");
2108 msg::send(&home, to_me.clone(), NOW).unwrap();
2109 msg::send(&home, to_drone, NOW).unwrap();
2110 (home, work.display().to_string(), to_me)
2111 }
2112
2113 #[test]
2117 fn the_screen_acknowledges_the_messages_to_the_engineer_it_prints() {
2118 let base = std::env::temp_dir().join(format!("borg-top-ack-{}", std::process::id()));
2119 let (home, cwd, to_me) = acked_fixture(&base);
2120 let routes = [Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local }];
2121 let terminal = || shell(Vec::new(), Vec::new());
2122 let json = screen_of(&routes, true, &cwd, terminal(), true, NOW);
2123 assert!(json.contains("the build is green") && seen_lines(&home).is_empty(), "--json acknowledges nothing: {json}");
2124 let text = screen_of(&routes, true, &cwd, terminal(), false, NOW);
2125 assert!(text.contains("the build is green") && text.contains("look at recon"), "{text}");
2126 assert_eq!(seen_lines(&home), [("engineer:ann@x".to_string(), vec![to_me.id.clone()])], "the engineer's message only, as theirs");
2127 screen_of(&routes, true, &cwd, terminal(), false, NOW + 1_000);
2128 assert_eq!(seen_lines(&home).len(), 1, "acknowledged once: a later look writes no second line");
2129 let _ = std::fs::remove_dir_all(&base);
2130 }
2131
2132 #[test]
2137 fn a_session_running_the_screen_acknowledges_nothing_as_the_engineer() {
2138 let base = std::env::temp_dir().join(format!("borg-top-ack-session-{}", std::process::id()));
2139 let (home, cwd, to_me) = acked_fixture(&base);
2140 let other = base.join("other");
2141 std::fs::create_dir_all(&other).unwrap();
2142 let pid = std::process::id();
2143 let oc = Presence { harness: "opencode".into(), pid: Some(pid), ..p("dddd4444-x", "ann@x", NOW - 9_000, NOW - 1_000, "prompt", "docs", &[]) };
2144 presence::write(&other, &oc).unwrap();
2145 let routes = [
2146 Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local },
2147 Route::Collective { name: "side".into(), home: other.clone(), share: Share::Local },
2148 ];
2149 let shells = [
2150 ("Claude Code", shell(vec![("eeee5555-x".into(), "claude")], Vec::new())),
2151 ("Codex in Claude Code, neither with presence", shell(vec![("eeee5555-x".into(), "claude"), ("ffff6666-x".into(), "codex")], Vec::new())),
2152 ("opencode, its drone in the other collective", shell(Vec::new(), vec![(pid, "opencode")])),
2153 ];
2154 for (what, caller) in shells {
2155 let text = screen_of(&routes, false, &cwd, caller, false, NOW);
2156 assert!(text.contains("the build is green"), "{what}: shown {text}");
2157 assert!(seen_lines(&home).is_empty() && seen_lines(&other).is_empty(), "{what}: acknowledged nothing as the engineer");
2158 }
2159 screen_of(&routes, false, &cwd, shell(Vec::new(), vec![(pid + 1, "opencode")]), false, NOW);
2160 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");
2161 let _ = std::fs::remove_dir_all(&base);
2162 }
2163
2164 #[test]
2169 fn the_printed_screen_acknowledges_only_the_messages_whose_text_it_prints() {
2170 let base = std::env::temp_dir().join(format!("borg-top-ack-cut-{}", std::process::id()));
2171 let (home, cwd, to_me) = acked_fixture(&base);
2172 let email = format!("{}@x", "a".repeat(58));
2174 assert!(std::process::Command::new("git").args(["-C", &cwd, "config", "user.email", &email]).output().is_ok_and(|o| o.status.success()));
2175 let (here, to) = (msg::node(&home), format!("{}:{email}", msg::ENGINEER));
2176 let one = message_on(&here, "bbbb2222-x", &"n".repeat(52), &to, msg::TELL, NOW - 1_500, "one character shows");
2177 let none = message_on(&here, "dddd4444-x", &"n".repeat(53), &to, msg::TELL, NOW - 1_000, "none shows");
2178 msg::send(&home, one.clone(), NOW).unwrap();
2179 msg::send(&home, none.clone(), NOW).unwrap();
2180 let routes = [Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local }];
2181 let text = screen_of(&routes, true, &cwd, shell(Vec::new(), Vec::new()), false, NOW);
2182 assert!(text.contains(&format!("{} -> {email}: o…", one.name)) && text.contains(&format!("{} -> {email}: …", none.name)), "{text}");
2183 let mut acked: Vec<String> = seen_lines(&home).into_iter().flat_map(|(by, ids)| ids.into_iter().map(move |id| format!("{by} {id}"))).collect();
2184 acked.sort();
2185 let mut want = [format!("{to} {}", to_me.id), format!("{to} {}", one.id)];
2186 want.sort();
2187 assert_eq!(acked, want, "the line that shows one character of its text is read; the one cut before it is not");
2188 let _ = std::fs::remove_dir_all(&base);
2189 }
2190
2191 #[test]
2195 fn a_live_screen_acknowledges_each_message_once() {
2196 let base = std::env::temp_dir().join(format!("borg-top-ack-once-{}", std::process::id()));
2197 let _ = std::fs::remove_dir_all(&base);
2198 std::fs::create_dir_all(&base).unwrap();
2199 let home = base.clone();
2200 let here = msg::node(&home);
2201 let to_me = |by: &str, ms: i64, text: &str| message_on(&here, by, "a drone", msg::ENGINEER, msg::TELL, ms, text);
2202 let first = to_me("aaaa1111-x", NOW - 3_000, "the build is green");
2203 msg::send(&home, first.clone(), NOW).unwrap();
2204 let cwd = base.display().to_string();
2205 let mut done = HashSet::new();
2206 let recent = |now: i64| inbox(&home, &[], now).recent;
2207 acknowledge(&home, &cwd, &recent(NOW).iter().collect::<Vec<_>>(), &mut done, NOW);
2208 assert_eq!(seen_lines(&home).len(), 1, "{:?}", seen_lines(&home));
2209 assert!(done.contains(&first.id));
2210 let kept: String = std::fs::read_to_string(msg::path(&home)).unwrap().lines().filter(|l| !l.contains("\"seen\"")).map(|l| format!("{l}\n")).collect();
2213 std::fs::write(msg::path(&home), kept).unwrap();
2214 acknowledge(&home, &cwd, &recent(NOW + 1_000).iter().collect::<Vec<_>>(), &mut done, NOW + 1_000);
2215 assert!(seen_lines(&home).is_empty(), "skipped: no second ack of a done id");
2216 let second = to_me("cccc3333-x", NOW + 2_000, "and the docs");
2218 msg::send(&home, second.clone(), NOW + 2_000).unwrap();
2219 acknowledge(&home, &cwd, &recent(NOW + 3_000).iter().collect::<Vec<_>>(), &mut done, NOW + 3_000);
2220 assert_eq!(seen_lines(&home).into_iter().map(|(_, ids)| ids).collect::<Vec<_>>(), [vec![second.id.clone()]], "only the new one");
2221 let _ = std::fs::remove_dir_all(&base);
2222 }
2223
2224 #[test]
2227 fn a_sender_is_tied_to_the_node_its_line_names() {
2228 let ds = fixture();
2229 let own = message("aaaa1111-x", "anything", msg::ALL, msg::TELL, NOW, "hi");
2230 assert_eq!(sender(&ds, "n1", &own), "One of Three", "this node's drone");
2231 let forged = message_on("evil0123456789", "aaaa1111-x", "One of Three", msg::ALL, msg::TELL, NOW, "hi");
2232 assert_eq!(sender(&ds, "n1", &forged), "One of Three @evil01234567", "the claim, with the node it came from");
2233 let remote = message_on(&"beefcafe".repeat(8), &ds[1].p.session, "x", msg::ALL, msg::TELL, NOW, "hi");
2234 assert_eq!(sender(&ds, "n1", &remote), "Two of Three @beefcafebeef", "the remote drone on its own node");
2235 assert_eq!(recipient(&ds, "n1", "aaaa1111-x@n1"), "One of Three");
2236 assert_eq!(recipient(&ds, "n1", "aaaa1111-x@evil"), "aaaa1111", "not the local drone of that id");
2237 assert_eq!((recipient(&ds, "n1", "engineer"), recipient(&ds, "n1", msg::ALL)), ("engineer".into(), "collective".into()), "an `all` line");
2238 assert_eq!(recipient(&ds, "n1", "local@n1"), "local");
2239 assert_eq!(recipient(&ds, "n1", "local@beefcafebeefcafe00"), "local @beefcafebeef");
2240 }
2241
2242 #[test]
2246 fn the_line_on_borg_itself_names_a_release_and_the_protocols_that_differ() {
2247 let home = std::env::temp_dir().join(format!("borg-top-version-{}", std::process::id()));
2248 let _ = std::fs::remove_dir_all(&home);
2249 std::fs::create_dir_all(&home).unwrap();
2250 let newer = update::Checked { checked_ms: 1, latest: Some("99.0.0".into()), error: None };
2251 let current = update::Checked { latest: Some(update::VERSION.into()), ..newer.clone() };
2252 assert_eq!(version_line([home.as_path()], None, true), None);
2253 assert_eq!(version_line([home.as_path()], Some(¤t), true), None);
2254 let release = format!("borg 99.0.0 is released, this is {}: borg update", update::VERSION);
2255 assert_eq!(version_line([home.as_path()], Some(&newer), true), Some(release.clone()));
2256 assert_eq!(version_line([home.as_path()], Some(&newer), false), None, "BORG_UPDATE_CHECK=off");
2257
2258 let ours = protocol::current();
2259 let v = |protocol, engineer: Option<&str>| protocol::PeerVersion { borg: None, protocol, ms: 1, engineer: engineer.map(str::to_string) };
2260 protocol::note_version(&home, "k-same", &v(ours, Some("ann@x"))).unwrap();
2261 assert_eq!(version_line([home.as_path()], None, true), None, "a member on this node's protocol is no news");
2262 protocol::note_version(&home, "k-old", &v(ours - 1, Some("bob@x"))).unwrap();
2263 let mismatch = protocol::Mismatch { theirs: format!("borg/mesh/{}", ours + 1), ours: format!("borg/mesh/{ours}"), ms: 2 };
2264 protocol::note_mismatch(&home, "k-new", mismatch).unwrap();
2265 let line = version_line([home.as_path()], Some(&newer), true).unwrap();
2266 let old = format!("bob@x on /{} misses /{ours}: {}", ours - 1, protocol::note(ours).unwrap());
2267 let none = format!("nothing syncs with peer k-new (borg/mesh/{}, this node borg/mesh/{ours})", ours + 1);
2268 assert_eq!(line, format!("{release} · {none} · {old}"));
2269 let _ = std::fs::remove_dir_all(&home);
2270 }
2271
2272 #[test]
2273 fn json_for_scripts() {
2274 let v = to_json_with("corp", Share::Team, &fixture(), &inbox_fixture());
2275 assert_eq!((v["drones"][0]["unread"].as_u64(), v["drones"][1].get("unread")), (Some(2), None));
2276 assert_eq!((v["messages"]["day"].as_u64(), v["messages"]["recent"][0]["kind"].as_str()), (Some(3), Some("ask")));
2277 assert_eq!(v["collective"], "corp");
2278 let d = &v["drones"];
2279 assert_eq!(d.as_array().unwrap().len(), 3);
2280 assert_eq!(d[0]["designation"], "One of Three");
2281 assert_eq!(d[0]["session"], "aaaa1111-x");
2282 assert_eq!(d[0]["state"], "working");
2283 assert_eq!(d[0]["paused"], true);
2284 assert_eq!(d[1]["node"], "beefcafe".repeat(8));
2285 assert_eq!(d[1]["state"], "waiting");
2286 assert_eq!(d[2]["state"], "idle");
2287 assert_eq!(d[2]["files"], json!([]));
2288 }
2289
2290 #[test]
2292 fn a_name_matching_here_and_on_a_member_names_the_drone_here() {
2293 let local = vec![p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "post_tool", "a", &[])];
2294 let remote = ["aaaa1111-x", "aaaa2222-z"].map(|s| presence::Remote {
2295 p: p(s, "bob@y", 50, NOW - 5_000, "post_tool", "b", &[]),
2296 key: "beefcafe".repeat(8),
2297 sig: String::new(),
2298 });
2299 let ds = assemble(local, remote.into(), NOW, |_| true, |_| false);
2300 for t in ["aaaa", "aaaa1111", "aaaa1111-x"] {
2301 let d = resolve(&ds, t).unwrap().unwrap();
2302 assert!(d.node.is_empty() && d.p.session == "aaaa1111-x", "{t}");
2303 }
2304 assert_eq!(resolve(&ds, "aaaa2").unwrap().unwrap().node, "beefcafe".repeat(8), "only a member's matches");
2305 assert!(!resolve(&ds, "One").unwrap().unwrap().node.is_empty(), "a number names one drone of all, wherever it runs");
2306 }
2307
2308 #[test]
2310 fn a_name_matching_an_ended_drone_here_and_a_live_member_names_the_live_one() {
2311 let local = vec![p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "end", "a", &[])];
2312 let remote =
2313 vec![presence::Remote { p: p("aaaa2222-z", "bob@y", 50, NOW - 5_000, "post_tool", "b", &[]), key: "beefcafe".repeat(8), sig: String::new() }];
2314 let ds = assemble(local, remote, NOW, |_| true, |_| false);
2315 let d = resolve(&ds, "aaaa").unwrap().unwrap();
2316 assert_eq!((d.p.session.as_str(), d.node.as_str()), ("aaaa2222-z", "beefcafe".repeat(8).as_str()), "the live one, though remote");
2317 assert!(resolve(&ds, "aaaa1111").unwrap().unwrap().node.is_empty(), "an ended drone is still named by its own id");
2318 }
2319
2320 #[test]
2321 fn resolves_drones_by_designation_or_session() {
2322 let ds = fixture();
2323 assert_eq!(resolve(&ds, "Two").unwrap().unwrap().p.session, "bbbb2222-y");
2324 assert_eq!(resolve(&ds, "two of three").unwrap().unwrap().p.session, "bbbb2222-y");
2325 assert_eq!(resolve(&ds, "3").unwrap().unwrap().p.session, "cccc3333-x");
2326 assert_eq!(resolve(&ds, "aaaa").unwrap().unwrap().p.session, "aaaa1111-x");
2327 assert!(resolve(&ds, "all").unwrap().is_none());
2328 assert!(resolve(&ds, "Four").is_err());
2329 assert!(resolve(&ds, "zzzz").is_err());
2330 assert!(resolve(&ds, "aa").is_err(), "too short a prefix");
2331 assert!(["collective", "local"].iter().all(|w| resolve(&ds, w).unwrap().is_none()));
2332 assert_eq!(resolve(&ds, "bbbb").unwrap().unwrap().p.session, "bbbb2222-y", "nothing here matches: the member's");
2333 assert_eq!(word(21), "21");
2334 }
2335}