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