1use std::collections::{BTreeMap, HashSet};
23use std::path::Path;
24
25use serde::Serialize;
26use serde_json::{Value, json};
27
28use crate::clock::{DAY_MS, ago};
29use crate::collective::{self, Route, Share};
30use crate::msg::{self, Address, Message};
31use crate::presence::{self, Presence, State};
32use crate::{install, learn, select, steer};
33
34#[derive(Debug, Clone, Serialize)]
35pub struct Drone {
36 pub designation: String,
38 pub number: usize,
40 pub node: String,
42 pub state: State,
43 pub paused: bool,
45 #[serde(flatten)]
46 pub p: Presence,
47}
48
49#[derive(Debug, Clone, Default, Serialize)]
51pub struct Inbox {
52 pub unread: BTreeMap<String, usize>,
54 pub recent: Vec<Message>,
56 pub day: usize,
58 #[serde(skip)]
60 pub here: String,
61 pub listen: BTreeMap<String, crate::listen::Status>,
63}
64
65pub const RECENT: usize = 5;
67
68const NUMBERS: [&str; 20] = [
69 "One",
70 "Two",
71 "Three",
72 "Four",
73 "Five",
74 "Six",
75 "Seven",
76 "Eight",
77 "Nine",
78 "Ten",
79 "Eleven",
80 "Twelve",
81 "Thirteen",
82 "Fourteen",
83 "Fifteen",
84 "Sixteen",
85 "Seventeen",
86 "Eighteen",
87 "Nineteen",
88 "Twenty",
89];
90
91pub fn word(n: usize) -> String {
92 NUMBERS.get(n.wrapping_sub(1)).map(|s| s.to_string()).unwrap_or_else(|| n.to_string())
93}
94
95pub fn assemble(local: Vec<Presence>, remote: Vec<presence::Remote>, now: i64, alive: impl Fn(u32) -> bool, paused: impl Fn(&Presence) -> bool) -> Vec<Drone> {
97 let mut all: Vec<(String, Presence, State, bool)> = local
98 .into_iter()
99 .map(|p| {
100 let s = presence::state(&p, now, &alive);
101 let z = s != State::Ended && paused(&p);
102 (String::new(), p, s, z)
103 })
104 .collect();
105 all.extend(remote.into_iter().map(|r| {
107 let s = presence::state(&r.p, now, |_| true);
108 (r.key, r.p, s, false)
109 }));
110 all.sort_by(|a, b| (a.1.started, &a.0, &a.1.session).cmp(&(b.1.started, &b.0, &b.1.session)));
111 let n = all.len();
112 all.into_iter()
113 .enumerate()
114 .map(|(i, (node, p, state, paused))| Drone { designation: format!("{} of {}", word(i + 1), word(n)), number: i + 1, node, state, paused, p })
115 .collect()
116}
117
118pub fn drones(home: &Path, may_sync: bool, now: i64) -> Vec<Drone> {
120 let remote = if may_sync { presence::remote(home, now) } else { Vec::new() };
121 assemble(presence::local(home, now), remote, now, presence::alive, |p| steer::paused(home, &p.session, Some(p.started), now))
122}
123
124pub fn inbox(home: &Path, drones: &[Drone], now: i64) -> Inbox {
127 let listen = listen_of(home, drones);
128 if std::fs::metadata(msg::path(home)).is_err() {
129 return Inbox { listen, ..Inbox::default() };
130 }
131 let all: Vec<Message> = msg::read(home).into_iter().filter(|m| now - m.ms <= msg::TTL_MS).collect();
132 let here = msg::node(home);
133 let unread = unread_of(home, drones, &here, &all, now);
134 let day = all.iter().filter(|m| now - m.ms <= DAY_MS).count();
135 let mut recent = all;
136 recent.sort_by(|a, b| (b.ms, &b.id).cmp(&(a.ms, &a.id)));
137 recent.truncate(RECENT);
138 Inbox { unread, recent, day, here, listen }
139}
140
141pub fn acknowledge(home: &Path, cwd: &str, shown: &[&Message], done: &mut HashSet<String>, now: i64) {
150 let to_engineer: Vec<&Message> =
151 shown.iter().copied().filter(|m| matches!(Address::parse(&m.to), Some(Address::Engineer { .. })) && !done.contains(&m.id)).collect();
152 if to_engineer.is_empty() {
153 return;
154 }
155 let _ = msg::ack_engineer(home, cwd, &to_engineer, now);
156 done.extend(to_engineer.iter().map(|m| m.id.clone()));
157}
158
159pub fn engineer_looks<'a>(caller: impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String>, mut drones: impl Iterator<Item = &'a [Drone]>) -> bool {
169 drones.all(|ds| matches!(caller(ds), Ok(None)))
170}
171
172pub(crate) fn engineer_dir(dir: Option<&str>) -> String {
175 let d = dir.map_or_else(|| std::env::current_dir().unwrap_or_default(), std::path::PathBuf::from);
176 std::path::absolute(&d).unwrap_or(d).display().to_string()
177}
178
179fn listen_of(home: &Path, drones: &[Drone]) -> BTreeMap<String, crate::listen::Status> {
181 drones
182 .iter()
183 .filter(|d| d.node.is_empty() && d.state != State::Ended)
184 .filter_map(|d| crate::listen::status(home, &d.p.session).map(|s| (d.p.session.clone(), s)))
185 .collect()
186}
187
188pub fn wake_cell(inbox: &Inbox, d: &Drone) -> &'static str {
191 inbox.listen.get(&d.p.session).map_or("-", |s| s.name())
192}
193
194pub fn unread_of(home: &Path, drones: &[Drone], here: &str, all: &[Message], now: i64) -> BTreeMap<String, usize> {
197 drones
198 .iter()
199 .filter(|d| d.node.is_empty() && d.state != State::Ended)
200 .map(|d| (d.p.session.clone(), msg::unread(home, all, here, &d.p.session, d.p.started, now)))
201 .collect()
202}
203
204fn drone_at<'a>(drones: &'a [Drone], here: &str, node: &str, session: &str) -> Option<&'a Drone> {
206 drones.iter().find(|d| d.p.session == session && if d.node.is_empty() { node == here } else { d.node == node })
207}
208
209pub fn sender(drones: &[Drone], here: &str, m: &Message) -> String {
214 let name = match Address::parse(&m.by) {
215 Some(Address::Engineer { email }) if email.is_empty() => msg::ENGINEER.into(),
216 Some(Address::Engineer { email }) => collective::aliases().display(&email),
217 _ => match drone_at(drones, here, &m.node, &m.by) {
218 Some(d) => d.designation.clone(),
219 None => msg::sender_label(m),
220 },
221 };
222 if m.node == here { name } else { format!("{name} @{}", msg::short_node(&m.node)) }
223}
224
225pub fn recipient(drones: &[Drone], here: &str, to: &str) -> String {
230 match Address::parse(to) {
231 Some(Address::Engineer { email }) if !email.is_empty() => collective::aliases().display(&email),
232 Some(Address::Local { node }) if node == here => msg::LOCAL.into(),
233 Some(Address::Local { node }) => format!("{} @{}", msg::LOCAL, msg::short_node(&node)),
234 Some(Address::All) => msg::COLLECTIVE.into(),
235 Some(Address::Session { id, node }) => {
236 let d = match node {
237 Some(n) => drone_at(drones, here, &n, &id),
238 None => drones.iter().find(|d| d.p.session == id),
239 };
240 d.map_or_else(|| msg::short_session(&id), |d| d.designation.clone())
241 }
242 Some(a) => a.to_string(),
243 None => msg::short_session(to),
244 }
245}
246
247pub fn message_line(drones: &[Drone], here: &str, m: &Message) -> String {
250 let ask = if m.asks() { " asks" } else { "" };
251 format!("{}{ask} -> {}: {}", sender(drones, here, m), recipient(drones, here, &m.to), m.text)
252}
253
254pub fn unread_cell(inbox: &Inbox, d: &Drone) -> String {
256 inbox.unread.get(&d.p.session).map_or_else(|| "-".into(), usize::to_string)
257}
258
259pub fn resolve<'a>(drones: &'a [Drone], target: &str) -> Result<Option<&'a Drone>, String> {
265 let t = target.trim();
266 if [msg::ALL, msg::COLLECTIVE, msg::LOCAL].iter().any(|w| t.eq_ignore_ascii_case(w)) {
267 return Ok(None);
268 }
269 let first = t.split_whitespace().next().unwrap_or_default();
270 let number = first.parse::<usize>().ok().or_else(|| NUMBERS.iter().position(|w| w.eq_ignore_ascii_case(first)).map(|i| i + 1));
271 if let Some(n) = number {
272 return drones.iter().find(|d| d.number == n).map(Some).ok_or_else(|| format!("no drone {target}: there are {} (`borg top`)", drones.len()));
273 }
274 let mut hits: Vec<&Drone> = drones.iter().filter(|d| d.p.session == t || (t.len() >= 4 && d.p.session.starts_with(t))).collect();
275 if hits.iter().any(|d| d.state != State::Ended) {
276 hits.retain(|d| d.state != State::Ended);
277 }
278 if hits.iter().any(|d| d.node.is_empty()) {
279 hits.retain(|d| d.node.is_empty());
280 }
281 match hits.as_slice() {
282 [d] => Ok(Some(d)),
283 [] => Err(format!("no drone or session {target:?} (`borg top`)")),
284 _ => Err(format!("{target:?} names {} sessions: give more of the id", hits.len())),
285 }
286}
287
288pub const COLUMN_NODE_CHARS: usize = 6;
292
293pub fn column_node(node: &str) -> String {
295 node.chars().take(COLUMN_NODE_CHARS).collect()
296}
297
298const MESSAGE_CHARS: usize = 120;
301
302fn cut(s: &str, n: usize) -> String {
303 if s.chars().count() <= n { s.to_string() } else { format!("{}…", s.chars().take(n - 1).collect::<String>()) }
304}
305
306pub fn shows_last(total: usize, last: usize, width: usize) -> bool {
311 total <= width || total.saturating_sub(last) < width.saturating_sub(1)
312}
313
314fn prints_text(drones: &[Drone], here: &str, m: &Message) -> bool {
318 shows_last(message_line(drones, here, m).chars().count(), m.text.chars().count(), MESSAGE_CHARS)
319}
320
321pub fn render(name: &str, share: Share, drones: &[Drone], inbox: &Inbox, now: i64) -> String {
323 let working = drones.iter().filter(|d| matches!(d.state, State::Working | State::Waiting)).count();
324 let remote = drones.iter().filter(|d| !d.node.is_empty()).count();
325 let share = match share {
326 Share::Team => "team",
327 Share::Local => "local, never leaves this machine",
328 };
329 let mut s = format!("{name} ({share}): {} drone(s), {working} active, {remote} remote\n", drones.len());
330 if !drones.is_empty() {
331 s.push_str(&drone_table(drones, inbox, now));
332 }
333 if !inbox.recent.is_empty() {
334 s.push_str(&format!(" messages: {} in the last day, {} unread\n", inbox.day, inbox.unread.values().sum::<usize>()));
335 for m in &inbox.recent {
336 s.push_str(&format!(" {:>6} {}\n", ago(now - m.ms), cut(&message_line(drones, &inbox.here, m), MESSAGE_CHARS)));
337 }
338 }
339 s
340}
341
342fn drone_table(drones: &[Drone], inbox: &Inbox, now: i64) -> String {
343 let mut s = format!(
344 " {:<18} {:<8} {:<24} {:<14} {:<8} {:<6} {:>6} {:>6} {:<40} {}\n",
345 "DRONE", "SESSION", "ENGINEER", "REPO", "STATE", "WAKE", "IDLE", "UNREAD", "DOING", "FILES"
346 );
347 for d in drones {
348 let state = if d.paused { "paused" } else { d.state.name() };
349 let engineer = if d.node.is_empty() { d.p.engineer.clone() } else { format!("{} @{}", d.p.engineer, column_node(&d.node)) };
350 let line = format!(
351 " {:<18} {:<8} {:<24} {:<14} {:<8} {:<6} {:>6} {:>6} {:<40} {}",
352 d.designation,
353 cut(&d.p.session, 8),
354 cut(if engineer.is_empty() { "?" } else { &engineer }, 24),
355 cut(&d.p.repo, 14),
356 state,
357 wake_cell(inbox, d),
358 ago(now - d.p.ms),
359 unread_cell(inbox, d),
360 cut(&d.p.prompt, 40),
361 d.p.files.join(", ")
362 );
363 s.push_str(line.trim_end());
364 s.push('\n');
365 }
366 s
367}
368
369pub fn to_json(name: &str, share: Share, drones: &[Drone]) -> Value {
370 json!({"collective": name, "share": if share == Share::Team { "team" } else { "local" }, "drones": drones})
371}
372
373pub fn to_json_with(name: &str, share: Share, drones: &[Drone], inbox: &Inbox) -> Value {
375 let mut v = to_json(name, share, drones);
376 if let Some(ds) = v["drones"].as_array_mut() {
377 for (o, d) in ds.iter_mut().zip(drones) {
378 if let Some(n) = inbox.unread.get(&d.p.session) {
379 o["unread"] = json!(n);
380 }
381 if let Some(l) = inbox.listen.get(&d.p.session) {
382 o["listen"] = json!(l);
383 }
384 }
385 }
386 v["messages"] = json!({"day": inbox.day, "recent": inbox.recent});
387 v
388}
389
390#[derive(Debug, Clone, Default, PartialEq, Serialize)]
400pub struct Selection {
401 pub policy: &'static str,
404 pub default: &'static str,
407 pub source: &'static str,
410 pub gate: f64,
415 pub fit: String,
417 pub at: u64,
419 pub checks: u64,
421 pub sessions: u64,
422 pub deviations: Vec<Count>,
427 pub fallbacks: Vec<Count>,
431 pub spend: Vec<Spend>,
436 pub budget: Option<Budget>,
438 pub effects: Vec<Estimate>,
441 pub half_life: Option<u32>,
443 pub labels: Vec<Count>,
445 pub declaration: Option<Estimate>,
447 pub delivery: Option<select::policy::Rates>,
450 pub evaluation: Option<Evaluation>,
452 pub measurement: Option<Measurement>,
455 pub pool: Vec<Yield>,
457 pub latency: Vec<Latency>,
459 pub test: Option<Test>,
462}
463
464#[derive(Debug, Clone, Default, PartialEq, Serialize)]
469pub struct Test {
470 pub risk: f64,
473 pub set: bool,
475 pub ran: u64,
478 pub control: u64,
479 pub refused: u64,
482 pub sessions: u64,
484 pub kept: u64,
487 pub by_k: Vec<Draws>,
489 pub due: bool,
495 pub due_checks: u64,
496 pub due_sessions: u64,
497 pub dilution: Vec<Estimate>,
498 pub ask_free: Option<Estimate>,
499}
500
501#[derive(Debug, Clone, Default, PartialEq, Serialize)]
503pub struct Draws {
504 pub k: u32,
505 pub draws: Vec<u64>,
506 pub refused: u64,
507}
508
509impl Test {
510 #[must_use]
513 pub fn of(summary: &select::summary::Summary, risk: Option<f64>) -> Test {
514 use select::replay::{TEST_CHECKS, TEST_SESSIONS, test_readout};
515 let e = &summary.experiment;
516 let (ran, refused) = e.totals();
517 let r = test_readout(summary);
518 let estimate = |name: String, unit: &str, x: &crate::ope::Estimate| {
520 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 })
521 };
522 let mut by_k: Vec<Draws> = Vec::new();
523 let at = |by_k: &mut Vec<Draws>, k: u32| match by_k.iter().position(|d| d.k == k) {
524 Some(i) => i,
525 None => {
526 by_k.push(Draws { k, draws: vec![0; k as usize + 1], refused: 0 });
527 by_k.len() - 1
528 }
529 };
530 for &(k, s, n) in &e.ran {
531 let i = at(&mut by_k, k);
532 if let Some(c) = by_k[i].draws.get_mut(s as usize) {
533 *c += n;
534 }
535 }
536 for &(k, n) in &e.refused {
537 let i = at(&mut by_k, k);
538 by_k[i].refused += n;
539 }
540 by_k.sort_by_key(|d| d.k);
541 Test {
542 risk: risk.unwrap_or(select::policy::ALPHA),
543 set: risk.is_some(),
544 ran,
545 control: e.control,
546 refused,
547 sessions: u64::from(e.sessions),
548 kept: r.kept,
549 by_k,
550 due: r.due,
551 due_checks: TEST_CHECKS,
552 due_sessions: u64::from(TEST_SESSIONS),
553 dilution: r.by_asked.iter().filter_map(|(m, x)| estimate(format!("{m} asked − 1 asked"), "points changed", x)).collect(),
554 ask_free: r.ask_free.as_ref().and_then(|x| estimate("grade 2 asked − handed over".into(), "points changed", x)),
555 }
556 }
557}
558
559#[derive(Debug, Clone, Default, PartialEq, Serialize)]
561pub struct Count {
562 pub name: String,
563 pub part: u64,
564 pub whole: u64,
565}
566
567#[derive(Debug, Clone, Default, PartialEq, Serialize)]
569pub struct Spend {
570 pub source: String,
572 pub calls: u64,
573 pub rounds: u64,
575 pub usd: f64,
576}
577
578#[derive(Debug, Clone, Default, PartialEq, Serialize)]
582pub struct Budget {
583 pub tokens: f64,
584 pub spent: f64,
585 pub price: f64,
586}
587
588#[derive(Debug, Clone, Default, PartialEq, Serialize)]
590pub struct Estimate {
591 pub name: String,
592 pub unit: String,
597 pub mean: f64,
598 pub lo: f64,
599 pub hi: f64,
600 pub n: u64,
602}
603
604#[derive(Debug, Clone, Default, PartialEq, Serialize)]
607pub struct Evaluation {
608 pub sessions: u64,
610 pub mean: f64,
613 pub ucb: Option<f64>,
614 pub units: u64,
616 pub ess: f64,
617 pub max_weight: f64,
619 pub truncated: u64,
621}
622
623#[derive(Debug, Clone, Default, PartialEq, Serialize)]
628pub struct Measurement {
629 pub placebos: u64,
631 pub judged: u64,
633 pub tail_shown: u64,
636 pub tail_applied: u64,
637}
638
639#[derive(Debug, Clone, Default, PartialEq, Serialize)]
642pub struct Yield {
643 pub from: u32,
644 pub to: u32,
645 pub shown: u64,
646 pub applied: u64,
647 pub rate: f64,
648}
649
650#[derive(Debug, Clone, Default, PartialEq, Serialize)]
653pub struct Latency {
654 pub harness: String,
655 pub checks: u64,
656 pub p50_ms: u64,
658 pub p95_ms: u64,
659 pub budget_ms: Option<u64>,
661 pub over: u64,
664}
665
666const QUANTILES: [f64; 2] = [0.50, 0.95];
669
670fn quantile(sorted: &[u64], q: f64) -> u64 {
672 let rank = ((q * sorted.len() as f64).ceil() as usize).clamp(1, sorted.len().max(1));
673 sorted.get(rank - 1).copied().unwrap_or(0)
674}
675
676impl Selection {
677 #[must_use]
681 pub fn of(summary: &select::summary::Summary, policy: select::Policy, risk: Option<f64>) -> Selection {
682 use select::summary::{LIVE, Label, STOP};
683 let s = summary;
684 let learned = policy == select::Policy::Learned;
685 let reverted = learned && s.monitor.reverted;
686 let default = select::precedence(None, None, false).name();
687 let count = |name: &str, part: usize, whole: usize| Count { name: name.into(), part: part as u64, whole: whole as u64 };
688 let (sessions, checks) = s.checked();
690 let deviations = match s.gate.checks {
691 0 => Vec::new(),
692 n => vec![count("route", s.gate.deviated as usize, n as usize)],
693 };
694 let fallbacks = select::summary::FELL_BACK
696 .iter()
697 .zip(s.fell_back)
698 .filter(|_| checks > 0)
699 .map(|(reason, n)| Count { name: (*reason).into(), part: n, whole: checks })
700 .collect();
701 let spend = [("stop", STOP), ("live", LIVE)]
702 .into_iter()
703 .filter(|(_, i)| s.spend[*i].calls > 0)
704 .map(|(source, i)| Spend { source: source.into(), calls: s.spend[i].calls, rounds: s.spend[i].rounds, usd: s.spend[i].usd })
705 .collect();
706 let budget = s.fit.budget.map(|tokens| Budget { tokens, spent: s.pool.tokens as f64 / sessions.max(1) as f64, price: s.fit.lambda });
707 let judged = |l: Label| matches!(l, Label::Changed | Label::Ignored);
708 let of = |route: select::Route| {
714 let of: Vec<_> = s.applies.iter().filter(|a| a.route == route && (route != select::Route::Free || !a.had)).collect();
715 (of.iter().filter(|a| judged(a.label)).count(), of.len())
716 };
717 let mut labels = Vec::new();
718 for (name, route) in [("judged, asked", select::Route::Ask), ("judged, free", select::Route::Free), ("judged, placebo", select::Route::Placebo)] {
719 let (part, whole) = of(route);
720 labels.push(count(name, part, whole));
721 }
722 let verdicts = s.applies.iter().filter(|a| judged(a.label) || a.label == Label::Unclear).count();
723 labels.push(count("unclear", s.applies.iter().filter(|a| a.label == Label::Unclear).count(), verdicts));
724 let later = s.applies.iter().filter(|a| matches!(a.route, select::Route::Not | select::Route::Placebo)).count();
725 let (placebos_judged, placebos) = of(select::Route::Placebo);
726 labels.push(count("placebo", placebos, later));
727 labels.retain(|c| c.whole > 0);
728 let measurement = (checks > 0).then_some(Measurement {
729 placebos: placebos as u64,
730 judged: placebos_judged as u64,
731 tail_shown: s.pool.tail.0,
732 tail_applied: s.pool.tail.1,
733 });
734 let share = |stripped: bool| {
736 let of: Vec<_> = s.applies.iter().filter(|a| a.route == select::Route::Ask && judged(a.label) && a.stripped == stripped).collect();
737 (of.iter().filter(|a| a.label == Label::Changed).count() as f64, of.len() as f64)
738 };
739 let ((c1, n1), (c0, n0)) = (share(false), share(true));
740 let declaration = (n1 > 0.0 && n0 > 0.0).then(|| {
741 let (r1, r0) = (c1 / n1, c0 / n0);
742 let se = (r1 * (1.0 - r1) / n1 + r0 * (1.0 - r0) / n0).sqrt();
743 let mean = 100.0 * (r1 - r0);
744 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 }
745 });
746 let pool = s
747 .pool
748 .bins
749 .iter()
750 .enumerate()
751 .filter(|(_, (shown, _))| *shown > 0)
752 .map(|(b, &(shown, applied))| Yield {
753 from: (1u32 << b) - 1,
754 to: (1u32 << (b + 1)) - 1,
755 shown,
756 applied,
757 rate: s.fit.pool.yields.get(b).map_or(applied as f64 / shown as f64, |(a, b)| a / (a + b)),
758 })
759 .collect();
760 let mon = &s.monitor;
761 let evaluation = (mon.labelled > 0).then(|| Evaluation {
762 sessions: u64::from(mon.sessions),
763 mean: mon.mean,
764 ucb: mon.ucb,
765 units: u64::from(mon.labelled),
766 ess: mon.ess,
767 max_weight: mon.max_weight,
768 truncated: u64::from(mon.truncated),
769 });
770 let latency = s
771 .pool
772 .stops
773 .iter()
774 .filter(|(_, calls)| !calls.is_empty())
775 .map(|(harness, calls)| {
776 let mut ms: Vec<u64> = calls.iter().map(|&(ms, _)| u64::from(ms)).collect();
777 ms.sort_unstable();
778 let budget = crate::harness::ALL.iter().find(|h| h.name() == harness).and_then(|h| h.caps().stop_budget);
779 Latency {
780 harness: harness.clone(),
781 checks: ms.len() as u64,
782 p50_ms: quantile(&ms, QUANTILES[0]),
783 p95_ms: quantile(&ms, QUANTILES[1]),
784 budget_ms: budget.map(|b| u64::try_from(b.as_millis()).unwrap_or(u64::MAX)),
785 over: calls.iter().filter(|(_, failed)| *failed).count() as u64,
786 }
787 })
788 .collect();
789 Selection {
790 policy: policy.name(),
791 default,
792 source: if reverted { select::REVERTED_SOURCE } else { policy.name() },
793 gate: if learned && !reverted { select::policy::remaining_at(&s.gate, risk.unwrap_or(select::policy::ALPHA)) } else { 0.0 },
794 fit: s.fit.key.clone(),
795 at: s.at,
796 checks,
797 sessions,
798 deviations,
799 fallbacks,
800 spend,
801 budget,
802 effects: s.fit.model.as_ref().map(|m| effects(m, s.fit.labels)).unwrap_or_default(),
803 half_life: s.fit.model.as_ref().map(|_| s.fit.half_life.unwrap_or(0)),
804 labels,
805 declaration,
806 delivery: (!s.fit.key.is_empty()).then(|| s.fit.rates.clone()),
807 evaluation,
808 measurement,
809 pool,
810 latency,
811 test: (learned || s.experiment.totals() != (0, 0)).then(|| Test::of(s, risk)),
812 }
813 }
814
815 #[must_use]
818 pub fn shown(&self) -> bool {
819 self.checks > 0 || self.source != self.default
820 }
821}
822
823const Z95: f64 = 1.96;
825
826fn effects(m: &select::policy::Model, n: u32) -> Vec<Estimate> {
830 let p = select::policy::P;
831 if m.beta.len() != p || m.chol.len() != p * p {
832 return Vec::new();
833 }
834 let sd = |k: usize| m.chol[k * p..=k * p + k].iter().map(|x| x * x).sum::<f64>().sqrt();
836 let mut out: Vec<Estimate> = (0..p)
837 .map(|k| Estimate {
838 name: select::policy::NAMES[k].into(),
839 unit: "coefficient".into(),
840 mean: m.beta[k],
841 lo: m.beta[k] - Z95 * sd(k),
842 hi: m.beta[k] + Z95 * sd(k),
843 n: u64::from(n),
844 })
845 .collect();
846 for (name, v) in ["session spread", "lesson spread", "judge spread"].into_iter().zip(m.spread) {
847 out.push(Estimate { name: name.into(), unit: "logit".into(), mean: v, lo: v, hi: v, n: u64::from(n) });
848 }
849 out
850}
851
852#[must_use]
854pub fn selection(lessons: &Path, policy: select::Policy) -> Selection {
855 Selection::of(&select::summary::load(lessons), policy, select::risk_of(lessons))
856}
857
858#[must_use]
867pub fn selection_line(s: &Selection) -> String {
868 let default = s.source == s.default;
869 let head = match s.source {
870 select::REVERTED_SOURCE => "REVERTED to the fixed rule: the monitor saw harm beyond the gate's share (sticky)".to_string(),
871 "learned" => {
872 let opt = if default { "" } else { " (opt-in, BORG_SELECT=learned)" };
873 if s.gate > 0.0 {
874 format!("learned{opt}, gate {:.3} (quality it may still risk below the fixed rule)", s.gate)
875 } else {
876 format!("learned{opt}, cold start: gate 0, every decision the fixed rule's")
877 }
878 }
879 "fixed" if default => "the fixed rule (the default; BORG_SELECT=learned opts in to the learned policy)".into(),
880 "fixed" => "the fixed rule (BORG_SELECT=fixed)".into(),
881 "cap" => "the fixed rule with an ask cap (BORG_STOP_ASK_CAP, evaluation only)".into(),
882 "attention" => "attention's cut (BORG_STOP_ATTENTION=on, evaluation only)".into(),
883 other => other.to_string(),
884 };
885 if s.checks == 0 {
886 return format!("{head} · no graded check folded yet");
887 }
888 let mut parts = vec![head, format!("{} checks", s.checks)];
889 if !s.deviations.is_empty() {
890 parts.push(format!("deviated: {}", s.deviations.iter().map(|c| format!("{} {}/{}", c.name, c.part, c.whole)).collect::<Vec<_>>().join(", ")));
891 }
892 let fell: Vec<String> = s.fallbacks.iter().filter(|c| c.part > 0).map(|c| format!("{} {}", c.name, c.part)).collect();
893 if !fell.is_empty() {
894 parts.push(format!("learned decisions the fixed rule took over: {}", fell.join(", ")));
895 }
896 if let Some(t) = &s.test {
897 parts.push(test_line(t));
898 }
899 if let Some(m) = &s.measurement {
900 parts.push(format!("placebos {} drawn, {} judged", m.placebos, m.judged));
901 if m.tail_shown > 0 {
902 parts.push(format!("tail {} shown past the pool, {} applied", m.tail_shown, m.tail_applied));
903 }
904 }
905 if let Some(e) = &s.evaluation {
906 let reading = match e.ucb {
907 Some(u) => format!("monitor {:+.3}, bound {u:+.3} over {} sessions (below 0 reverts)", e.mean, e.sessions),
908 None => format!("monitor {:+.3} over {} sessions, no bound yet", e.mean, e.sessions),
909 };
910 parts.push(format!("{reading}, ESS {:.0} of {} weighed, largest weight {:.1}× the mean", e.ess, e.units, e.max_weight));
911 }
912 if !s.latency.is_empty() {
913 let secs = |ms: u64| format!("{:.1} s", ms as f64 / 1000.0);
914 let each: Vec<String> = s
915 .latency
916 .iter()
917 .map(|l| {
918 let budget = l.budget_ms.map_or_else(String::new, |b| format!(" of {}", secs(b)));
919 let failed = if l.over > 0 { format!(", {} failed", l.over) } else { String::new() };
920 format!("{} {}{budget}{failed}", l.harness, secs(l.p95_ms))
921 })
922 .collect();
923 parts.push(format!("stop p95 {}", each.join("; ")));
924 }
925 if s.sessions > 0 && !s.spend.is_empty() {
926 let n = s.sessions as f64;
927 let (usd, rounds) = s.spend.iter().fold((0.0, 0), |(u, r), x| (u + x.usd, r + x.rounds));
928 parts.push(format!("${:.2} and {:.1} extra rounds per session ({} sessions)", usd / n, rounds as f64 / n, s.sessions));
929 }
930 if let Some(b) = &s.budget {
931 parts.push(format!("budget {:.0} pool tokens per session: {:.0} read, price {:.2e} per token", b.tokens, b.spent, b.price));
932 }
933 parts.join(" · ")
934}
935
936fn test_line(t: &Test) -> String {
941 let risk = if t.set { format!("risk {:.2} (the collective's)", t.risk) } else { format!("risk {:.2}", t.risk) };
942 match (t.ran, t.risk <= 0.0) {
943 (0, true) => return "question-count test off (risk 0)".into(),
944 (0, false) => return format!("question-count test: no check yet, {risk}"),
945 _ => {}
946 }
947 let kept = if t.kept < t.ran { format!(", {} still in the summary", t.kept) } else { String::new() };
948 let head =
949 format!("question-count test {} drawn ({} control, {} refused by the gate{kept}) in {} sessions, {risk}", t.ran, t.control, t.refused, t.sessions);
950 match (&t.ask_free, t.due) {
951 (Some(e), true) => format!("{head}; readout: grade 2 asked − handed over {:+.1} [{:+.1}, {:+.1}] points changed", e.mean, e.lo, e.hi),
952 (None, true) => format!("{head}; readout due, no judged grade 2 on both routes yet"),
953 (_, false) => format!("{head}; readout after {} checks or {} sessions", t.due_checks, t.due_sessions),
954 }
955}
956
957#[must_use]
959pub fn render_with(name: &str, share: Share, drones: &[Drone], inbox: &Inbox, select: Option<&Selection>, now: i64) -> String {
960 let mut s = render(name, share, drones, inbox, now);
961 if let Some(sel) = select.filter(|s| s.shown()) {
962 s.push_str(&format!(" selection: {}\n", selection_line(sel)));
963 }
964 s
965}
966
967fn routes(dir: Option<&str>) -> Result<Vec<Route>, String> {
969 let cfg = collective::load()?;
970 let base = learn::home();
971 Ok(match dir {
972 Some(d) => vec![collective::route(cfg.as_ref(), d, &base)].into_iter().filter(|r| *r != Route::Off).collect(),
973 None => install::collectives(cfg.as_ref(), &base),
974 })
975}
976
977pub fn gather(dir: Option<&str>, now: i64) -> Result<Vec<crate::top_live::Coll>, String> {
979 let mut out = Vec::new();
980 for r in routes(dir)? {
981 let Route::Collective { name, home, share } = &r else { continue };
982 presence::prune(home, now);
983 let mut ds = drones(home, collective::may_sync(&r), now);
984 for d in &mut ds {
985 d.p.engineer = collective::aliases().display(&d.p.engineer);
986 }
987 let inbox = inbox(home, &ds, now);
988 if ds.is_empty() && inbox.recent.is_empty() && dir.is_none() {
989 continue;
990 }
991 let lessons = home.join("lessons.jsonl");
994 let summary = select::summary::peek(&lessons);
995 let select = Some(Selection::of(&summary, select::policy(), select::risk_of(&lessons))).filter(Selection::shown).map(|s| selection_line(&s));
996 out.push(crate::top_live::Coll { name: name.clone(), home: home.clone(), share: *share, drones: ds, inbox, select });
997 }
998 Ok(out)
999}
1000
1001fn screen(dir: Option<&str>, json_out: bool, now: i64) -> Result<String, String> {
1002 Ok(screen_of(&routes(dir)?, dir.is_some(), &engineer_dir(dir), msg::caller, json_out, now))
1003}
1004
1005fn screen_of(
1010 routes: &[Route],
1011 one: bool,
1012 cwd: &str,
1013 caller: impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String>,
1014 json_out: bool,
1015 now: i64,
1016) -> String {
1017 let mut text = String::new();
1018 let mut all = Vec::new();
1019 let (mut gathered, mut printed) = (Vec::new(), Vec::new());
1022 for r in routes {
1023 let Route::Collective { name, home, share } = r else { continue };
1024 presence::prune(home, now);
1025 let mut ds = drones(home, collective::may_sync(r), now);
1026 for d in &mut ds {
1028 d.p.engineer = collective::aliases().display(&d.p.engineer);
1029 }
1030 let inbox = inbox(home, &ds, now);
1031 let sel = selection(&home.join("lessons.jsonl"), select::policy());
1032 if json_out {
1033 let mut v = to_json_with(name, *share, &ds, &inbox);
1034 v["select"] = json!(sel);
1035 all.push(v);
1036 } else if !ds.is_empty() || !inbox.recent.is_empty() || one {
1037 text.push_str(&render_with(name, *share, &ds, &inbox, Some(&sel), now));
1038 text.push('\n');
1039 printed.push((home, inbox.recent.into_iter().filter(|m| prints_text(&ds, &inbox.here, m)).collect::<Vec<_>>()));
1040 }
1041 gathered.push(ds);
1042 }
1043 if !printed.is_empty() && engineer_looks(caller, gathered.iter().map(Vec::as_slice)) {
1044 let mut done = HashSet::new();
1045 for (home, recent) in &printed {
1046 acknowledge(home, cwd, &recent.iter().collect::<Vec<_>>(), &mut done, now);
1047 }
1048 }
1049 if json_out {
1050 return serde_json::to_string_pretty(&all).unwrap_or_default() + "\n";
1051 }
1052 if text.is_empty() {
1053 text = "no drones (borg top lists sessions whose hooks ran in the last day)\n".into();
1054 }
1055 text
1056}
1057
1058pub fn main(args: &[&str]) -> i32 {
1059 let (mut dir, mut once, mut json_out) = (None, false, false);
1060 let mut it = args.iter();
1061 while let Some(a) = it.next() {
1062 match *a {
1063 "--dir" if dir.is_none() => match it.next() {
1064 Some(d) => dir = Some(*d),
1065 None => return usage("--dir needs a value"),
1066 },
1067 "--once" => once = true,
1068 "--json" => json_out = true,
1069 a => return usage(&format!("unknown flag {a}")),
1070 }
1071 }
1072 if !json_out && !once && unsafe { libc::isatty(1) } == 1 {
1074 return crate::top_live::run(dir);
1075 }
1076 match screen(dir, json_out, learn::now_ms()) {
1077 Ok(s) => {
1078 print!("{s}");
1079 0
1080 }
1081 Err(e) => {
1082 eprintln!("borg top: {e}");
1083 1
1084 }
1085 }
1086}
1087
1088fn usage(why: &str) -> i32 {
1089 eprintln!("borg top: {why}\nusage: borg top [--once] [--json] [--dir DIR]");
1090 2
1091}
1092
1093#[cfg(test)]
1094pub(crate) fn tests_fixture() -> Vec<Drone> {
1095 tests::fixture()
1096}
1097
1098#[cfg(test)]
1099pub(crate) fn tests_inbox() -> Inbox {
1100 tests::inbox_fixture()
1101}
1102
1103#[cfg(test)]
1104mod tests {
1105 use super::*;
1106
1107 fn p(session: &str, engineer: &str, started: i64, ms: i64, event: &str, prompt: &str, files: &[&str]) -> Presence {
1108 Presence {
1109 session: session.into(),
1110 harness: "claude".into(),
1111 engineer: engineer.into(),
1112 repo: "borg-ai".into(),
1113 collective: "corp".into(),
1114 prompt: prompt.into(),
1115 files: files.iter().map(|f| f.to_string()).collect(),
1116 event: event.into(),
1117 ms,
1118 started,
1119 pid: None,
1120 }
1121 }
1122
1123 const NOW: i64 = 10_000_000;
1124
1125 pub(crate) fn fixture() -> Vec<Drone> {
1126 let local = vec![
1127 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"]),
1128 p("cccc3333-x", "alice@x", 300, NOW - presence::IDLE_MS - 60_000, "stop", "write the release notes for the mesh daemon and more words", &[]),
1129 ];
1130 let remote = vec![presence::Remote {
1131 p: p("bbbb2222-y", "bob@y", 200, NOW - 120_000, "stop", "review PR 42", &["README.md"]),
1132 key: "beefcafe".repeat(8),
1133 sig: String::new(),
1134 }];
1135 assemble(local, remote, NOW, |_| true, |p| p.session.starts_with("aaaa"))
1136 }
1137
1138 fn message(by: &str, name: &str, to: &str, kind: msg::Kind, ms: i64, text: &str) -> Message {
1139 message_on("n1", by, name, to, kind, ms, text)
1140 }
1141
1142 fn message_on(node: &str, by: &str, name: &str, to: &str, kind: msg::Kind, ms: i64, text: &str) -> Message {
1143 let from = msg::Sender { by: by.into(), name: name.into(), harness: "claude".into(), node: node.into() };
1144 Message::new(kind, &from, to, None, ms, text)
1145 }
1146
1147 pub(crate) fn inbox_fixture() -> Inbox {
1148 Inbox {
1149 unread: [("aaaa1111-x".to_string(), 2), ("cccc3333-x".to_string(), 0)].into_iter().collect(),
1150 recent: vec![
1151 message("cccc3333-x", "Three of Three", "aaaa1111-x", msg::ASK, NOW - 60_000, "which test\nflakes?"),
1152 message("engineer:alice@x", "alice@x", msg::ALL, msg::TELL, NOW - 120_000, "release at five"),
1153 message_on(&"feed".repeat(16), "gone-session", "Six of Six", "zzzz9999-q", msg::TELL, NOW - 180_000, "done"),
1154 ],
1155 day: 3,
1156 here: "n1".into(),
1157 listen: [("aaaa1111-x".to_string(), crate::listen::Status::Waits), ("cccc3333-x".to_string(), crate::listen::Status::Listens)]
1158 .into_iter()
1159 .collect(),
1160 }
1161 }
1162
1163 #[test]
1164 fn renders_one_screen_per_collective() {
1165 let out = render("corp", Share::Team, &fixture(), &inbox_fixture(), NOW);
1166 let expected = "\
1167corp (team): 3 drone(s), 2 active, 1 remote
1168 DRONE SESSION ENGINEER REPO STATE WAKE IDLE UNREAD DOING FILES
1169 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
1170 Two of Three bbbb222… bob@y @beefca borg-ai waiting - 2m - review PR 42 README.md
1171 Three of Three cccc333… alice@x borg-ai idle on 31m 0 write the release notes for the mesh da…
1172 messages: 3 in the last day, 2 unread
1173 1m Three of Three asks -> One of Three: which test flakes?
1174 2m alice@x -> collective: release at five
1175 3m Six of Six @feedfeedfeed -> zzzz9999: done
1176";
1177 assert_eq!(out, expected, "\n{out}");
1178 let none = Inbox::default();
1179 assert_eq!(render("private", Share::Local, &[], &none, NOW), "private (local, never leaves this machine): 0 drone(s), 0 active, 0 remote\n");
1180 }
1181
1182 fn learned() -> Selection {
1186 let spend = |source: &str, rounds, usd| Spend { source: source.into(), calls: 1, rounds, usd };
1187 let evaluation = Evaluation { sessions: 12, mean: 0.004, ucb: Some(0.031), units: 300, ess: 180.0, max_weight: 6.5, truncated: 3 };
1188 Selection {
1189 policy: "learned",
1190 default: "fixed",
1191 source: "learned",
1192 gate: 0.12,
1193 checks: 480,
1194 sessions: 12,
1195 deviations: vec![Count { name: "route".into(), part: 5, whole: 400 }],
1196 spend: vec![spend("stop", 4, 3.0), spend("live", 0, 2.04)],
1197 evaluation: Some(evaluation),
1198 measurement: Some(Measurement { placebos: 40, judged: 31, tail_shown: 12, tail_applied: 3 }),
1199 test: Some(Test { risk: 0.1, ran: 30, control: 9, refused: 4, sessions: 6, kept: 30, due_checks: 300, due_sessions: 30, ..Test::default() }),
1200 ..Selection::default()
1201 }
1202 }
1203
1204 #[test]
1205 fn the_selection_line_says_what_decides_and_prices_spend_only_under_a_budget() {
1206 use select::Policy;
1207 use select::summary::Summary;
1208 let fresh = |policy, default| Selection { default, ..Selection::of(&Summary::default(), policy, None) };
1209 let fixed = fresh(Policy::Fixed, "fixed");
1211 assert_eq!(selection_line(&fixed), "the fixed rule (the default; BORG_SELECT=learned opts in to the learned policy) · no graded check folded yet");
1212 assert!(!fixed.shown(), "the default with nothing folded: nothing to show on the drones screen");
1213 let opted = fresh(Policy::Learned, "fixed");
1214 assert_eq!(
1215 selection_line(&opted),
1216 "learned (opt-in, BORG_SELECT=learned), cold start: gate 0, every decision the fixed rule's · no graded check folded yet"
1217 );
1218 assert!(opted.shown(), "a switch decides: shown");
1219 assert_eq!(selection_line(&fresh(Policy::Fixed, "learned")), "the fixed rule (BORG_SELECT=fixed) · no graded check folded yet");
1221 assert!(!fresh(Policy::Learned, "learned").shown() && fresh(Policy::Fixed, "learned").shown());
1222 let d = select::precedence(None, None, false);
1223 assert_eq!(Selection::of(&Summary::default(), Policy::Cap(2), None).default, d.name(), "the default is select.rs's, no switch set");
1224 assert!(!Selection::of(&Summary::default(), d, None).shown());
1225 let mut s = learned();
1226 let line = selection_line(&s);
1227 assert_eq!(
1228 line,
1229 "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)"
1230 );
1231 assert!(!line.contains("budget") && !line.contains("price"), "no budget set: spend measured, never priced: {line}");
1232 let mut early = learned();
1233 early.evaluation = Some(Evaluation { ucb: None, sessions: 3, units: 40, ess: 12.4, max_weight: 3.0, ..Evaluation::default() });
1234 let line = selection_line(&early);
1235 assert!(line.contains(" · monitor +0.000 over 3 sessions, no bound yet, ESS 12 of 40 weighed, largest weight 3.0× the mean · "), "{line}");
1236 early.latency = vec![
1237 Latency { harness: "codex".into(), checks: 20, p50_ms: 10_000, p95_ms: 19_000, budget_ms: Some(30_000), over: 1 },
1238 Latency { harness: "claude".into(), checks: 1, p50_ms: 4_000, p95_ms: 4_000, budget_ms: None, over: 0 },
1239 ];
1240 let line = selection_line(&early);
1241 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}");
1242 s.budget = Some(Budget { tokens: 30_000.0, spent: 24_000.0, price: 0.0002 });
1243 assert!(selection_line(&s).ends_with(" · budget 30000 pool tokens per session: 24000 read, price 2.00e-4 per token"), "{}", selection_line(&s));
1244 s.source = select::REVERTED_SOURCE;
1245 assert!(selection_line(&s).starts_with("REVERTED to the fixed rule"), "{}", selection_line(&s));
1246 }
1247
1248 #[test]
1251 fn a_store_says_how_its_lessons_are_selected() {
1252 use select::Policy;
1253 let home = std::env::temp_dir().join(format!("borg-top-select-{}", std::process::id()));
1254 let _ = std::fs::remove_dir_all(&home);
1255 std::fs::create_dir_all(&home).unwrap();
1256 let lessons = home.join("lessons.jsonl");
1257 std::fs::write(&lessons, "").unwrap();
1258 let line = |lessons: &Path, policy| selection_line(&selection(lessons, policy));
1259 assert!(line(&lessons, Policy::Learned).ends_with("cold start: gate 0, every decision the fixed rule's · no graded check folded yet"));
1260 assert!(line(&lessons, Policy::Fixed).starts_with("the fixed rule ("), "{}", line(&lessons, Policy::Fixed));
1261 assert!(line(&lessons, Policy::Cap(2)).contains("(BORG_STOP_ASK_CAP, evaluation only)"));
1262 let absent = home.join("absent").join("lessons.jsonl");
1263 assert!(line(&absent, Policy::Attention).contains("(BORG_STOP_ATTENTION=on, evaluation only)"), "no store: the prior");
1264 let _ = std::fs::remove_dir_all(&home);
1265 }
1266
1267 #[test]
1269 fn the_selection_line_follows_the_screen_when_shown() {
1270 let (ds, inbox) = (fixture(), inbox_fixture());
1271 let plain = render("corp", Share::Team, &ds, &inbox, NOW);
1272 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, None, NOW), plain);
1273 let cold = Selection::of(&select::summary::Summary::default(), select::precedence(None, None, false), None);
1274 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, Some(&cold), NOW), plain, "the default at cold start: no line");
1275 let s = learned();
1276 assert_eq!(render_with("corp", Share::Team, &ds, &inbox, Some(&s), NOW), format!("{plain} selection: {}\n", selection_line(&s)));
1277 }
1278
1279 #[test]
1281 fn the_selection_serialises_every_section_for_the_page() {
1282 let v = serde_json::to_value(learned()).unwrap();
1283 let mut keys: Vec<&str> = v.as_object().unwrap().keys().map(String::as_str).collect();
1284 keys.sort_unstable();
1285 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";
1286 assert_eq!(keys, want.split(' ').collect::<Vec<_>>());
1287 assert_eq!((&v["budget"], &v["spend"][0]["source"], &v["deviations"][0]["whole"]), (&Value::Null, &json!("stop"), &json!(400)));
1288 let mut keys: Vec<&str> = v["evaluation"].as_object().unwrap().keys().map(String::as_str).collect();
1289 keys.sort_unstable();
1290 assert_eq!(keys, ["ess", "max_weight", "mean", "sessions", "truncated", "ucb", "units"]);
1291 assert_eq!((&v["default"], &v["measurement"]), (&json!("fixed"), &json!({"placebos": 40, "judged": 31, "tail_shown": 12, "tail_applied": 3})));
1292 let mut keys: Vec<&str> = v["test"].as_object().unwrap().keys().map(String::as_str).collect();
1293 keys.sort_unstable();
1294 let want = "ask_free by_k control dilution due due_checks due_sessions kept ran refused risk sessions set";
1295 assert_eq!(keys, want.split(' ').collect::<Vec<_>>());
1296 }
1297
1298 #[test]
1303 fn the_judged_share_leaves_out_what_its_session_already_had() {
1304 use crate::outcome::{self, Outcome};
1305 use select::summary::{Summary, fold};
1306 let rec = |kind: &str, check: &str, channel: &str, ms: i64| Outcome {
1307 kind: kind.into(),
1308 session: "s".into(),
1309 lessons: vec!["A".into()],
1310 check: check.into(),
1311 channel: channel.into(),
1312 ms,
1313 ..Default::default()
1314 };
1315 let graded = |check: &str, ms: i64| Outcome {
1316 grades: vec![2],
1317 ranks: vec![0],
1318 applied: 1,
1319 free: vec!["A".into()],
1320 pf: vec![1.0],
1321 policy: "fixed".into(),
1322 shown: 20,
1323 tokens: 6_000,
1324 ..rec(outcome::GRADED, check, "live", ms)
1325 };
1326 let log = [
1327 graded("c1", 10),
1328 Outcome { route: outcome::ROUTE_FREE.into(), ..rec(outcome::DELIVERY, "c1", "", 11) },
1329 Outcome { judge: 2, ..rec(outcome::CHANGED, "c1", "queued", 12) },
1330 graded("c2", 20),
1331 Outcome { route: outcome::ROUTE_DROPPED_DONE.into(), ..rec(outcome::DELIVERY, "c2", "", 21) },
1332 Outcome { lessons: vec!["B".into()], ..rec(outcome::ASKED, "", "early", 25) },
1334 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) },
1335 Outcome { lessons: vec!["B".into()], route: outcome::ROUTE_ASKED.into(), ..rec(outcome::DELIVERY, "c3", "", 31) },
1336 Outcome { lessons: vec!["B".into()], judge: 2, ..rec(outcome::CHANGED, "c3", "live", 32) },
1337 ];
1338 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1339 let s = fold(&text, Summary::default());
1340 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");
1341 let sel = Selection::of(&s, select::Policy::Fixed, None);
1342 let row = |name: &str| sel.labels.iter().find(|c| c.name == name).map(|c| (c.part, c.whole));
1343 assert_eq!(row("judged, free"), Some((1, 1)), "c1's delivered and judged; c2's had, out: {:?}", sel.labels);
1344 assert_eq!(row("judged, asked"), Some((1, 1)), "c3's question of what B's session had, judged: {:?}", sel.labels);
1345 }
1346
1347 #[test]
1352 fn the_measurement_counts_the_placebos_the_fixed_rule_drew_and_judged() {
1353 use crate::outcome::{self, Outcome};
1354 use select::summary::{Summary, fold};
1355 let rec = |kind: &str, lessons: &[&str], check: &str, channel: &str, ms: i64| Outcome {
1356 kind: kind.into(),
1357 session: "s".into(),
1358 lessons: lessons.iter().map(|l| l.to_string()).collect(),
1359 check: check.into(),
1360 channel: channel.into(),
1361 ms,
1362 ..Default::default()
1363 };
1364 let graded = Outcome {
1365 grades: vec![1, 1, 2],
1366 ranks: vec![0, 1, 2],
1367 applied: 3,
1368 free: vec!["C".into()],
1369 placebo: vec!["A".into(), "B".into()],
1370 psi: 0.5,
1371 policy: "fixed".into(),
1372 shown: 20,
1373 tokens: 6_000,
1374 ..rec(outcome::GRADED, &["A", "B", "C"], "c1", "stop", 10)
1375 };
1376 let tailed = Outcome {
1378 grades: vec![1],
1379 ranks: vec![0],
1380 applied: 1,
1381 policy: "fixed".into(),
1382 shown: 20,
1383 tokens: 6_000,
1384 tail: vec!["T1".into(), "T2".into()],
1385 pt: 0.5,
1386 fill: vec![21, 25, 21],
1387 ..rec(outcome::GRADED, &["T2"], "c2", "stop", 14)
1388 };
1389 let log = [
1390 graded,
1391 tailed,
1392 Outcome { route: outcome::ROUTE_PLACEBO.into(), ..rec(outcome::DELIVERY, &["A"], "c1", "", 11) },
1393 Outcome { route: outcome::ROUTE_PLACEBO.into(), ..rec(outcome::DELIVERY, &["B"], "c1", "", 11) },
1394 Outcome { judge: 2, ..rec(outcome::IGNORED, &["A"], "c1", outcome::PLACEBO, 12) },
1395 Outcome { judge: 2, ..rec(outcome::CHANGED, &["C"], "c1", "queued", 13) },
1396 ];
1397 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1398 let sel = Selection::of(&fold(&text, Summary::default()), select::Policy::Fixed, None);
1399 assert_eq!(sel.measurement, Some(Measurement { placebos: 2, judged: 1, tail_shown: 2, tail_applied: 1 }), "{sel:?}");
1400 assert!(selection_line(&sel).contains(" · 2 checks · placebos 2 drawn, 1 judged · tail 2 shown past the pool, 1 applied"), "{}", selection_line(&sel));
1401 assert_eq!(Selection::of(&Summary::default(), select::Policy::Fixed, None).measurement, None, "nothing folded: not measured");
1402 }
1403
1404 #[test]
1408 fn the_selection_counts_the_learned_decisions_the_fixed_rule_took_over_by_reason() {
1409 use crate::outcome::{self, Outcome};
1410 use select::summary::{Summary, fold};
1411 let graded = |check: &str, policy: &str, reason: &str, ms: i64| Outcome {
1412 kind: outcome::GRADED.into(),
1413 session: "s".into(),
1414 lessons: vec!["A".into()],
1415 grades: vec![1],
1416 ranks: vec![0],
1417 applied: 1,
1418 free: vec!["A".into()],
1419 pf: vec![1.0],
1420 channel: "live".into(),
1421 check: check.into(),
1422 policy: policy.into(),
1423 reason: reason.into(),
1424 ms,
1425 ..Default::default()
1426 };
1427 let log = [
1428 graded("c1", "learned", "", 1),
1429 graded("c2", "fixed", outcome::DECIDED_BUSY, 2),
1430 graded("c3", "fixed", outcome::DECIDED_NO_RECORD, 3),
1431 graded("c4", "fixed", outcome::DECIDED_BUSY, 4),
1432 ];
1433 let lines = |log: &[Outcome]| log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect::<String>();
1434 let sel = Selection::of(&fold(&lines(&log), Summary::default()), select::Policy::Learned, None);
1435 let rows: Vec<(&str, u64, u64)> = sel.fallbacks.iter().map(|c| (c.name.as_str(), c.part, c.whole)).collect();
1436 assert_eq!(rows, [("lock_busy", 2, 4), ("lock_failed", 0, 4), ("unrecorded", 1, 4)]);
1437 let line = selection_line(&sel);
1438 assert!(line.contains(" · 4 checks · learned decisions the fixed rule took over: lock_busy 2, unrecorded 1 · "), "{line}");
1439 let none = Selection::of(&fold(&lines(&log[..1]), Summary::default()), select::Policy::Learned, None);
1440 assert!(none.fallbacks.iter().all(|c| c.part == 0) && none.fallbacks.len() == 3, "{none:?}");
1441 assert!(!selection_line(&none).contains("took over"), "{}", selection_line(&none));
1442 assert!(Selection::of(&Summary::default(), select::Policy::Learned, None).fallbacks.is_empty(), "nothing folded: not measured");
1443 }
1444
1445 #[test]
1450 fn the_selection_fills_latency_evaluation_and_delivery_from_the_summary() {
1451 use select::summary::{Summary, fold};
1452 let call = |harness: &str, channel: &str, dur_ms: u64, failed: bool| {
1453 let o = crate::outcome::Outcome {
1454 kind: "graded".into(),
1455 session: "s".into(),
1456 channel: channel.into(),
1457 harness: harness.into(),
1458 dur_ms,
1459 failed,
1460 tokens: 6_000,
1461 ..Default::default()
1462 };
1463 format!("{}\n", serde_json::to_string(&o).unwrap())
1464 };
1465 let mut log: String = (1..=20).map(|k| call("codex", "stop", k * 1_000, k == 20)).collect();
1466 log += &call("claude", "stop", 4_000, false);
1467 log += &call("codex", "live", 99_000, false);
1468 log += &call("nonesuch", "stop", 5_000, false);
1469 let mut s = fold(&log, Summary::default());
1470 s.monitor = select::policy::Monitor {
1471 sessions: 12,
1472 units: 400,
1473 mean: 0.004,
1474 ucb: Some(0.031),
1475 labelled: 300,
1476 ess: 180.0,
1477 max_weight: 6.5,
1478 truncated: 3,
1479 ..Default::default()
1480 };
1481 s.fit.key = "fit".into();
1482 s.fit.rates = select::policy::Rates { free_stop: 0.6, free_live: 0.8, later: 0.4, wait: 0.25, ask_live: 1.0 };
1483 let sel = Selection::of(&s, select::Policy::Learned, None);
1484 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);
1485 assert!(budget("codex").is_some() && budget("claude").is_none(), "codex names a stop budget, claude none");
1486 let want = vec![
1487 Latency { harness: "claude".into(), checks: 1, p50_ms: 4_000, p95_ms: 4_000, budget_ms: None, over: 0 },
1488 Latency { harness: "codex".into(), checks: 20, p50_ms: 10_000, p95_ms: 19_000, budget_ms: budget("codex"), over: 1 },
1489 ];
1490 assert_eq!(sel.latency, want);
1491 let e = sel.evaluation.as_ref().expect("the monitor weighed labelled applies");
1492 assert_eq!(*e, Evaluation { sessions: 12, mean: 0.004, ucb: Some(0.031), units: 300, ess: 180.0, max_weight: 6.5, truncated: 3 });
1493 assert_eq!(sel.delivery.as_ref().map(|r| r.wait), Some(0.25));
1494 let keep = select::summary::LATENCY_KEEP as u64;
1496 let many: String = (1..=keep + 1).map(|k| call("claude", "stop", k, false)).collect();
1497 let kept = Selection::of(&fold(&many, Summary::default()), select::Policy::Learned, None);
1498 assert_eq!((kept.latency[0].checks, kept.latency[0].p50_ms), (keep, 1 + keep / 2), "the oldest call dropped");
1499 let cold = Selection::of(&Summary::default(), select::Policy::Learned, None);
1500 assert!(cold.latency.is_empty() && cold.evaluation.is_none() && cold.delivery.is_none(), "{cold:?}");
1501 assert_eq!((quantile(&[], 0.95), quantile(&[7], 0.5)), (0, 7));
1502 }
1503
1504 fn tested(check: &str, session: &str, ms: i64, source: &str, xk: Option<u32>, xs: Option<u32>) -> crate::outcome::Outcome {
1507 crate::outcome::Outcome {
1508 kind: crate::outcome::GRADED.into(),
1509 session: session.into(),
1510 lessons: vec!["A".into(), "B".into()],
1511 grades: vec![2, 1],
1512 ranks: vec![0, 1],
1513 applied: 2,
1514 check: check.into(),
1515 channel: "live".into(),
1516 policy: "learned".into(),
1517 source: source.into(),
1518 xk,
1519 xs,
1520 ms,
1521 ..Default::default()
1522 }
1523 }
1524
1525 #[test]
1531 fn the_test_section_is_the_summarys_fold_of_its_checks() {
1532 use select::policy::{ALPHA, EVIDENCE, EXPERIMENT, EXPERIMENT_CONTROL, GATE};
1533 use select::summary::{Summary, fold};
1534 let log = [
1535 tested("c1", "s1", 1, EXPERIMENT, Some(2), Some(1)),
1536 tested("c2", "s1", 2, EXPERIMENT_CONTROL, Some(2), Some(0)),
1537 tested("c3", "s2", 3, EXPERIMENT, Some(2), Some(1)),
1538 tested("c4", "s2", 4, GATE, Some(2), Some(1)),
1539 tested("c5", "s2", 5, GATE, Some(1), Some(0)),
1540 tested("c6", "s3", 6, EVIDENCE, None, None),
1541 tested("c7", "s3", 7, GATE, None, None),
1542 crate::outcome::Outcome { policy: "fixed".into(), ..tested("c8", "s3", 8, EXPERIMENT, Some(2), Some(2)) },
1543 ];
1544 let text: String = log.iter().map(|o| format!("{}\n", serde_json::to_string(o).unwrap())).collect();
1545 let s = fold(&text, Summary::default());
1546 let t = Selection::of(&s, select::Policy::Learned, Some(0.2)).test.expect("the learned policy shows its test");
1547 let draws = |k: u32, draws: Vec<u64>, refused: u64| Draws { k, draws, refused };
1548 assert_eq!((t.ran, t.control, t.refused, t.sessions), (5, 1, 2, 2), "{t:?}");
1549 assert_eq!(t.by_k, vec![draws(1, vec![1, 0], 1), draws(2, vec![1, 3, 0], 1)]);
1550 assert_eq!((t.risk, t.set), (0.2, true), "the collective's risk");
1551 assert!(!t.due && t.dilution.is_empty() && t.ask_free.is_none(), "5 checks, 2 sessions: not due, nothing estimated");
1552 assert_eq!((t.due_checks, t.due_sessions), (select::replay::TEST_CHECKS, u64::from(select::replay::TEST_SESSIONS)));
1553 let line = selection_line(&Selection::of(&s, select::Policy::Learned, Some(0.2)));
1554 assert!(
1555 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 · "),
1556 "{line}"
1557 );
1558 assert_eq!(t.kept, t.ran, "every drawn check still held: {t:?}");
1559 let mut compacted = s.clone();
1561 compacted.checks.remove(0);
1562 let c = Selection::of(&compacted, select::Policy::Learned, Some(0.2)).test.unwrap();
1563 assert_eq!((c.ran, c.kept), (5, 4), "{c:?}");
1564 assert!(
1565 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"),
1566 "{}",
1567 test_line(&c)
1568 );
1569 let default = Selection::of(&s, select::Policy::Learned, None).test.unwrap();
1570 assert_eq!((default.risk, default.set), (ALPHA, false), "unset: the policy's ALPHA");
1571 let cold = Selection::of(&Summary::default(), select::Policy::Learned, Some(0.0));
1573 assert_eq!(cold.test.as_ref().map(|t| (t.ran, t.risk)), Some((0, 0.0)));
1574 assert_eq!(test_line(cold.test.as_ref().unwrap()), "question-count test off (risk 0)");
1575 let none = Selection::of(&Summary::default(), select::Policy::Learned, None).test.unwrap();
1576 assert_eq!(test_line(&none), "question-count test: no check yet, risk 0.10");
1577 assert!(Selection::of(&Summary::default(), select::Policy::Fixed, None).test.is_none());
1579 assert_eq!(Selection::of(&s, select::Policy::Fixed, None).test.map(|t| t.ran), Some(5), "an earlier opt-in's checks");
1580 }
1581
1582 #[test]
1586 fn the_gate_shown_is_the_one_in_force_at_the_collectives_risk() {
1587 use select::summary::{Gate, Summary};
1588 let s = Summary { gate: Gate { gain: vec![0.0; select::policy::DRAWS], allowance: 0.3, ..Gate::default() }, ..Summary::default() };
1589 let gate = |risk: Option<f64>| Selection::of(&s, select::Policy::Learned, risk).gate;
1590 assert!((gate(None) - 0.3).abs() < 1e-12, "{}", gate(None));
1591 assert!((gate(Some(0.2)) - 0.6).abs() < 1e-12, "twice ALPHA: twice the allowance, {}", gate(Some(0.2)));
1592 assert_eq!(gate(Some(0.0)), 0.0, "risk 0");
1593 }
1594
1595 #[test]
1598 fn the_test_readout_shows_the_replays_estimates_once_due() {
1599 use crate::outcome::{self, Outcome};
1600 use select::policy::EXPERIMENT;
1601 use select::summary::{Summary, fold};
1602 let sessions = u64::from(select::replay::TEST_SESSIONS);
1603 let mut text = String::new();
1604 let mut line = |o: &Outcome| text.push_str(&format!("{}\n", serde_json::to_string(o).unwrap()));
1605 for i in 0..sessions {
1606 let (session, check, ms) = (format!("s{i}"), format!("{i:012x}"), 10 * i as i64);
1607 let asked = i % 2 == 0;
1609 let g = Outcome {
1610 lessons: vec!["A".into()],
1611 grades: vec![2],
1612 ranks: vec![0],
1613 applied: 1,
1614 chosen: if asked { vec!["A".into()] } else { Vec::new() },
1615 free: if asked { Vec::new() } else { vec!["A".into()] },
1616 p: vec![0.5],
1617 pf: vec![0.5],
1618 ..tested(&check, &session, ms, EXPERIMENT, Some(1), Some(u32::from(asked)))
1619 };
1620 line(&g);
1621 let rec = |kind: &str, ms: i64| Outcome {
1622 kind: kind.into(),
1623 session: session.clone(),
1624 lessons: vec!["A".into()],
1625 check: check.clone(),
1626 ms,
1627 ..Default::default()
1628 };
1629 let route = if asked { outcome::ROUTE_ASKED } else { outcome::ROUTE_FREE };
1630 line(&Outcome { route: route.into(), ..rec(outcome::DELIVERY, ms + 1) });
1631 let verdict = if asked { outcome::CHANGED } else { outcome::IGNORED };
1633 line(&Outcome { judge: 2, channel: if asked { "live" } else { "queued" }.into(), ..rec(verdict, ms + 2) });
1634 }
1635 let s = fold(&text, Summary::default());
1636 let t = Selection::of(&s, select::Policy::Learned, None).test.unwrap();
1637 assert!(t.due && t.sessions == sessions, "{t:?}");
1638 let r = select::replay::test_readout(&s);
1639 let want = r.ask_free.as_ref().and_then(|x| x.value).expect("judged 2s on both routes");
1640 let e = t.ask_free.as_ref().expect("estimated once due");
1641 assert_eq!((e.mean, e.unit.as_str()), (100.0 * want, "points changed"));
1642 assert!(e.mean > 0.0, "asking changed the work, handing over did not: {e:?}");
1643 assert!(t.dilution.is_empty() && r.by_asked.is_empty(), "{t:?}");
1645 let line = selection_line(&Selection::of(&s, select::Policy::Learned, None));
1646 assert!(line.contains(&format!("; readout: grade 2 asked − handed over {:+.1} [", e.mean)), "{line}");
1647 }
1648
1649 #[test]
1652 fn unread_counts_follow_the_shown_once_claims() {
1653 let home = std::env::temp_dir().join(format!("borg-top-inbox-{}", std::process::id()));
1654 let _ = std::fs::remove_dir_all(&home);
1655 std::fs::create_dir_all(&home).unwrap();
1656 let ds = fixture();
1657 assert!(inbox(&home, &ds, NOW).recent.is_empty());
1658 let here = msg::node(&home);
1659 msg::send(&home, message_on(&here, "cccc3333-x", "Three of Three", "aaaa1111-x", msg::TELL, NOW - 3_000, "look at recon"), NOW).unwrap();
1660 msg::send(&home, message_on(&here, "cccc3333-x", "Three of Three", msg::ALL, msg::TELL, NOW - 2_000, "lunch"), NOW).unwrap();
1661 let i = inbox(&home, &ds, NOW);
1662 assert_eq!(i.unread.get("aaaa1111-x"), Some(&2), "{i:?}");
1663 assert_eq!(i.unread.get("cccc3333-x"), Some(&0), "its own broadcast");
1664 assert!(!i.unread.contains_key(&ds[1].p.session), "remote: claimed on its node");
1665 assert_eq!((i.day, i.recent[0].text.as_str()), (2, "lunch"));
1666 let caps = crate::harness::by_name("claude").expect("a harness").caps();
1667 let ev = json!({"session_id": "aaaa1111-x"});
1668 assert!(msg::deliver(&home, caps, "prompt", &ev, Some(100), NOW, None).is_some());
1669 assert_eq!(inbox(&home, &ds, NOW).unread.get("aaaa1111-x"), Some(&0));
1670 let _ = std::fs::remove_dir_all(&home);
1671 }
1672
1673 fn seen_lines(home: &Path) -> Vec<(String, Vec<String>)> {
1675 msg::lines(home)
1676 .into_iter()
1677 .filter_map(|l| match l {
1678 msg::Line::Seen(s) => Some((s.by, s.ids)),
1679 msg::Line::Message(_) => None,
1680 })
1681 .collect()
1682 }
1683
1684 fn shell(sessions: Vec<(String, &'static str)>, pids: Vec<(u32, &'static str)>) -> impl Fn(&[Drone]) -> Result<Option<(String, &'static str)>, String> {
1687 move |ds| msg::pick(sessions.clone(), &pids, ds)
1688 }
1689
1690 fn acked_fixture(base: &Path) -> (std::path::PathBuf, String, Message) {
1693 let _ = std::fs::remove_dir_all(base);
1694 let (home, work) = (base.join("store"), base.join("work"));
1695 std::fs::create_dir_all(&home).unwrap();
1696 std::fs::create_dir_all(&work).unwrap();
1697 let git = |a: &[&str]| std::process::Command::new("git").arg("-C").arg(&work).args(a).output().is_ok_and(|o| o.status.success());
1698 assert!(git(&["init", "-q"]) && git(&["config", "user.email", "ann@x"]));
1699 let here = msg::node(&home);
1700 let to_me = message_on(&here, "aaaa1111-x", "One of Three", msg::ENGINEER, msg::TELL, NOW - 3_000, "the build is green");
1701 let to_drone = message_on(&here, "cccc3333-x", "Three of Three", "aaaa1111-x", msg::TELL, NOW - 2_000, "look at recon");
1702 msg::send(&home, to_me.clone(), NOW).unwrap();
1703 msg::send(&home, to_drone, NOW).unwrap();
1704 (home, work.display().to_string(), to_me)
1705 }
1706
1707 #[test]
1711 fn the_screen_acknowledges_the_messages_to_the_engineer_it_prints() {
1712 let base = std::env::temp_dir().join(format!("borg-top-ack-{}", std::process::id()));
1713 let (home, cwd, to_me) = acked_fixture(&base);
1714 let routes = [Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local }];
1715 let terminal = || shell(Vec::new(), Vec::new());
1716 let json = screen_of(&routes, true, &cwd, terminal(), true, NOW);
1717 assert!(json.contains("the build is green") && seen_lines(&home).is_empty(), "--json acknowledges nothing: {json}");
1718 let text = screen_of(&routes, true, &cwd, terminal(), false, NOW);
1719 assert!(text.contains("the build is green") && text.contains("look at recon"), "{text}");
1720 assert_eq!(seen_lines(&home), [("engineer:ann@x".to_string(), vec![to_me.id.clone()])], "the engineer's message only, as theirs");
1721 screen_of(&routes, true, &cwd, terminal(), false, NOW + 1_000);
1722 assert_eq!(seen_lines(&home).len(), 1, "acknowledged once: a later look writes no second line");
1723 let _ = std::fs::remove_dir_all(&base);
1724 }
1725
1726 #[test]
1731 fn a_session_running_the_screen_acknowledges_nothing_as_the_engineer() {
1732 let base = std::env::temp_dir().join(format!("borg-top-ack-session-{}", std::process::id()));
1733 let (home, cwd, to_me) = acked_fixture(&base);
1734 let other = base.join("other");
1735 std::fs::create_dir_all(&other).unwrap();
1736 let pid = std::process::id();
1737 let oc = Presence { harness: "opencode".into(), pid: Some(pid), ..p("dddd4444-x", "ann@x", NOW - 9_000, NOW - 1_000, "prompt", "docs", &[]) };
1738 presence::write(&other, &oc).unwrap();
1739 let routes = [
1740 Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local },
1741 Route::Collective { name: "side".into(), home: other.clone(), share: Share::Local },
1742 ];
1743 let shells = [
1744 ("Claude Code", shell(vec![("eeee5555-x".into(), "claude")], Vec::new())),
1745 ("Codex in Claude Code, neither with presence", shell(vec![("eeee5555-x".into(), "claude"), ("ffff6666-x".into(), "codex")], Vec::new())),
1746 ("opencode, its drone in the other collective", shell(Vec::new(), vec![(pid, "opencode")])),
1747 ];
1748 for (what, caller) in shells {
1749 let text = screen_of(&routes, false, &cwd, caller, false, NOW);
1750 assert!(text.contains("the build is green"), "{what}: shown {text}");
1751 assert!(seen_lines(&home).is_empty() && seen_lines(&other).is_empty(), "{what}: acknowledged nothing as the engineer");
1752 }
1753 screen_of(&routes, false, &cwd, shell(Vec::new(), vec![(pid + 1, "opencode")]), false, NOW);
1754 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");
1755 let _ = std::fs::remove_dir_all(&base);
1756 }
1757
1758 #[test]
1763 fn the_printed_screen_acknowledges_only_the_messages_whose_text_it_prints() {
1764 let base = std::env::temp_dir().join(format!("borg-top-ack-cut-{}", std::process::id()));
1765 let (home, cwd, to_me) = acked_fixture(&base);
1766 let email = format!("{}@x", "a".repeat(58));
1768 assert!(std::process::Command::new("git").args(["-C", &cwd, "config", "user.email", &email]).output().is_ok_and(|o| o.status.success()));
1769 let (here, to) = (msg::node(&home), format!("{}:{email}", msg::ENGINEER));
1770 let one = message_on(&here, "bbbb2222-x", &"n".repeat(52), &to, msg::TELL, NOW - 1_500, "one character shows");
1771 let none = message_on(&here, "dddd4444-x", &"n".repeat(53), &to, msg::TELL, NOW - 1_000, "none shows");
1772 msg::send(&home, one.clone(), NOW).unwrap();
1773 msg::send(&home, none.clone(), NOW).unwrap();
1774 let routes = [Route::Collective { name: "corp".into(), home: home.clone(), share: Share::Local }];
1775 let text = screen_of(&routes, true, &cwd, shell(Vec::new(), Vec::new()), false, NOW);
1776 assert!(text.contains(&format!("{} -> {email}: o…", one.name)) && text.contains(&format!("{} -> {email}: …", none.name)), "{text}");
1777 let mut acked: Vec<String> = seen_lines(&home).into_iter().flat_map(|(by, ids)| ids.into_iter().map(move |id| format!("{by} {id}"))).collect();
1778 acked.sort();
1779 let mut want = [format!("{to} {}", to_me.id), format!("{to} {}", one.id)];
1780 want.sort();
1781 assert_eq!(acked, want, "the line that shows one character of its text is read; the one cut before it is not");
1782 let _ = std::fs::remove_dir_all(&base);
1783 }
1784
1785 #[test]
1789 fn a_live_screen_acknowledges_each_message_once() {
1790 let base = std::env::temp_dir().join(format!("borg-top-ack-once-{}", std::process::id()));
1791 let _ = std::fs::remove_dir_all(&base);
1792 std::fs::create_dir_all(&base).unwrap();
1793 let home = base.clone();
1794 let here = msg::node(&home);
1795 let to_me = |by: &str, ms: i64, text: &str| message_on(&here, by, "a drone", msg::ENGINEER, msg::TELL, ms, text);
1796 let first = to_me("aaaa1111-x", NOW - 3_000, "the build is green");
1797 msg::send(&home, first.clone(), NOW).unwrap();
1798 let cwd = base.display().to_string();
1799 let mut done = HashSet::new();
1800 let recent = |now: i64| inbox(&home, &[], now).recent;
1801 acknowledge(&home, &cwd, &recent(NOW).iter().collect::<Vec<_>>(), &mut done, NOW);
1802 assert_eq!(seen_lines(&home).len(), 1, "{:?}", seen_lines(&home));
1803 assert!(done.contains(&first.id));
1804 let kept: String = std::fs::read_to_string(msg::path(&home)).unwrap().lines().filter(|l| !l.contains("\"seen\"")).map(|l| format!("{l}\n")).collect();
1807 std::fs::write(msg::path(&home), kept).unwrap();
1808 acknowledge(&home, &cwd, &recent(NOW + 1_000).iter().collect::<Vec<_>>(), &mut done, NOW + 1_000);
1809 assert!(seen_lines(&home).is_empty(), "skipped: no second ack of a done id");
1810 let second = to_me("cccc3333-x", NOW + 2_000, "and the docs");
1812 msg::send(&home, second.clone(), NOW + 2_000).unwrap();
1813 acknowledge(&home, &cwd, &recent(NOW + 3_000).iter().collect::<Vec<_>>(), &mut done, NOW + 3_000);
1814 assert_eq!(seen_lines(&home).into_iter().map(|(_, ids)| ids).collect::<Vec<_>>(), [vec![second.id.clone()]], "only the new one");
1815 let _ = std::fs::remove_dir_all(&base);
1816 }
1817
1818 #[test]
1821 fn a_sender_is_tied_to_the_node_its_line_names() {
1822 let ds = fixture();
1823 let own = message("aaaa1111-x", "anything", msg::ALL, msg::TELL, NOW, "hi");
1824 assert_eq!(sender(&ds, "n1", &own), "One of Three", "this node's drone");
1825 let forged = message_on("evil0123456789", "aaaa1111-x", "One of Three", msg::ALL, msg::TELL, NOW, "hi");
1826 assert_eq!(sender(&ds, "n1", &forged), "One of Three @evil01234567", "the claim, with the node it came from");
1827 let remote = message_on(&"beefcafe".repeat(8), &ds[1].p.session, "x", msg::ALL, msg::TELL, NOW, "hi");
1828 assert_eq!(sender(&ds, "n1", &remote), "Two of Three @beefcafebeef", "the remote drone on its own node");
1829 assert_eq!(recipient(&ds, "n1", "aaaa1111-x@n1"), "One of Three");
1830 assert_eq!(recipient(&ds, "n1", "aaaa1111-x@evil"), "aaaa1111", "not the local drone of that id");
1831 assert_eq!((recipient(&ds, "n1", "engineer"), recipient(&ds, "n1", msg::ALL)), ("engineer".into(), "collective".into()), "an `all` line");
1832 assert_eq!(recipient(&ds, "n1", "local@n1"), "local");
1833 assert_eq!(recipient(&ds, "n1", "local@beefcafebeefcafe00"), "local @beefcafebeef");
1834 }
1835
1836 #[test]
1837 fn json_for_scripts() {
1838 let v = to_json_with("corp", Share::Team, &fixture(), &inbox_fixture());
1839 assert_eq!((v["drones"][0]["unread"].as_u64(), v["drones"][1].get("unread")), (Some(2), None));
1840 assert_eq!((v["messages"]["day"].as_u64(), v["messages"]["recent"][0]["kind"].as_str()), (Some(3), Some("ask")));
1841 assert_eq!(v["collective"], "corp");
1842 let d = &v["drones"];
1843 assert_eq!(d.as_array().unwrap().len(), 3);
1844 assert_eq!(d[0]["designation"], "One of Three");
1845 assert_eq!(d[0]["session"], "aaaa1111-x");
1846 assert_eq!(d[0]["state"], "working");
1847 assert_eq!(d[0]["paused"], true);
1848 assert_eq!(d[1]["node"], "beefcafe".repeat(8));
1849 assert_eq!(d[1]["state"], "waiting");
1850 assert_eq!(d[2]["state"], "idle");
1851 assert_eq!(d[2]["files"], json!([]));
1852 }
1853
1854 #[test]
1856 fn a_name_matching_here_and_on_a_member_names_the_drone_here() {
1857 let local = vec![p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "post_tool", "a", &[])];
1858 let remote = ["aaaa1111-x", "aaaa2222-z"].map(|s| presence::Remote {
1859 p: p(s, "bob@y", 50, NOW - 5_000, "post_tool", "b", &[]),
1860 key: "beefcafe".repeat(8),
1861 sig: String::new(),
1862 });
1863 let ds = assemble(local, remote.into(), NOW, |_| true, |_| false);
1864 for t in ["aaaa", "aaaa1111", "aaaa1111-x"] {
1865 let d = resolve(&ds, t).unwrap().unwrap();
1866 assert!(d.node.is_empty() && d.p.session == "aaaa1111-x", "{t}");
1867 }
1868 assert_eq!(resolve(&ds, "aaaa2").unwrap().unwrap().node, "beefcafe".repeat(8), "only a member's matches");
1869 assert!(!resolve(&ds, "One").unwrap().unwrap().node.is_empty(), "a number names one drone of all, wherever it runs");
1870 }
1871
1872 #[test]
1874 fn a_name_matching_an_ended_drone_here_and_a_live_member_names_the_live_one() {
1875 let local = vec![p("aaaa1111-x", "alice@x", 100, NOW - 5_000, "end", "a", &[])];
1876 let remote =
1877 vec![presence::Remote { p: p("aaaa2222-z", "bob@y", 50, NOW - 5_000, "post_tool", "b", &[]), key: "beefcafe".repeat(8), sig: String::new() }];
1878 let ds = assemble(local, remote, NOW, |_| true, |_| false);
1879 let d = resolve(&ds, "aaaa").unwrap().unwrap();
1880 assert_eq!((d.p.session.as_str(), d.node.as_str()), ("aaaa2222-z", "beefcafe".repeat(8).as_str()), "the live one, though remote");
1881 assert!(resolve(&ds, "aaaa1111").unwrap().unwrap().node.is_empty(), "an ended drone is still named by its own id");
1882 }
1883
1884 #[test]
1885 fn resolves_drones_by_designation_or_session() {
1886 let ds = fixture();
1887 assert_eq!(resolve(&ds, "Two").unwrap().unwrap().p.session, "bbbb2222-y");
1888 assert_eq!(resolve(&ds, "two of three").unwrap().unwrap().p.session, "bbbb2222-y");
1889 assert_eq!(resolve(&ds, "3").unwrap().unwrap().p.session, "cccc3333-x");
1890 assert_eq!(resolve(&ds, "aaaa").unwrap().unwrap().p.session, "aaaa1111-x");
1891 assert!(resolve(&ds, "all").unwrap().is_none());
1892 assert!(resolve(&ds, "Four").is_err());
1893 assert!(resolve(&ds, "zzzz").is_err());
1894 assert!(resolve(&ds, "aa").is_err(), "too short a prefix");
1895 assert!(["collective", "local"].iter().all(|w| resolve(&ds, w).unwrap().is_none()));
1896 assert_eq!(resolve(&ds, "bbbb").unwrap().unwrap().p.session, "bbbb2222-y", "nothing here matches: the member's");
1897 assert_eq!(word(21), "21");
1898 }
1899}