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>,
455 pub declaration: Option<Estimate>,
457 pub delivery: Option<select::policy::Rates>,
460 pub evaluation: Option<Evaluation>,
462 pub measurement: Option<Measurement>,
465 pub pool: Vec<Yield>,
467 pub latency: Vec<Latency>,
469 pub test: Option<Test>,
472}
473
474#[derive(Debug, Clone, Default, PartialEq, Serialize)]
479pub struct Test {
480 pub risk: f64,
483 pub set: bool,
485 pub ran: u64,
488 pub control: u64,
489 pub refused: u64,
492 pub sessions: u64,
494 pub kept: u64,
497 pub by_k: Vec<Draws>,
499 pub due: bool,
506 pub due_checks: u64,
507 pub due_sessions: u64,
508 pub due_min_sessions: u64,
509 pub dilution: Vec<Estimate>,
510 pub ask_free: Option<Estimate>,
511}
512
513#[derive(Debug, Clone, Default, PartialEq, Serialize)]
515pub struct Draws {
516 pub k: u32,
517 pub draws: Vec<u64>,
518 pub refused: u64,
519}
520
521impl Test {
522 #[must_use]
525 pub fn of(summary: &select::summary::Summary, risk: Option<f64>) -> Test {
526 use select::replay::{TEST_CHECKS, TEST_MIN_SESSIONS, TEST_SESSIONS, test_readout};
527 let e = &summary.experiment;
528 let (ran, refused) = e.totals();
529 let r = test_readout(summary);
530 let estimate = |name: String, unit: &str, x: &crate::ope::Estimate| {
532 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 })
533 };
534 let mut by_k: Vec<Draws> = Vec::new();
535 let at = |by_k: &mut Vec<Draws>, k: u32| match by_k.iter().position(|d| d.k == k) {
536 Some(i) => i,
537 None => {
538 by_k.push(Draws { k, draws: vec![0; k as usize + 1], refused: 0 });
539 by_k.len() - 1
540 }
541 };
542 for &(k, s, n) in &e.ran {
543 let i = at(&mut by_k, k);
544 if let Some(c) = by_k[i].draws.get_mut(s as usize) {
545 *c += n;
546 }
547 }
548 for &(k, n) in &e.refused {
549 let i = at(&mut by_k, k);
550 by_k[i].refused += n;
551 }
552 by_k.sort_by_key(|d| d.k);
553 Test {
554 risk: risk.unwrap_or(select::policy::ALPHA),
555 set: risk.is_some(),
556 ran,
557 control: e.control,
558 refused,
559 sessions: u64::from(e.sessions),
560 kept: r.kept,
561 by_k,
562 due: r.due,
563 due_checks: TEST_CHECKS,
564 due_sessions: u64::from(TEST_SESSIONS),
565 due_min_sessions: u64::from(TEST_MIN_SESSIONS),
566 dilution: r.by_asked.iter().filter_map(|(m, x)| estimate(format!("{m} asked − 1 asked"), "points changed", x)).collect(),
567 ask_free: r.ask_free.as_ref().and_then(|x| estimate("grade 2 asked − handed over".into(), "points changed", x)),
568 }
569 }
570}
571
572#[derive(Debug, Clone, Default, PartialEq, Serialize)]
574pub struct Count {
575 pub name: String,
576 pub part: u64,
577 pub whole: u64,
578}
579
580#[derive(Debug, Clone, Default, PartialEq, Serialize)]
582pub struct Spend {
583 pub source: String,
585 pub calls: u64,
586 pub rounds: u64,
588 pub usd: f64,
589}
590
591#[derive(Debug, Clone, Default, PartialEq, Serialize)]
595pub struct Budget {
596 pub tokens: f64,
597 pub spent: f64,
598 pub price: f64,
599}
600
601#[derive(Debug, Clone, Default, PartialEq, Serialize)]
603pub struct Estimate {
604 pub name: String,
605 pub unit: String,
610 pub mean: f64,
611 pub lo: f64,
612 pub hi: f64,
613 pub n: u64,
615}
616
617#[derive(Debug, Clone, Default, PartialEq, Serialize)]
620pub struct Evaluation {
621 pub sessions: u64,
623 pub mean: f64,
626 pub ucb: Option<f64>,
627 pub units: u64,
629 pub ess: f64,
630 pub max_weight: f64,
632 pub truncated: u64,
634}
635
636#[derive(Debug, Clone, Default, PartialEq, Serialize)]
641pub struct Measurement {
642 pub placebos: u64,
644 pub judged: u64,
646 pub tail_shown: u64,
649 pub tail_applied: u64,
650}
651
652#[derive(Debug, Clone, Default, PartialEq, Serialize)]
655pub struct Yield {
656 pub from: u32,
657 pub to: u32,
658 pub shown: u64,
659 pub applied: u64,
660 pub rate: f64,
661}
662
663#[derive(Debug, Clone, Default, PartialEq, Serialize)]
666pub struct Latency {
667 pub harness: String,
668 pub checks: u64,
669 pub p50_ms: u64,
671 pub p95_ms: u64,
672 pub budget_ms: Option<u64>,
674 pub over: u64,
677}
678
679const QUANTILES: [f64; 2] = [0.50, 0.95];
682
683fn quantile(sorted: &[u64], q: f64) -> u64 {
685 let rank = ((q * sorted.len() as f64).ceil() as usize).clamp(1, sorted.len().max(1));
686 sorted.get(rank - 1).copied().unwrap_or(0)
687}
688
689impl Selection {
690 #[must_use]
694 pub fn of(summary: &select::summary::Summary, policy: select::Policy, risk: Option<f64>) -> Selection {
695 use select::summary::{LIVE, Label, STOP};
696 let s = summary;
697 let learned = policy == select::Policy::Learned;
698 let reverted = learned && s.monitor.reverted;
699 let default = select::precedence(None, None, false).name();
700 let count = |name: &str, part: usize, whole: usize| Count { name: name.into(), part: part as u64, whole: whole as u64 };
701 let (sessions, checks) = s.checked();
703 let deviations = match s.gate.checks {
704 0 => Vec::new(),
705 n => vec![count("route", s.gate.deviated as usize, n as usize)],
706 };
707 let fallbacks = select::summary::FELL_BACK
709 .iter()
710 .zip(s.fell_back)
711 .filter(|_| checks > 0)
712 .map(|(reason, n)| Count { name: (*reason).into(), part: n, whole: checks })
713 .collect();
714 let spend = [("stop", STOP), ("live", LIVE)]
715 .into_iter()
716 .filter(|(_, i)| s.spend[*i].calls > 0)
717 .map(|(source, i)| Spend { source: source.into(), calls: s.spend[i].calls, rounds: s.spend[i].rounds, usd: s.spend[i].usd })
718 .collect();
719 let budget = s.fit.budget.map(|tokens| Budget { tokens, spent: s.pool.tokens as f64 / sessions.max(1) as f64, price: s.fit.lambda });
720 let judged = |l: Label| matches!(l, Label::Changed | Label::Ignored);
721 let of = |route: select::Route| {
727 let of: Vec<_> = s.applies.iter().filter(|a| a.route == route && (route != select::Route::Free || !a.had)).collect();
728 (of.iter().filter(|a| judged(a.label)).count(), of.len())
729 };
730 let mut labels = Vec::new();
731 for (name, route) in [("judged, asked", select::Route::Ask), ("judged, free", select::Route::Free), ("judged, placebo", select::Route::Placebo)] {
732 let (part, whole) = of(route);
733 labels.push(count(name, part, whole));
734 }
735 let verdicts = s.applies.iter().filter(|a| judged(a.label) || a.label == Label::Unclear).count();
736 labels.push(count("unclear", s.applies.iter().filter(|a| a.label == Label::Unclear).count(), verdicts));
737 let later = s.applies.iter().filter(|a| matches!(a.route, select::Route::Not | select::Route::Placebo)).count();
738 let (placebos_judged, placebos) = of(select::Route::Placebo);
739 labels.push(count("placebo", placebos, later));
740 labels.retain(|c| c.whole > 0);
741 let measurement = (checks > 0).then_some(Measurement {
742 placebos: placebos as u64,
743 judged: placebos_judged as u64,
744 tail_shown: s.pool.tail.0,
745 tail_applied: s.pool.tail.1,
746 });
747 let share = |stripped: bool| {
749 let of: Vec<_> = s.applies.iter().filter(|a| a.route == select::Route::Ask && judged(a.label) && a.stripped == stripped).collect();
750 (of.iter().filter(|a| a.label == Label::Changed).count() as f64, of.len() as f64)
751 };
752 let ((c1, n1), (c0, n0)) = (share(false), share(true));
753 let declaration = (n1 > 0.0 && n0 > 0.0).then(|| {
754 let (r1, r0) = (c1 / n1, c0 / n0);
755 let se = (r1 * (1.0 - r1) / n1 + r0 * (1.0 - r0) / n0).sqrt();
756 let mean = 100.0 * (r1 - r0);
757 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 }
758 });
759 let pool = s
760 .pool
761 .bins
762 .iter()
763 .enumerate()
764 .filter(|(_, (shown, _))| *shown > 0)
765 .map(|(b, &(shown, applied))| Yield {
766 from: (1u32 << b) - 1,
767 to: (1u32 << (b + 1)) - 1,
768 shown,
769 applied,
770 rate: s.fit.pool.yields.get(b).map_or(applied as f64 / shown as f64, |(a, b)| a / (a + b)),
771 })
772 .collect();
773 let (allowance, ledger) = if learned && !reverted { select::policy::gate_parts(&s.gate, risk.unwrap_or(select::policy::ALPHA)) } else { (0.0, 0.0) };
774 let mon = &s.monitor;
775 let evaluation = (mon.sessions > 0).then(|| Evaluation {
776 sessions: u64::from(mon.sessions),
777 mean: mon.mean,
778 ucb: mon.ucb,
779 units: u64::from(mon.labelled),
780 ess: mon.ess,
781 max_weight: mon.max_weight,
782 truncated: u64::from(mon.truncated),
783 });
784 let latency = s
785 .pool
786 .stops
787 .iter()
788 .filter(|(_, calls)| !calls.is_empty())
789 .map(|(harness, calls)| {
790 let mut ms: Vec<u64> = calls.iter().map(|&(ms, _)| u64::from(ms)).collect();
791 ms.sort_unstable();
792 let budget = crate::harness::ALL.iter().find(|h| h.name() == harness).and_then(|h| h.caps().stop_budget);
793 Latency {
794 harness: harness.clone(),
795 checks: ms.len() as u64,
796 p50_ms: quantile(&ms, QUANTILES[0]),
797 p95_ms: quantile(&ms, QUANTILES[1]),
798 budget_ms: budget.map(|b| u64::try_from(b.as_millis()).unwrap_or(u64::MAX)),
799 over: calls.iter().filter(|(_, failed)| *failed).count() as u64,
800 }
801 })
802 .collect();
803 Selection {
804 policy: policy.name(),
805 default,
806 source: if reverted { select::REVERTED_SOURCE } else { policy.name() },
807 gate: if learned && !reverted { select::policy::remaining_at(&s.gate, risk.unwrap_or(select::policy::ALPHA)) } else { 0.0 },
808 allowance,
809 ledger,
810 fit: s.fit.key.clone(),
811 at: s.at,
812 checks,
813 sessions,
814 deviations,
815 fallbacks,
816 spend,
817 budget,
818 effects: s.fit.model.as_ref().map(|m| effects(m, s.fit.labels)).unwrap_or_default(),
819 half_life: s.fit.model.as_ref().map(|_| s.fit.half_life.unwrap_or(0)),
820 labels,
821 declaration,
822 delivery: (!s.fit.key.is_empty()).then(|| s.fit.rates.clone()),
823 evaluation,
824 measurement,
825 pool,
826 latency,
827 test: (learned || s.experiment.totals() != (0, 0)).then(|| Test::of(s, risk)),
828 }
829 }
830
831 #[must_use]
834 pub fn shown(&self) -> bool {
835 self.checks > 0 || self.source != self.default
836 }
837}
838
839const Z95: f64 = 1.96;
841
842fn effects(m: &select::policy::Model, n: u32) -> Vec<Estimate> {
846 let p = select::policy::P;
847 if m.beta.len() != p || m.chol.len() != p * p {
848 return Vec::new();
849 }
850 let sd = |k: usize| m.chol[k * p..=k * p + k].iter().map(|x| x * x).sum::<f64>().sqrt();
852 let mut out: Vec<Estimate> = (0..p)
853 .map(|k| Estimate {
854 name: select::policy::NAMES[k].into(),
855 unit: "coefficient".into(),
856 mean: m.beta[k],
857 lo: m.beta[k] - Z95 * sd(k),
858 hi: m.beta[k] + Z95 * sd(k),
859 n: u64::from(n),
860 })
861 .collect();
862 for (name, v) in ["session spread", "lesson spread", "judge spread"].into_iter().zip(m.spread) {
863 out.push(Estimate { name: name.into(), unit: "logit".into(), mean: v, lo: v, hi: v, n: u64::from(n) });
864 }
865 out
866}
867
868#[must_use]
870pub fn selection(lessons: &Path, policy: select::Policy) -> Selection {
871 Selection::of(&select::summary::load(lessons), policy, select::risk_of(lessons))
872}
873
874#[must_use]
883pub fn selection_line(s: &Selection) -> String {
884 let default = s.source == s.default;
885 let head = match s.source {
886 select::REVERTED_SOURCE => "REVERTED to the fixed rule: the monitor saw harm beyond the gate's share (sticky)".to_string(),
887 "learned" => {
888 let opt = if default { "" } else { " (opt-in, BORG_SELECT=learned)" };
889 if s.gate > 0.0 {
890 format!("learned{opt}, gate {:.3} (quality it may still risk below the fixed rule)", s.gate)
891 } else if s.fit.is_empty() || s.allowance <= 0.0 {
892 format!("learned{opt}, cold start: gate 0, every decision the fixed rule's")
893 } else {
894 format!("learned{opt}, gate spent (ledger q05 {:+.3} against allowance {:.3}), deviations need positive evidence", s.ledger, s.allowance)
897 }
898 }
899 "fixed" if default => "the fixed rule (the default; BORG_SELECT=learned opts in to the learned policy)".into(),
900 "fixed" => "the fixed rule (BORG_SELECT=fixed)".into(),
901 "cap" => "the fixed rule with an ask cap (BORG_STOP_ASK_CAP, evaluation only)".into(),
902 "attention" => "attention's cut (BORG_STOP_ATTENTION=on, evaluation only)".into(),
903 other => other.to_string(),
904 };
905 if s.checks == 0 {
906 return format!("{head} · no graded check folded yet");
907 }
908 let mut parts = vec![head, format!("{} checks", s.checks)];
909 if !s.deviations.is_empty() {
910 parts.push(format!("deviated: {}", s.deviations.iter().map(|c| format!("{} {}/{}", c.name, c.part, c.whole)).collect::<Vec<_>>().join(", ")));
911 }
912 let fell: Vec<String> = s.fallbacks.iter().filter(|c| c.part > 0).map(|c| format!("{} {}", c.name, c.part)).collect();
913 if !fell.is_empty() {
914 parts.push(format!("learned decisions the fixed rule took over: {}", fell.join(", ")));
915 }
916 if let Some(t) = &s.test {
917 parts.push(test_line(t));
918 }
919 if let Some(m) = &s.measurement {
920 parts.push(format!("placebos {} drawn, {} judged", m.placebos, m.judged));
921 if m.tail_shown > 0 {
922 parts.push(format!("tail {} shown past the pool, {} applied", m.tail_shown, m.tail_applied));
923 }
924 }
925 if let Some(e) = &s.evaluation {
926 let reading = match e.ucb {
927 Some(u) => format!("monitor {:+.3}, bound {u:+.3} over {} sessions (below 0 reverts)", e.mean, e.sessions),
928 None => format!("monitor {:+.3} over {} sessions, no bound yet", e.mean, e.sessions),
929 };
930 parts.push(format!("{reading}, ESS {:.0} of {} weighed, largest weight {:.1}× the mean", e.ess, e.units, e.max_weight));
931 }
932 if !s.latency.is_empty() {
933 let secs = |ms: u64| format!("{:.1} s", ms as f64 / 1000.0);
934 let each: Vec<String> = s
935 .latency
936 .iter()
937 .map(|l| {
938 let budget = l.budget_ms.map_or_else(String::new, |b| format!(" of {}", secs(b)));
939 let failed = if l.over > 0 { format!(", {} failed", l.over) } else { String::new() };
940 format!("{} {}{budget}{failed}", l.harness, secs(l.p95_ms))
941 })
942 .collect();
943 parts.push(format!("stop p95 {}", each.join("; ")));
944 }
945 if s.sessions > 0 && !s.spend.is_empty() {
946 let n = s.sessions as f64;
947 let (usd, rounds) = s.spend.iter().fold((0.0, 0), |(u, r), x| (u + x.usd, r + x.rounds));
948 parts.push(format!("${:.2} and {:.1} extra rounds per session ({} sessions)", usd / n, rounds as f64 / n, s.sessions));
949 }
950 if let Some(b) = &s.budget {
951 parts.push(format!("budget {:.0} pool tokens per session: {:.0} read, price {:.2e} per token", b.tokens, b.spent, b.price));
952 }
953 parts.join(" · ")
954}
955
956fn test_line(t: &Test) -> String {
961 let risk = if t.set { format!("risk {:.2} (the collective's)", t.risk) } else { format!("risk {:.2}", t.risk) };
962 match (t.ran, t.risk <= 0.0) {
963 (0, true) => return "question-count test off (risk 0)".into(),
964 (0, false) => return format!("question-count test: no check yet, {risk}"),
965 _ => {}
966 }
967 let kept = if t.kept < t.ran { format!(", {} still in the summary", t.kept) } else { String::new() };
968 let head =
969 format!("question-count test {} drawn ({} control, {} refused by the gate{kept}) in {} sessions, {risk}", t.ran, t.control, t.refused, t.sessions);
970 match (&t.ask_free, t.due) {
971 (Some(e), true) => format!("{head}; readout: grade 2 asked − handed over {:+.1} [{:+.1}, {:+.1}] points changed", e.mean, e.lo, e.hi),
972 (None, true) => format!("{head}; readout due, no judged grade 2 on both routes yet"),
973 (_, false) => format!("{head}; readout after {} checks in {} sessions, or {} sessions", t.due_checks, t.due_min_sessions, t.due_sessions),
974 }
975}
976
977#[must_use]
979pub fn render_with(name: &str, share: Share, drones: &[Drone], inbox: &Inbox, select: Option<&Selection>, now: i64) -> String {
980 let mut s = render(name, share, drones, inbox, now);
981 if let Some(sel) = select.filter(|s| s.shown()) {
982 s.push_str(&format!(" selection: {}\n", selection_line(sel)));
983 }
984 s
985}
986
987pub fn version_line<'a>(stores: impl IntoIterator<Item = &'a Path>, checked: Option<&update::Checked>, check: bool) -> Option<String> {
993 let mut parts: Vec<String> = Vec::new();
994 if let Some(l) = checked.and_then(|c| c.latest.as_deref()).filter(|l| check && update::newer(l, update::VERSION)) {
995 parts.push(format!("borg {l} is released, this is {}: borg update", update::VERSION));
996 }
997 let aliases = collective::aliases();
998 let ours = protocol::current();
999 for home in stores.into_iter().filter(|h| h.join("mesh").is_dir()) {
1000 let versions = protocol::versions(home);
1001 let mismatches = protocol::mismatches(home);
1002 let name = |key: &str| aliases.display(&protocol::member(key, versions.get(key)));
1003 for (key, m) in &mismatches {
1004 parts.push(format!("nothing syncs with {} ({}, this node {})", name(key), m.theirs, m.ours));
1005 }
1006 for (key, v) in versions.iter().filter(|(k, v)| v.protocol < ours && !mismatches.contains_key(*k)) {
1007 parts.push(format!("{} on /{} misses {}", name(key), v.protocol, protocol::misses(v.protocol, ours)));
1008 }
1009 }
1010 parts.dedup();
1011 (!parts.is_empty()).then(|| parts.join(" · "))
1012}
1013
1014fn routes(dir: Option<&str>) -> Result<Vec<Route>, String> {
1016 let cfg = collective::load()?;
1017 let base = learn::home();
1018 Ok(match dir {
1019 Some(d) => vec![collective::route(cfg.as_ref(), d, &base)].into_iter().filter(|r| *r != Route::Off).collect(),
1020 None => install::collectives(cfg.as_ref(), &base),
1021 })
1022}
1023
1024pub fn gather(dir: Option<&str>, now: i64) -> Result<(Vec<crate::top_live::Coll>, Option<String>), String> {
1027 let mut out = Vec::new();
1028 let routes = routes(dir)?;
1029 let stores: Vec<&Path> = routes
1030 .iter()
1031 .filter_map(|r| match r {
1032 Route::Collective { home, .. } => Some(home.as_path()),
1033 Route::Off => None,
1034 })
1035 .collect();
1036 let version = version_line(stores, update::checked().as_ref(), update::on());
1037 for r in &routes {
1038 let Route::Collective { name, home, share } = r else { continue };
1039 presence::prune(home, now);
1040 let mut ds = drones(home, collective::may_sync(r), now);
1041 for d in &mut ds {
1042 d.p.engineer = collective::aliases().display(&d.p.engineer);
1043 }
1044 let inbox = inbox(home, &ds, now);
1045 if ds.is_empty() && inbox.recent.is_empty() && dir.is_none() {
1046 continue;
1047 }
1048 let lessons = home.join("lessons.jsonl");
1051 let summary = select::summary::peek(&lessons);
1052 let select = Some(Selection::of(&summary, select::policy(), select::risk_of(&lessons))).filter(Selection::shown).map(|s| selection_line(&s));
1053 out.push(crate::top_live::Coll { name: name.clone(), home: home.clone(), share: *share, drones: ds, inbox, select });
1054 }
1055 Ok((out, version))
1056}
1057
1058fn screen(dir: Option<&str>, json_out: bool, now: i64) -> Result<String, String> {
1060 let routes = routes(dir)?;
1061 let text = screen_of(&routes, dir.is_some(), &engineer_dir(dir), msg::caller, json_out, now);
1062 let stores = routes.iter().filter_map(|r| match r {
1063 Route::Collective { home, .. } => Some(home.as_path()),
1064 Route::Off => None,
1065 });
1066 Ok(match version_line(stores, update::checked().as_ref(), update::on()).filter(|_| !json_out) {
1067 Some(v) => format!("{v}\n\n{text}"),
1068 None => text,
1069 })
1070}
1071
1072fn screen_of(
1077 routes: &[Route],
1078 one: bool,
1079 cwd: &str,
1080 caller: impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String>,
1081 json_out: bool,
1082 now: i64,
1083) -> String {
1084 let mut text = String::new();
1085 let mut all = Vec::new();
1086 let (mut gathered, mut printed) = (Vec::new(), Vec::new());
1089 for r in routes {
1090 let Route::Collective { name, home, share } = r else { continue };
1091 presence::prune(home, now);
1092 let mut ds = drones(home, collective::may_sync(r), now);
1093 for d in &mut ds {
1095 d.p.engineer = collective::aliases().display(&d.p.engineer);
1096 }
1097 let inbox = inbox(home, &ds, now);
1098 let sel = selection(&home.join("lessons.jsonl"), select::policy());
1099 if json_out {
1100 let mut v = to_json_with(name, *share, &ds, &inbox);
1101 v["select"] = json!(sel);
1102 all.push(v);
1103 } else if !ds.is_empty() || !inbox.recent.is_empty() || one {
1104 text.push_str(&render_with(name, *share, &ds, &inbox, Some(&sel), now));
1105 text.push('\n');
1106 printed.push((home, inbox.recent.into_iter().filter(|m| prints_text(&ds, &inbox.here, m)).collect::<Vec<_>>()));
1107 }
1108 gathered.push(ds);
1109 }
1110 if !printed.is_empty() && engineer_looks(caller, gathered.iter().map(Vec::as_slice)) {
1111 let mut done = HashSet::new();
1112 for (home, recent) in &printed {
1113 acknowledge(home, cwd, &recent.iter().collect::<Vec<_>>(), &mut done, now);
1114 }
1115 }
1116 if json_out {
1117 return serde_json::to_string_pretty(&all).unwrap_or_default() + "\n";
1118 }
1119 if text.is_empty() {
1120 text = "no drones (borg top lists sessions whose hooks ran in the last day)\n".into();
1121 }
1122 text
1123}
1124
1125pub fn main(args: &[&str]) -> i32 {
1126 let (mut dir, mut once, mut json_out) = (None, false, false);
1127 let mut it = args.iter();
1128 while let Some(a) = it.next() {
1129 match *a {
1130 "--dir" if dir.is_none() => match it.next() {
1131 Some(d) => dir = Some(*d),
1132 None => return usage("--dir needs a value"),
1133 },
1134 "--once" => once = true,
1135 "--json" => json_out = true,
1136 a => return usage(&format!("unknown flag {a}")),
1137 }
1138 }
1139 if !json_out && !once && unsafe { libc::isatty(1) } == 1 {
1141 return crate::top_live::run(dir);
1142 }
1143 match screen(dir, json_out, learn::now_ms()) {
1144 Ok(s) => {
1145 print!("{s}");
1146 0
1147 }
1148 Err(e) => {
1149 eprintln!("borg top: {e}");
1150 1
1151 }
1152 }
1153}
1154
1155fn usage(why: &str) -> i32 {
1156 eprintln!("borg top: {why}\nusage: borg top [--once] [--json] [--dir DIR]");
1157 2
1158}
1159
1160#[cfg(test)]
1161pub(crate) fn tests_fixture() -> Vec<Drone> {
1162 tests::fixture()
1163}
1164
1165#[cfg(test)]
1166pub(crate) fn tests_inbox() -> Inbox {
1167 tests::inbox_fixture()
1168}
1169
1170#[cfg(test)]
1171mod tests {
1172 use super::*;
1173
1174 fn p(session: &str, engineer: &str, started: i64, ms: i64, event: &str, prompt: &str, files: &[&str]) -> Presence {
1175 Presence {
1176 session: session.into(),
1177 harness: "claude".into(),
1178 engineer: engineer.into(),
1179 repo: "borg-ai".into(),
1180 collective: "corp".into(),
1181 prompt: prompt.into(),
1182 files: files.iter().map(|f| f.to_string()).collect(),
1183 event: event.into(),
1184 ms,
1185 started,
1186 pid: None,
1187 }
1188 }
1189
1190 const NOW: i64 = 10_000_000;
1191
1192 pub(crate) fn fixture() -> Vec<Drone> {
1193 let local = vec![
1194 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"]),
1195 p("cccc3333-x", "alice@x", 300, NOW - presence::IDLE_MS - 60_000, "stop", "write the release notes for the mesh daemon and more words", &[]),
1196 ];
1197 let remote = vec![presence::Remote {
1198 p: p("bbbb2222-y", "bob@y", 200, NOW - 120_000, "stop", "review PR 42", &["README.md"]),
1199 key: "beefcafe".repeat(8),
1200 sig: String::new(),
1201 }];
1202 assemble(local, remote, NOW, |_| true, |p| p.session.starts_with("aaaa"))
1203 }
1204
1205 fn message(by: &str, name: &str, to: &str, kind: msg::Kind, ms: i64, text: &str) -> Message {
1206 message_on("n1", by, name, to, kind, ms, text)
1207 }
1208
1209 fn message_on(node: &str, by: &str, name: &str, to: &str, kind: msg::Kind, ms: i64, text: &str) -> Message {
1210 let from = msg::Sender { by: by.into(), name: name.into(), harness: "claude".into(), node: node.into() };
1211 Message::new(kind, &from, to, None, ms, text)
1212 }
1213
1214 pub(crate) fn inbox_fixture() -> Inbox {
1215 Inbox {
1216 unread: [("aaaa1111-x".to_string(), 2), ("cccc3333-x".to_string(), 0)].into_iter().collect(),
1217 recent: vec![
1218 message("cccc3333-x", "Three of Three", "aaaa1111-x", msg::ASK, NOW - 60_000, "which test\nflakes?"),
1219 message("engineer:alice@x", "alice@x", msg::ALL, msg::TELL, NOW - 120_000, "release at five"),
1220 message_on(&"feed".repeat(16), "gone-session", "Six of Six", "zzzz9999-q", msg::TELL, NOW - 180_000, "done"),
1221 ],
1222 day: 3,
1223 here: "n1".into(),
1224 listen: [("aaaa1111-x".to_string(), crate::listen::Status::Waits), ("cccc3333-x".to_string(), crate::listen::Status::Listens)]
1225 .into_iter()
1226 .collect(),
1227 }
1228 }
1229
1230 #[test]
1231 fn renders_one_screen_per_collective() {
1232 let out = render("corp", Share::Team, &fixture(), &inbox_fixture(), NOW);
1233 let expected = "\
1234corp (team): 3 drone(s), 2 active, 1 remote
1235 DRONE SESSION ENGINEER REPO STATE WAKE IDLE UNREAD DOING FILES
1236 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
1237 Two of Three bbbb222… bob@y @beefca borg-ai waiting - 2m - review PR 42 README.md
1238 Three of Three cccc333… alice@x borg-ai idle on 31m 0 write the release notes for the mesh da…
1239 messages: 3 in the last day, 2 unread
1240 1m Three of Three asks -> One of Three: which test flakes?
1241 2m alice@x -> collective: release at five
1242 3m Six of Six @feedfeedfeed -> zzzz9999: done
1243";
1244 assert_eq!(out, expected, "\n{out}");
1245 let none = Inbox::default();
1246 assert_eq!(render("private", Share::Local, &[], &none, NOW), "private (local, never leaves this machine): 0 drone(s), 0 active, 0 remote\n");
1247 }
1248
1249 fn learned() -> Selection {
1253 let spend = |source: &str, rounds, usd| Spend { source: source.into(), calls: 1, rounds, usd };
1254 let evaluation = Evaluation { sessions: 12, mean: 0.004, ucb: Some(0.031), units: 300, ess: 180.0, max_weight: 6.5, truncated: 3 };
1255 Selection {
1256 policy: "learned",
1257 default: "fixed",
1258 source: "learned",
1259 gate: 0.12,
1260 checks: 480,
1261 sessions: 12,
1262 deviations: vec![Count { name: "route".into(), part: 5, whole: 400 }],
1263 spend: vec![spend("stop", 4, 3.0), spend("live", 0, 2.04)],
1264 evaluation: Some(evaluation),
1265 measurement: Some(Measurement { placebos: 40, judged: 31, tail_shown: 12, tail_applied: 3 }),
1266 test: Some(Test {
1267 risk: 0.1,
1268 ran: 30,
1269 control: 9,
1270 refused: 4,
1271 sessions: 6,
1272 kept: 30,
1273 due_checks: 300,
1274 due_sessions: 30,
1275 due_min_sessions: 15,
1276 ..Test::default()
1277 }),
1278 ..Selection::default()
1279 }
1280 }
1281
1282 #[test]
1283 fn the_selection_line_says_what_decides_and_prices_spend_only_under_a_budget() {
1284 use select::Policy;
1285 use select::summary::Summary;
1286 let fresh = |policy, default| Selection { default, ..Selection::of(&Summary::default(), policy, None) };
1287 let fixed = fresh(Policy::Fixed, "fixed");
1289 assert_eq!(selection_line(&fixed), "the fixed rule (the default; BORG_SELECT=learned opts in to the learned policy) · no graded check folded yet");
1290 assert!(!fixed.shown(), "the default with nothing folded: nothing to show on the drones screen");
1291 let opted = fresh(Policy::Learned, "fixed");
1292 assert_eq!(
1293 selection_line(&opted),
1294 "learned (opt-in, BORG_SELECT=learned), cold start: gate 0, every decision the fixed rule's · no graded check folded yet"
1295 );
1296 assert!(opted.shown(), "a switch decides: shown");
1297 assert_eq!(selection_line(&fresh(Policy::Fixed, "learned")), "the fixed rule (BORG_SELECT=fixed) · no graded check folded yet");
1299 assert!(!fresh(Policy::Learned, "learned").shown() && fresh(Policy::Fixed, "learned").shown());
1300 let d = select::precedence(None, None, false);
1301 assert_eq!(Selection::of(&Summary::default(), Policy::Cap(2), None).default, d.name(), "the default is select.rs's, no switch set");
1302 assert!(!Selection::of(&Summary::default(), d, None).shown());
1303 let mut s = learned();
1304 let line = selection_line(&s);
1305 assert_eq!(
1306 line,
1307 "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)"
1308 );
1309 assert!(!line.contains("budget") && !line.contains("price"), "no budget set: spend measured, never priced: {line}");
1310 let mut early = learned();
1311 early.evaluation = Some(Evaluation { ucb: None, sessions: 3, units: 40, ess: 12.4, max_weight: 3.0, ..Evaluation::default() });
1312 let line = selection_line(&early);
1313 assert!(line.contains(" · monitor +0.000 over 3 sessions, no bound yet, ESS 12 of 40 weighed, largest weight 3.0× the mean · "), "{line}");
1314 early.latency = vec![
1315 Latency { harness: "codex".into(), checks: 20, p50_ms: 10_000, p95_ms: 19_000, budget_ms: Some(30_000), over: 1 },
1316 Latency { harness: "claude".into(), checks: 1, p50_ms: 4_000, p95_ms: 4_000, budget_ms: None, over: 0 },
1317 ];
1318 let line = selection_line(&early);
1319 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}");
1320 s.budget = Some(Budget { tokens: 30_000.0, spent: 24_000.0, price: 0.0002 });
1321 assert!(selection_line(&s).ends_with(" · budget 30000 pool tokens per session: 24000 read, price 2.00e-4 per token"), "{}", selection_line(&s));
1322 s.source = select::REVERTED_SOURCE;
1323 assert!(selection_line(&s).starts_with("REVERTED to the fixed rule"), "{}", selection_line(&s));
1324 }
1325
1326 #[test]
1330 fn a_spent_gate_is_not_a_cold_start() {
1331 use select::Policy;
1332 use select::summary::{Gate, Summary};
1333 let line = |s: &Summary| selection_line(&Selection { default: "learned", ..Selection::of(s, Policy::Learned, None) });
1334 let mut s = Summary::default();
1335 s.fit.key = "3556601-bc6d6191".into();
1336 assert!(line(&s).starts_with("learned, cold start: gate 0, every decision the fixed rule's"), "{}", line(&s));
1338 s.gate = Gate { gain: vec![-0.1018; select::policy::DRAWS], allowance: 0.0760, ..Gate::default() };
1339 let spent = Selection::of(&s, Policy::Learned, None);
1340 assert_eq!((spent.gate, spent.allowance, spent.ledger), (0.0, 0.0760, -0.1018));
1341 assert!(line(&s).starts_with("learned, gate spent (ledger q05 -0.102 against allowance 0.076), deviations need positive evidence · "), "{}", line(&s));
1342 s.fit.key.clear();
1344 assert!(line(&s).starts_with("learned, cold start: gate 0"), "{}", line(&s));
1345 s.fit.key = "3556601-bc6d6191".into();
1347 s.gate.gain = vec![-0.05; select::policy::DRAWS];
1348 assert!(line(&s).starts_with("learned, gate 0.026 (quality it may still risk below the fixed rule)"), "{}", line(&s));
1349 let fixed = Selection::of(&s, Policy::Fixed, None);
1351 assert_eq!((fixed.allowance, fixed.ledger), (0.0, 0.0));
1352 }
1353
1354 #[test]
1357 fn a_store_says_how_its_lessons_are_selected() {
1358 use select::Policy;
1359 let home = std::env::temp_dir().join(format!("borg-top-select-{}", std::process::id()));
1360 let _ = std::fs::remove_dir_all(&home);
1361 std::fs::create_dir_all(&home).unwrap();
1362 let lessons = home.join("lessons.jsonl");
1363 std::fs::write(&lessons, "").unwrap();
1364 let line = |lessons: &Path, policy| selection_line(&selection(lessons, policy));
1365 assert!(line(&lessons, Policy::Learned).ends_with("cold start: gate 0, every decision the fixed rule's · no graded check folded yet"));
1366 assert!(line(&lessons, Policy::Fixed).starts_with("the fixed rule ("), "{}", line(&lessons, Policy::Fixed));
1367 assert!(line(&lessons, Policy::Cap(2)).contains("(BORG_STOP_ASK_CAP, evaluation only)"));
1368 let absent = home.join("absent").join("lessons.jsonl");
1369 assert!(line(&absent, Policy::Attention).contains("(BORG_STOP_ATTENTION=on, evaluation only)"), "no store: the prior");
1370 let _ = std::fs::remove_dir_all(&home);
1371 }
1372
1373 #[test]
1375 fn the_selection_line_follows_the_screen_when_shown() {
1376 let (ds, inbox) = (fixture(), inbox_fixture());
1377 let plain = render("corp", Share::Team, &ds, &inbox, NOW);
1378 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, None, NOW), plain);
1379 let cold = Selection::of(&select::summary::Summary::default(), select::precedence(None, None, false), None);
1380 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, Some(&cold), NOW), plain, "the default at cold start: no line");
1381 let s = learned();
1382 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, Some(&s), NOW), format!("{plain} selection: {}\n", selection_line(&s)));
1383 }
1384
1385 #[test]
1387 fn the_selection_serialises_every_section_for_the_page() {
1388 let v = serde_json::to_value(learned()).unwrap();
1389 let mut keys: Vec<&str> = v.as_object().unwrap().keys().map(String::as_str).collect();
1390 keys.sort_unstable();
1391 let want = "allowance at budget checks declaration default delivery deviations effects evaluation fallbacks fit gate half_life labels latency ledger measurement policy pool sessions source spend test";
1392 assert_eq!(keys, want.split(' ').collect::<Vec<_>>());
1393 assert_eq!((&v["budget"], &v["spend"][0]["source"], &v["deviations"][0]["whole"]), (&Value::Null, &json!("stop"), &json!(400)));
1394 let mut keys: Vec<&str> = v["evaluation"].as_object().unwrap().keys().map(String::as_str).collect();
1395 keys.sort_unstable();
1396 assert_eq!(keys, ["ess", "max_weight", "mean", "sessions", "truncated", "ucb", "units"]);
1397 assert_eq!((&v["default"], &v["measurement"]), (&json!("fixed"), &json!({"placebos": 40, "judged": 31, "tail_shown": 12, "tail_applied": 3})));
1398 let mut keys: Vec<&str> = v["test"].as_object().unwrap().keys().map(String::as_str).collect();
1399 keys.sort_unstable();
1400 let want = "ask_free by_k control dilution due due_checks due_min_sessions due_sessions kept ran refused risk sessions set";
1401 assert_eq!(keys, want.split(' ').collect::<Vec<_>>());
1402 }
1403
1404 #[test]
1409 fn the_judged_share_leaves_out_what_its_session_already_had() {
1410 use crate::outcome::{self, Outcome};
1411 use select::summary::{Summary, fold};
1412 let rec = |kind: &str, check: &str, channel: &str, ms: i64| Outcome {
1413 kind: kind.into(),
1414 session: "s".into(),
1415 lessons: vec!["A".into()],
1416 check: check.into(),
1417 channel: channel.into(),
1418 ms,
1419 ..Default::default()
1420 };
1421 let graded = |check: &str, ms: i64| Outcome {
1422 grades: vec![2],
1423 ranks: vec![0],
1424 applied: 1,
1425 free: vec!["A".into()],
1426 pf: vec![1.0],
1427 policy: "fixed".into(),
1428 shown: 20,
1429 tokens: 6_000,
1430 ..rec(outcome::GRADED, check, "live", ms)
1431 };
1432 let log = [
1433 graded("c1", 10),
1434 Outcome { route: outcome::ROUTE_FREE.into(), ..rec(outcome::DELIVERY, "c1", "", 11) },
1435 Outcome { judge: 2, ..rec(outcome::CHANGED, "c1", "queued", 12) },
1436 graded("c2", 20),
1437 Outcome { route: outcome::ROUTE_DROPPED_DONE.into(), ..rec(outcome::DELIVERY, "c2", "", 21) },
1438 Outcome { lessons: vec!["B".into()], ..rec(outcome::ASKED, "", "early", 25) },
1440 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) },
1441 Outcome { lessons: vec!["B".into()], route: outcome::ROUTE_ASKED.into(), ..rec(outcome::DELIVERY, "c3", "", 31) },
1442 Outcome { lessons: vec!["B".into()], judge: 2, ..rec(outcome::CHANGED, "c3", "live", 32) },
1443 ];
1444 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1445 let s = fold(&text, Summary::default());
1446 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");
1447 let sel = Selection::of(&s, select::Policy::Fixed, None);
1448 let row = |name: &str| sel.labels.iter().find(|c| c.name == name).map(|c| (c.part, c.whole));
1449 assert_eq!(row("judged, free"), Some((1, 1)), "c1's delivered and judged; c2's had, out: {:?}", sel.labels);
1450 assert_eq!(row("judged, asked"), Some((1, 1)), "c3's question of what B's session had, judged: {:?}", sel.labels);
1451 }
1452
1453 #[test]
1458 fn the_measurement_counts_the_placebos_the_fixed_rule_drew_and_judged() {
1459 use crate::outcome::{self, Outcome};
1460 use select::summary::{Summary, fold};
1461 let rec = |kind: &str, lessons: &[&str], check: &str, channel: &str, ms: i64| Outcome {
1462 kind: kind.into(),
1463 session: "s".into(),
1464 lessons: lessons.iter().map(|l| l.to_string()).collect(),
1465 check: check.into(),
1466 channel: channel.into(),
1467 ms,
1468 ..Default::default()
1469 };
1470 let graded = Outcome {
1471 grades: vec![1, 1, 2],
1472 ranks: vec![0, 1, 2],
1473 applied: 3,
1474 free: vec!["C".into()],
1475 placebo: vec!["A".into(), "B".into()],
1476 psi: 0.5,
1477 policy: "fixed".into(),
1478 shown: 20,
1479 tokens: 6_000,
1480 ..rec(outcome::GRADED, &["A", "B", "C"], "c1", "stop", 10)
1481 };
1482 let tailed = Outcome {
1484 grades: vec![1],
1485 ranks: vec![0],
1486 applied: 1,
1487 policy: "fixed".into(),
1488 shown: 20,
1489 tokens: 6_000,
1490 tail: vec!["T1".into(), "T2".into()],
1491 pt: 0.5,
1492 fill: vec![21, 25, 21],
1493 ..rec(outcome::GRADED, &["T2"], "c2", "stop", 14)
1494 };
1495 let log = [
1496 graded,
1497 tailed,
1498 Outcome { route: outcome::ROUTE_PLACEBO.into(), ..rec(outcome::DELIVERY, &["A"], "c1", "", 11) },
1499 Outcome { route: outcome::ROUTE_PLACEBO.into(), ..rec(outcome::DELIVERY, &["B"], "c1", "", 11) },
1500 Outcome { judge: 2, ..rec(outcome::IGNORED, &["A"], "c1", outcome::PLACEBO, 12) },
1501 Outcome { judge: 2, ..rec(outcome::CHANGED, &["C"], "c1", "queued", 13) },
1502 ];
1503 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1504 let sel = Selection::of(&fold(&text, Summary::default()), select::Policy::Fixed, None);
1505 assert_eq!(sel.measurement, Some(Measurement { placebos: 2, judged: 1, tail_shown: 2, tail_applied: 1 }), "{sel:?}");
1506 assert!(selection_line(&sel).contains(" · 2 checks · placebos 2 drawn, 1 judged · tail 2 shown past the pool, 1 applied"), "{}", selection_line(&sel));
1507 assert_eq!(Selection::of(&Summary::default(), select::Policy::Fixed, None).measurement, None, "nothing folded: not measured");
1508 }
1509
1510 #[test]
1514 fn the_selection_counts_the_learned_decisions_the_fixed_rule_took_over_by_reason() {
1515 use crate::outcome::{self, Outcome};
1516 use select::summary::{Summary, fold};
1517 let graded = |check: &str, policy: &str, reason: &str, ms: i64| Outcome {
1518 kind: outcome::GRADED.into(),
1519 session: "s".into(),
1520 lessons: vec!["A".into()],
1521 grades: vec![1],
1522 ranks: vec![0],
1523 applied: 1,
1524 free: vec!["A".into()],
1525 pf: vec![1.0],
1526 channel: "live".into(),
1527 check: check.into(),
1528 policy: policy.into(),
1529 reason: reason.into(),
1530 ms,
1531 ..Default::default()
1532 };
1533 let log = [
1534 graded("c1", "learned", "", 1),
1535 graded("c2", "fixed", outcome::DECIDED_BUSY, 2),
1536 graded("c3", "fixed", outcome::DECIDED_NO_RECORD, 3),
1537 graded("c4", "fixed", outcome::DECIDED_BUSY, 4),
1538 ];
1539 let lines = |log: &[Outcome]| log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect::<String>();
1540 let sel = Selection::of(&fold(&lines(&log), Summary::default()), select::Policy::Learned, None);
1541 let rows: Vec<(&str, u64, u64)> = sel.fallbacks.iter().map(|c| (c.name.as_str(), c.part, c.whole)).collect();
1542 assert_eq!(rows, [("lock_busy", 2, 4), ("lock_failed", 0, 4), ("unrecorded", 1, 4)]);
1543 let line = selection_line(&sel);
1544 assert!(line.contains(" · 4 checks · learned decisions the fixed rule took over: lock_busy 2, unrecorded 1 · "), "{line}");
1545 let none = Selection::of(&fold(&lines(&log[..1]), Summary::default()), select::Policy::Learned, None);
1546 assert!(none.fallbacks.iter().all(|c| c.part == 0) && none.fallbacks.len() == 3, "{none:?}");
1547 assert!(!selection_line(&none).contains("took over"), "{}", selection_line(&none));
1548 assert!(Selection::of(&Summary::default(), select::Policy::Learned, None).fallbacks.is_empty(), "nothing folded: not measured");
1549 }
1550
1551 #[test]
1556 fn the_selection_fills_latency_evaluation_and_delivery_from_the_summary() {
1557 use select::summary::{Summary, fold};
1558 let call = |harness: &str, channel: &str, dur_ms: u64, failed: bool| {
1559 let o = crate::outcome::Outcome {
1560 kind: "graded".into(),
1561 session: "s".into(),
1562 channel: channel.into(),
1563 harness: harness.into(),
1564 dur_ms,
1565 failed,
1566 tokens: 6_000,
1567 ..Default::default()
1568 };
1569 format!("{}\n", serde_json::to_string(&o).unwrap())
1570 };
1571 let mut log: String = (1..=20).map(|k| call("codex", "stop", k * 1_000, k == 20)).collect();
1572 log += &call("claude", "stop", 4_000, false);
1573 log += &call("codex", "live", 99_000, false);
1574 log += &call("nonesuch", "stop", 5_000, false);
1575 let mut s = fold(&log, Summary::default());
1576 s.monitor = select::policy::Monitor {
1577 sessions: 12,
1578 units: 400,
1579 mean: 0.004,
1580 ucb: Some(0.031),
1581 labelled: 300,
1582 ess: 180.0,
1583 max_weight: 6.5,
1584 truncated: 3,
1585 ..Default::default()
1586 };
1587 s.fit.key = "fit".into();
1588 s.fit.rates = select::policy::Rates { free_stop: 0.6, free_live: 0.8, later: 0.4, wait: 0.25, ask_live: 1.0 };
1589 let sel = Selection::of(&s, select::Policy::Learned, None);
1590 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);
1591 assert!(budget("codex").is_some() && budget("claude").is_none(), "codex names a stop budget, claude none");
1592 let want = vec![
1593 Latency { harness: "claude".into(), checks: 1, p50_ms: 4_000, p95_ms: 4_000, budget_ms: None, over: 0 },
1594 Latency { harness: "codex".into(), checks: 20, p50_ms: 10_000, p95_ms: 19_000, budget_ms: budget("codex"), over: 1 },
1595 ];
1596 assert_eq!(sel.latency, want);
1597 let e = sel.evaluation.as_ref().expect("the monitor weighed labelled applies");
1598 assert_eq!(*e, Evaluation { sessions: 12, mean: 0.004, ucb: Some(0.031), units: 300, ess: 180.0, max_weight: 6.5, truncated: 3 });
1599 assert_eq!(sel.delivery.as_ref().map(|r| r.wait), Some(0.25));
1600 let keep = select::summary::LATENCY_KEEP as u64;
1602 let many: String = (1..=keep + 1).map(|k| call("claude", "stop", k, false)).collect();
1603 let kept = Selection::of(&fold(&many, Summary::default()), select::Policy::Learned, None);
1604 assert_eq!((kept.latency[0].checks, kept.latency[0].p50_ms), (keep, 1 + keep / 2), "the oldest call dropped");
1605 let cold = Selection::of(&Summary::default(), select::Policy::Learned, None);
1606 assert!(cold.latency.is_empty() && cold.evaluation.is_none() && cold.delivery.is_none(), "{cold:?}");
1607 assert_eq!((quantile(&[], 0.95), quantile(&[7], 0.5)), (0, 7));
1608 }
1609
1610 fn tested(check: &str, session: &str, ms: i64, source: &str, xk: Option<u32>, xs: Option<u32>) -> crate::outcome::Outcome {
1613 crate::outcome::Outcome {
1614 kind: crate::outcome::GRADED.into(),
1615 session: session.into(),
1616 lessons: vec!["A".into(), "B".into()],
1617 grades: vec![2, 1],
1618 ranks: vec![0, 1],
1619 applied: 2,
1620 check: check.into(),
1621 channel: "live".into(),
1622 policy: "learned".into(),
1623 source: source.into(),
1624 xk,
1625 xs,
1626 ms,
1627 ..Default::default()
1628 }
1629 }
1630
1631 #[test]
1637 fn the_test_section_is_the_summarys_fold_of_its_checks() {
1638 use select::policy::{ALPHA, EVIDENCE, EXPERIMENT, EXPERIMENT_CONTROL, GATE};
1639 use select::summary::{Summary, fold};
1640 let log = [
1641 tested("c1", "s1", 1, EXPERIMENT, Some(2), Some(1)),
1642 tested("c2", "s1", 2, EXPERIMENT_CONTROL, Some(2), Some(0)),
1643 tested("c3", "s2", 3, EXPERIMENT, Some(2), Some(1)),
1644 tested("c4", "s2", 4, GATE, Some(2), Some(1)),
1645 tested("c5", "s2", 5, GATE, Some(1), Some(0)),
1646 tested("c6", "s3", 6, EVIDENCE, None, None),
1647 tested("c7", "s3", 7, GATE, None, None),
1648 crate::outcome::Outcome { policy: "fixed".into(), ..tested("c8", "s3", 8, EXPERIMENT, Some(2), Some(2)) },
1649 ];
1650 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1651 let s = fold(&text, Summary::default());
1652 let t = Selection::of(&s, select::Policy::Learned, Some(0.2)).test.expect("the learned policy shows its test");
1653 let draws = |k: u32, draws: Vec<u64>, refused: u64| Draws { k, draws, refused };
1654 assert_eq!((t.ran, t.control, t.refused, t.sessions), (5, 1, 2, 2), "{t:?}");
1655 assert_eq!(t.by_k, vec![draws(1, vec![1, 0], 1), draws(2, vec![1, 3, 0], 1)]);
1656 assert_eq!((t.risk, t.set), (0.2, true), "the collective's risk");
1657 assert!(!t.due && t.dilution.is_empty() && t.ask_free.is_none(), "5 checks, 2 sessions: not due, nothing estimated");
1658 assert_eq!((t.due_checks, t.due_sessions), (select::replay::TEST_CHECKS, u64::from(select::replay::TEST_SESSIONS)));
1659 let line = selection_line(&Selection::of(&s, select::Policy::Learned, Some(0.2)));
1660 assert!(
1661 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 · "),
1662 "{line}"
1663 );
1664 assert_eq!(t.kept, t.ran, "every drawn check still held: {t:?}");
1665 let mut compacted = s.clone();
1667 compacted.checks.remove(0);
1668 let c = Selection::of(&compacted, select::Policy::Learned, Some(0.2)).test.unwrap();
1669 assert_eq!((c.ran, c.kept), (5, 4), "{c:?}");
1670 assert!(
1671 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"),
1672 "{}",
1673 test_line(&c)
1674 );
1675 let default = Selection::of(&s, select::Policy::Learned, None).test.unwrap();
1676 assert_eq!((default.risk, default.set), (ALPHA, false), "unset: the policy's ALPHA");
1677 let cold = Selection::of(&Summary::default(), select::Policy::Learned, Some(0.0));
1679 assert_eq!(cold.test.as_ref().map(|t| (t.ran, t.risk)), Some((0, 0.0)));
1680 assert_eq!(test_line(cold.test.as_ref().unwrap()), "question-count test off (risk 0)");
1681 let none = Selection::of(&Summary::default(), select::Policy::Learned, None).test.unwrap();
1682 assert_eq!(test_line(&none), "question-count test: no check yet, risk 0.10");
1683 assert!(Selection::of(&Summary::default(), select::Policy::Fixed, None).test.is_none());
1685 assert_eq!(Selection::of(&s, select::Policy::Fixed, None).test.map(|t| t.ran), Some(5), "an earlier opt-in's checks");
1686 }
1687
1688 #[test]
1692 fn the_gate_shown_is_the_one_in_force_at_the_collectives_risk() {
1693 use select::summary::{Gate, Summary};
1694 let s = Summary { gate: Gate { gain: vec![0.0; select::policy::DRAWS], allowance: 0.3, ..Gate::default() }, ..Summary::default() };
1695 let gate = |risk: Option<f64>| Selection::of(&s, select::Policy::Learned, risk).gate;
1696 assert!((gate(None) - 0.3).abs() < 1e-12, "{}", gate(None));
1697 assert!((gate(Some(0.2)) - 0.6).abs() < 1e-12, "twice ALPHA: twice the allowance, {}", gate(Some(0.2)));
1698 assert_eq!(gate(Some(0.0)), 0.0, "risk 0");
1699 }
1700
1701 #[test]
1704 fn the_test_readout_shows_the_replays_estimates_once_due() {
1705 use crate::outcome::{self, Outcome};
1706 use select::policy::EXPERIMENT;
1707 use select::summary::{Summary, fold};
1708 let sessions = u64::from(select::replay::TEST_SESSIONS);
1709 let mut text = String::new();
1710 let mut line = |o: &Outcome| text.push_str(&format!("{}\n", serde_json::to_string(o).unwrap()));
1711 for i in 0..sessions {
1712 let (session, check, ms) = (format!("s{i}"), format!("{i:012x}"), 10 * i as i64);
1713 let asked = i % 2 == 0;
1715 let g = Outcome {
1716 lessons: vec!["A".into()],
1717 grades: vec![2],
1718 ranks: vec![0],
1719 applied: 1,
1720 chosen: if asked { vec!["A".into()] } else { Vec::new() },
1721 free: if asked { Vec::new() } else { vec!["A".into()] },
1722 p: vec![0.5],
1723 pf: vec![0.5],
1724 ..tested(&check, &session, ms, EXPERIMENT, Some(1), Some(u32::from(asked)))
1725 };
1726 line(&g);
1727 let rec = |kind: &str, ms: i64| Outcome {
1728 kind: kind.into(),
1729 session: session.clone(),
1730 lessons: vec!["A".into()],
1731 check: check.clone(),
1732 ms,
1733 ..Default::default()
1734 };
1735 let route = if asked { outcome::ROUTE_ASKED } else { outcome::ROUTE_FREE };
1736 line(&Outcome { route: route.into(), ..rec(outcome::DELIVERY, ms + 1) });
1737 let verdict = if asked { outcome::CHANGED } else { outcome::IGNORED };
1739 line(&Outcome { judge: 2, channel: if asked { "live" } else { "queued" }.into(), ..rec(verdict, ms + 2) });
1740 }
1741 let s = fold(&text, Summary::default());
1742 let t = Selection::of(&s, select::Policy::Learned, None).test.unwrap();
1743 assert!(t.due && t.sessions == sessions, "{t:?}");
1744 let r = select::replay::test_readout(&s);
1745 let want = r.ask_free.as_ref().and_then(|x| x.value).expect("judged 2s on both routes");
1746 let e = t.ask_free.as_ref().expect("estimated once due");
1747 assert_eq!((e.mean, e.unit.as_str()), (100.0 * want, "points changed"));
1748 assert!(e.mean > 0.0, "asking changed the work, handing over did not: {e:?}");
1749 assert!(t.dilution.is_empty() && r.by_asked.is_empty(), "{t:?}");
1751 let line = selection_line(&Selection::of(&s, select::Policy::Learned, None));
1752 assert!(line.contains(&format!("; readout: grade 2 asked − handed over {:+.1} [", e.mean)), "{line}");
1753 }
1754
1755 #[test]
1758 fn unread_counts_follow_the_shown_once_claims() {
1759 let home = std::env::temp_dir().join(format!("borg-top-inbox-{}", std::process::id()));
1760 let _ = std::fs::remove_dir_all(&home);
1761 std::fs::create_dir_all(&home).unwrap();
1762 let ds = fixture();
1763 assert!(inbox(&home, &ds, NOW).recent.is_empty());
1764 let here = msg::node(&home);
1765 msg::send(&home, message_on(&here, "cccc3333-x", "Three of Three", "aaaa1111-x", msg::TELL, NOW - 3_000, "look at recon"), NOW).unwrap();
1766 msg::send(&home, message_on(&here, "cccc3333-x", "Three of Three", msg::ALL, msg::TELL, NOW - 2_000, "lunch"), NOW).unwrap();
1767 let i = inbox(&home, &ds, NOW);
1768 assert_eq!(i.unread.get("aaaa1111-x"), Some(&2), "{i:?}");
1769 assert_eq!(i.unread.get("cccc3333-x"), Some(&0), "its own broadcast");
1770 assert!(!i.unread.contains_key(&ds[1].p.session), "remote: claimed on its node");
1771 assert_eq!((i.day, i.recent[0].text.as_str()), (2, "lunch"));
1772 let caps = crate::harness::by_name("claude").expect("a harness").caps();
1773 let ev = json!({"session_id": "aaaa1111-x"});
1774 assert!(msg::deliver(&home, caps, "prompt", &ev, Some(100), NOW, None).is_some());
1775 assert_eq!(inbox(&home, &ds, NOW).unread.get("aaaa1111-x"), Some(&0));
1776 let _ = std::fs::remove_dir_all(&home);
1777 }
1778
1779 fn seen_lines(home: &Path) -> Vec<(String, Vec<String>)> {
1781 msg::lines(home)
1782 .into_iter()
1783 .filter_map(|l| match l {
1784 msg::Line::Seen(s) => Some((s.by, s.ids)),
1785 msg::Line::Message(_) => None,
1786 })
1787 .collect()
1788 }
1789
1790 fn shell(sessions: Vec<(String, &'static str)>, pids: Vec<(u32, &'static str)>) -> impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String> {
1793 move |ds| msg::pick(sessions.clone(), &pids, ds)
1794 }
1795
1796 fn acked_fixture(base: &Path) -> (std::path::PathBuf, String, Message) {
1799 let _ = std::fs::remove_dir_all(base);
1800 let (home, work) = (base.join("store"), base.join("work"));
1801 std::fs::create_dir_all(&home).unwrap();
1802 std::fs::create_dir_all(&work).unwrap();
1803 let git = |a: &[&str]| std::process::Command::new("git").arg("-C").arg(&work).args(a).output().is_ok_and(|o| o.status.success());
1804 assert!(git(&["init", "-q"]) && git(&["config", "user.email", "ann@x"]));
1805 let here = msg::node(&home);
1806 let to_me = message_on(&here, "aaaa1111-x", "One of Three", msg::ENGINEER, msg::TELL, NOW - 3_000, "the build is green");
1807 let to_drone = message_on(&here, "cccc3333-x", "Three of Three", "aaaa1111-x", msg::TELL, NOW - 2_000, "look at recon");
1808 msg::send(&home, to_me.clone(), NOW).unwrap();
1809 msg::send(&home, to_drone, NOW).unwrap();
1810 (home, work.display().to_string(), to_me)
1811 }
1812
1813 #[test]
1817 fn the_screen_acknowledges_the_messages_to_the_engineer_it_prints() {
1818 let base = std::env::temp_dir().join(format!("borg-top-ack-{}", std::process::id()));
1819 let (home, cwd, to_me) = acked_fixture(&base);
1820 let routes = [Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local }];
1821 let terminal = || shell(Vec::new(), Vec::new());
1822 let json = screen_of(&routes, true, &cwd, terminal(), true, NOW);
1823 assert!(json.contains("the build is green") && seen_lines(&home).is_empty(), "--json acknowledges nothing: {json}");
1824 let text = screen_of(&routes, true, &cwd, terminal(), false, NOW);
1825 assert!(text.contains("the build is green") && text.contains("look at recon"), "{text}");
1826 assert_eq!(seen_lines(&home), [("engineer:ann@x".to_string(), vec![to_me.id.clone()])], "the engineer's message only, as theirs");
1827 screen_of(&routes, true, &cwd, terminal(), false, NOW + 1_000);
1828 assert_eq!(seen_lines(&home).len(), 1, "acknowledged once: a later look writes no second line");
1829 let _ = std::fs::remove_dir_all(&base);
1830 }
1831
1832 #[test]
1837 fn a_session_running_the_screen_acknowledges_nothing_as_the_engineer() {
1838 let base = std::env::temp_dir().join(format!("borg-top-ack-session-{}", std::process::id()));
1839 let (home, cwd, to_me) = acked_fixture(&base);
1840 let other = base.join("other");
1841 std::fs::create_dir_all(&other).unwrap();
1842 let pid = std::process::id();
1843 let oc = Presence { harness: "opencode".into(), pid: Some(pid), ..p("dddd4444-x", "ann@x", NOW - 9_000, NOW - 1_000, "prompt", "docs", &[]) };
1844 presence::write(&other, &oc).unwrap();
1845 let routes = [
1846 Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local },
1847 Route::Collective { name: "side".into(), home: other.clone(), share: Share::Local },
1848 ];
1849 let shells = [
1850 ("Claude Code", shell(vec![("eeee5555-x".into(), "claude")], Vec::new())),
1851 ("Codex in Claude Code, neither with presence", shell(vec![("eeee5555-x".into(), "claude"), ("ffff6666-x".into(), "codex")], Vec::new())),
1852 ("opencode, its drone in the other collective", shell(Vec::new(), vec![(pid, "opencode")])),
1853 ];
1854 for (what, caller) in shells {
1855 let text = screen_of(&routes, false, &cwd, caller, false, NOW);
1856 assert!(text.contains("the build is green"), "{what}: shown {text}");
1857 assert!(seen_lines(&home).is_empty() && seen_lines(&other).is_empty(), "{what}: acknowledged nothing as the engineer");
1858 }
1859 screen_of(&routes, false, &cwd, shell(Vec::new(), vec![(pid + 1, "opencode")]), false, NOW);
1860 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");
1861 let _ = std::fs::remove_dir_all(&base);
1862 }
1863
1864 #[test]
1869 fn the_printed_screen_acknowledges_only_the_messages_whose_text_it_prints() {
1870 let base = std::env::temp_dir().join(format!("borg-top-ack-cut-{}", std::process::id()));
1871 let (home, cwd, to_me) = acked_fixture(&base);
1872 let email = format!("{}@x", "a".repeat(58));
1874 assert!(std::process::Command::new("git").args(["-C", &cwd, "config", "user.email", &email]).output().is_ok_and(|o| o.status.success()));
1875 let (here, to) = (msg::node(&home), format!("{}:{email}", msg::ENGINEER));
1876 let one = message_on(&here, "bbbb2222-x", &"n".repeat(52), &to, msg::TELL, NOW - 1_500, "one character shows");
1877 let none = message_on(&here, "dddd4444-x", &"n".repeat(53), &to, msg::TELL, NOW - 1_000, "none shows");
1878 msg::send(&home, one.clone(), NOW).unwrap();
1879 msg::send(&home, none.clone(), NOW).unwrap();
1880 let routes = [Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local }];
1881 let text = screen_of(&routes, true, &cwd, shell(Vec::new(), Vec::new()), false, NOW);
1882 assert!(text.contains(&format!("{} -> {email}: o…", one.name)) && text.contains(&format!("{} -> {email}: …", none.name)), "{text}");
1883 let mut acked: Vec<String> = seen_lines(&home).into_iter().flat_map(|(by, ids)| ids.into_iter().map(move |id| format!("{by} {id}"))).collect();
1884 acked.sort();
1885 let mut want = [format!("{to} {}", to_me.id), format!("{to} {}", one.id)];
1886 want.sort();
1887 assert_eq!(acked, want, "the line that shows one character of its text is read; the one cut before it is not");
1888 let _ = std::fs::remove_dir_all(&base);
1889 }
1890
1891 #[test]
1895 fn a_live_screen_acknowledges_each_message_once() {
1896 let base = std::env::temp_dir().join(format!("borg-top-ack-once-{}", std::process::id()));
1897 let _ = std::fs::remove_dir_all(&base);
1898 std::fs::create_dir_all(&base).unwrap();
1899 let home = base.clone();
1900 let here = msg::node(&home);
1901 let to_me = |by: &str, ms: i64, text: &str| message_on(&here, by, "a drone", msg::ENGINEER, msg::TELL, ms, text);
1902 let first = to_me("aaaa1111-x", NOW - 3_000, "the build is green");
1903 msg::send(&home, first.clone(), NOW).unwrap();
1904 let cwd = base.display().to_string();
1905 let mut done = HashSet::new();
1906 let recent = |now: i64| inbox(&home, &[], now).recent;
1907 acknowledge(&home, &cwd, &recent(NOW).iter().collect::<Vec<_>>(), &mut done, NOW);
1908 assert_eq!(seen_lines(&home).len(), 1, "{:?}", seen_lines(&home));
1909 assert!(done.contains(&first.id));
1910 let kept: String = std::fs::read_to_string(msg::path(&home)).unwrap().lines().filter(|l| !l.contains("\"seen\"")).map(|l| format!("{l}\n")).collect();
1913 std::fs::write(msg::path(&home), kept).unwrap();
1914 acknowledge(&home, &cwd, &recent(NOW + 1_000).iter().collect::<Vec<_>>(), &mut done, NOW + 1_000);
1915 assert!(seen_lines(&home).is_empty(), "skipped: no second ack of a done id");
1916 let second = to_me("cccc3333-x", NOW + 2_000, "and the docs");
1918 msg::send(&home, second.clone(), NOW + 2_000).unwrap();
1919 acknowledge(&home, &cwd, &recent(NOW + 3_000).iter().collect::<Vec<_>>(), &mut done, NOW + 3_000);
1920 assert_eq!(seen_lines(&home).into_iter().map(|(_, ids)| ids).collect::<Vec<_>>(), [vec![second.id.clone()]], "only the new one");
1921 let _ = std::fs::remove_dir_all(&base);
1922 }
1923
1924 #[test]
1927 fn a_sender_is_tied_to_the_node_its_line_names() {
1928 let ds = fixture();
1929 let own = message("aaaa1111-x", "anything", msg::ALL, msg::TELL, NOW, "hi");
1930 assert_eq!(sender(&ds, "n1", &own), "One of Three", "this node's drone");
1931 let forged = message_on("evil0123456789", "aaaa1111-x", "One of Three", msg::ALL, msg::TELL, NOW, "hi");
1932 assert_eq!(sender(&ds, "n1", &forged), "One of Three @evil01234567", "the claim, with the node it came from");
1933 let remote = message_on(&"beefcafe".repeat(8), &ds[1].p.session, "x", msg::ALL, msg::TELL, NOW, "hi");
1934 assert_eq!(sender(&ds, "n1", &remote), "Two of Three @beefcafebeef", "the remote drone on its own node");
1935 assert_eq!(recipient(&ds, "n1", "aaaa1111-x@n1"), "One of Three");
1936 assert_eq!(recipient(&ds, "n1", "aaaa1111-x@evil"), "aaaa1111", "not the local drone of that id");
1937 assert_eq!((recipient(&ds, "n1", "engineer"), recipient(&ds, "n1", msg::ALL)), ("engineer".into(), "collective".into()), "an `all` line");
1938 assert_eq!(recipient(&ds, "n1", "local@n1"), "local");
1939 assert_eq!(recipient(&ds, "n1", "local@beefcafebeefcafe00"), "local @beefcafebeef");
1940 }
1941
1942 #[test]
1946 fn the_line_on_borg_itself_names_a_release_and_the_protocols_that_differ() {
1947 let home = std::env::temp_dir().join(format!("borg-top-version-{}", std::process::id()));
1948 let _ = std::fs::remove_dir_all(&home);
1949 std::fs::create_dir_all(&home).unwrap();
1950 let newer = update::Checked { checked_ms: 1, latest: Some("99.0.0".into()), error: None };
1951 let current = update::Checked { latest: Some(update::VERSION.into()), ..newer.clone() };
1952 assert_eq!(version_line([home.as_path()], None, true), None);
1953 assert_eq!(version_line([home.as_path()], Some(¤t), true), None);
1954 let release = format!("borg 99.0.0 is released, this is {}: borg update", update::VERSION);
1955 assert_eq!(version_line([home.as_path()], Some(&newer), true), Some(release.clone()));
1956 assert_eq!(version_line([home.as_path()], Some(&newer), false), None, "BORG_UPDATE_CHECK=off");
1957
1958 let ours = protocol::current();
1959 let v = |protocol, engineer: Option<&str>| protocol::PeerVersion { borg: None, protocol, ms: 1, engineer: engineer.map(str::to_string) };
1960 protocol::note_version(&home, "k-same", &v(ours, Some("ann@x"))).unwrap();
1961 assert_eq!(version_line([home.as_path()], None, true), None, "a member on this node's protocol is no news");
1962 protocol::note_version(&home, "k-old", &v(ours - 1, Some("bob@x"))).unwrap();
1963 let mismatch = protocol::Mismatch { theirs: format!("borg/mesh/{}", ours + 1), ours: format!("borg/mesh/{ours}"), ms: 2 };
1964 protocol::note_mismatch(&home, "k-new", mismatch).unwrap();
1965 let line = version_line([home.as_path()], Some(&newer), true).unwrap();
1966 let old = format!("bob@x on /{} misses /{ours}: {}", ours - 1, protocol::note(ours).unwrap());
1967 let none = format!("nothing syncs with peer k-new (borg/mesh/{}, this node borg/mesh/{ours})", ours + 1);
1968 assert_eq!(line, format!("{release} · {none} · {old}"));
1969 let _ = std::fs::remove_dir_all(&home);
1970 }
1971
1972 #[test]
1973 fn json_for_scripts() {
1974 let v = to_json_with("corp", Share::Team, &fixture(), &inbox_fixture());
1975 assert_eq!((v["drones"][0]["unread"].as_u64(), v["drones"][1].get("unread")), (Some(2), None));
1976 assert_eq!((v["messages"]["day"].as_u64(), v["messages"]["recent"][0]["kind"].as_str()), (Some(3), Some("ask")));
1977 assert_eq!(v["collective"], "corp");
1978 let d = &v["drones"];
1979 assert_eq!(d.as_array().unwrap().len(), 3);
1980 assert_eq!(d[0]["designation"], "One of Three");
1981 assert_eq!(d[0]["session"], "aaaa1111-x");
1982 assert_eq!(d[0]["state"], "working");
1983 assert_eq!(d[0]["paused"], true);
1984 assert_eq!(d[1]["node"], "beefcafe".repeat(8));
1985 assert_eq!(d[1]["state"], "waiting");
1986 assert_eq!(d[2]["state"], "idle");
1987 assert_eq!(d[2]["files"], json!([]));
1988 }
1989
1990 #[test]
1992 fn a_name_matching_here_and_on_a_member_names_the_drone_here() {
1993 let local = vec![p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "post_tool", "a", &[])];
1994 let remote = ["aaaa1111-x", "aaaa2222-z"].map(|s| presence::Remote {
1995 p: p(s, "bob@y", 50, NOW - 5_000, "post_tool", "b", &[]),
1996 key: "beefcafe".repeat(8),
1997 sig: String::new(),
1998 });
1999 let ds = assemble(local, remote.into(), NOW, |_| true, |_| false);
2000 for t in ["aaaa", "aaaa1111", "aaaa1111-x"] {
2001 let d = resolve(&ds, t).unwrap().unwrap();
2002 assert!(d.node.is_empty() && d.p.session == "aaaa1111-x", "{t}");
2003 }
2004 assert_eq!(resolve(&ds, "aaaa2").unwrap().unwrap().node, "beefcafe".repeat(8), "only a member's matches");
2005 assert!(!resolve(&ds, "One").unwrap().unwrap().node.is_empty(), "a number names one drone of all, wherever it runs");
2006 }
2007
2008 #[test]
2010 fn a_name_matching_an_ended_drone_here_and_a_live_member_names_the_live_one() {
2011 let local = vec![p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "end", "a", &[])];
2012 let remote =
2013 vec![presence::Remote { p: p("aaaa2222-z", "bob@y", 50, NOW - 5_000, "post_tool", "b", &[]), key: "beefcafe".repeat(8), sig: String::new() }];
2014 let ds = assemble(local, remote, NOW, |_| true, |_| false);
2015 let d = resolve(&ds, "aaaa").unwrap().unwrap();
2016 assert_eq!((d.p.session.as_str(), d.node.as_str()), ("aaaa2222-z", "beefcafe".repeat(8).as_str()), "the live one, though remote");
2017 assert!(resolve(&ds, "aaaa1111").unwrap().unwrap().node.is_empty(), "an ended drone is still named by its own id");
2018 }
2019
2020 #[test]
2021 fn resolves_drones_by_designation_or_session() {
2022 let ds = fixture();
2023 assert_eq!(resolve(&ds, "Two").unwrap().unwrap().p.session, "bbbb2222-y");
2024 assert_eq!(resolve(&ds, "two of three").unwrap().unwrap().p.session, "bbbb2222-y");
2025 assert_eq!(resolve(&ds, "3").unwrap().unwrap().p.session, "cccc3333-x");
2026 assert_eq!(resolve(&ds, "aaaa").unwrap().unwrap().p.session, "aaaa1111-x");
2027 assert!(resolve(&ds, "all").unwrap().is_none());
2028 assert!(resolve(&ds, "Four").is_err());
2029 assert!(resolve(&ds, "zzzz").is_err());
2030 assert!(resolve(&ds, "aa").is_err(), "too short a prefix");
2031 assert!(["collective", "local"].iter().all(|w| resolve(&ds, w).unwrap().is_none()));
2032 assert_eq!(resolve(&ds, "bbbb").unwrap().unwrap().p.session, "bbbb2222-y", "nothing here matches: the member's");
2033 assert_eq!(word(21), "21");
2034 }
2035}