1use std::collections::BTreeMap;
12use std::convert::Infallible;
13use std::path::{Path, PathBuf};
14use std::sync::Arc;
15use std::time::Duration;
16
17use axum::Router;
18use axum::extract::{Query, State};
19use axum::http::{StatusCode, header};
20use axum::response::sse::{Event, KeepAlive, Sse};
21use axum::response::{IntoResponse, Json, Response};
22use axum::routing::{get, post};
23use borg::collective::{self, Route, Share};
24use borg::{conscience, install, learn, memory, msg, outcome, presence, stats, top};
25use futures_util::{Stream, StreamExt};
26
27pub mod learning;
28use serde::Deserialize;
29use serde_json::{Value, json};
30
31pub const USAGE: &str = "usage: borg web [--port PORT] [--dir DIR] (default port 7979; DIR picks the collective shown first)";
32
33#[derive(Clone)]
35pub struct Mind {
36 pub name: String,
37 pub home: PathBuf,
38 pub share: Share,
39 pub mode: &'static str,
41}
42
43pub struct App {
44 pub minds: Vec<Mind>,
45 pub first: String,
47 pub cwd: String,
50 pub stop: tokio::sync::watch::Receiver<bool>,
52}
53
54impl App {
55 fn mind(&self, name: Option<&str>) -> Option<&Mind> {
56 let name = name.unwrap_or(&self.first);
57 self.minds.iter().find(|m| m.name == name)
58 }
59}
60
61#[derive(Deserialize)]
62struct Pick {
63 c: Option<String>,
64}
65
66pub fn minds(dir: &str) -> Result<(Vec<Mind>, String), String> {
68 let cfg = collective::load()?;
69 let base = learn::home();
70 let minds: Vec<Mind> = install::collectives(cfg.as_ref(), &base)
71 .into_iter()
72 .filter_map(|r| match r {
73 Route::Collective { name, home, share } => {
74 let mode = borg::live::of(cfg.as_ref(), &name).mode.name();
75 Some(Mind { name, home, share, mode })
76 }
77 Route::Off => None,
78 })
79 .collect();
80 let first = match collective::route(cfg.as_ref(), dir, &base) {
81 Route::Collective { name, .. } => name,
82 Route::Off => minds.first().map(|m| m.name.clone()).unwrap_or_default(),
83 };
84 Ok((minds, first))
85}
86
87fn not_found(what: &str) -> Response {
88 (StatusCode::NOT_FOUND, format!("no collective {what}")).into_response()
89}
90
91async fn collectives(State(app): State<Arc<App>>) -> Json<Value> {
92 let list: Vec<Value> =
93 app.minds.iter().map(|m| json!({"name": m.name, "share": if m.share == Share::Team { "team" } else { "local" }, "mode": m.mode})).collect();
94 Json(json!({"first": app.first, "collectives": list}))
95}
96
97fn sum_by(buckets: &[stats::Bucket], key: impl Fn(&stats::Bucket) -> String) -> Vec<(String, stats::Bucket)> {
99 let mut out: BTreeMap<String, stats::Bucket> = BTreeMap::new();
100 for b in buckets {
101 out.entry(key(b)).or_default().add(b);
102 }
103 out.into_iter().collect()
104}
105
106fn totals(b: &stats::Bucket) -> Value {
107 json!({"counts": b.counts, "usd": b.usd, "usd_by": b.usd_by, "usd_mode": b.usd_mode, "channels": b.channels, "dur_ms": b.dur_ms})
108}
109
110fn series(buckets: &[stats::Bucket], first: &str, last: &str) -> Vec<Value> {
112 let by: BTreeMap<String, stats::Bucket> = sum_by(buckets, |b| b.day.clone()).into_iter().collect();
113 stats::days_between(first, last)
114 .into_iter()
115 .map(|d| {
116 let t = by.get(&d).map(totals).unwrap_or_else(|| totals(&stats::Bucket::default()));
117 json!({"day": d, "totals": t})
118 })
119 .collect()
120}
121
122async fn stats_of(State(app): State<Arc<App>>, Query(p): Query<Pick>) -> Response {
123 let Some(m) = app.mind(p.c.as_deref()) else { return not_found(p.c.as_deref().unwrap_or_default()) };
124 let buckets = stats::current(&m.home);
125 let mut all = stats::Bucket::default();
126 for b in &buckets {
127 all.add(b);
128 }
129 let aliases = collective::aliases();
130 let first = buckets.iter().map(|b| b.day.as_str()).min().unwrap_or_default().to_string();
131 let last = buckets.iter().map(|b| b.day.as_str()).max().unwrap_or_default().to_string();
132 let days = series(&buckets, &first, &last);
133 let engineers: Vec<Value> = sum_by(&buckets, |b| aliases.display(&b.engineer))
134 .into_iter()
135 .map(|(e, t)| {
136 let mine: Vec<stats::Bucket> = buckets.iter().filter(|b| aliases.display(&b.engineer) == e).cloned().collect();
137 json!({"engineer": e, "totals": totals(&t), "days": series(&mine, &first, &last)})
138 })
139 .collect();
140 let nodes = buckets.iter().map(|b| b.node.as_str()).collect::<std::collections::BTreeSet<_>>().len();
141 let (ds, skipped) = borg::ope::decisions(&outcome::read(&m.home.join("outcomes.jsonl")));
144 let ope = json!({"decisions": ds.len(), "no_p": skipped.no_p, "no_pool": skipped.no_pool});
145 Json(json!({"collective": m.name, "totals": totals(&all), "days": days, "engineers": engineers, "nodes": nodes, "ope": ope})).into_response()
146}
147
148async fn lessons(State(app): State<Arc<App>>, Query(p): Query<Pick>) -> Response {
149 let Some(m) = app.mind(p.c.as_deref()) else { return not_found(p.c.as_deref().unwrap_or_default()) };
150 let path = m.home.join("lessons.jsonl");
151 let list = conscience::load_unweighted(&path, None);
152 let outcomes = outcome::read_team(&path);
153 let mem = memory::derive(&outcomes, &list, None);
154 let mut usage: BTreeMap<String, BTreeMap<String, u64>> = BTreeMap::new();
156 for b in stats::current(&m.home) {
157 for (id, kinds) in b.lessons {
158 let u = usage.entry(id).or_default();
159 for (k, v) in kinds {
160 *u.entry(k).or_default() += v;
161 }
162 }
163 }
164 let aliases = collective::aliases();
165 let stored: BTreeMap<String, (u32, String)> = learn::lessons(&path).into_iter().map(|l| (l.id, (l.version, l.repo))).collect();
166 let mut out: Vec<Value> = list
167 .iter()
168 .map(|c| {
169 let mm = mem.get(&c.id);
170 json!({
171 "id": c.id, "question": c.question, "situations": c.situations, "consequence": c.consequence,
172 "area": c.area, "scope": c.scope, "source": c.source, "origin_ms": c.origin_ms,
173 "engineers": c.engineers.iter().map(|e| aliases.display(e)).collect::<Vec<_>>(),
174 "activation": mm.map_or(0.0, |x| x.activation), "changed": mm.map_or(0, |x| x.changed),
175 "ignored": mm.map_or(0, |x| x.ignored), "corrected": mm.map_or(0, |x| x.corrected), "vetoed": mm.map_or(0, |x| x.vetoed),
176 "regretted": mm.map_or(0.0, |x| x.regretted),
177 "tested": mm.map_or(0, |x| x.tested),
178 "dormant": mm.is_some_and(|x| x.dormant), "consolidated": mm.is_some_and(|x| x.consolidated),
179 "usage": usage.get(&c.id).cloned().unwrap_or_default(),
180 "version": stored.get(&c.id).map_or(1, |s| s.0), "repo": stored.get(&c.id).map(|s| s.1.clone()).unwrap_or_default(),
181 "reach": c.reach,
182 })
183 })
184 .collect();
185 out.sort_by(|a, b| b["activation"].as_f64().partial_cmp(&a["activation"].as_f64()).unwrap_or(std::cmp::Ordering::Equal));
186 Json(json!({"collective": m.name, "lessons": out})).into_response()
187}
188
189const NOTABLE: [&str; 28] = [
191 stats::CORRECTION,
192 stats::VETO,
193 stats::VETO_REPEAT,
194 stats::REVIEW_FINDING,
195 stats::BLOCK,
196 stats::CHANGED,
197 stats::IGNORED,
198 stats::CORRECTED,
199 stats::REGRETTED,
200 stats::TESTED,
201 stats::LESSON_NEW,
202 stats::LESSON_MERGED,
203 stats::NOT_A_LESSON,
204 stats::STEER_ISSUED,
205 stats::STEER_DELIVERED,
206 stats::STEER_DROPPED,
207 stats::PAUSE_REFUSED,
208 stats::QUEEN_QUEUED,
209 stats::QUEEN_DELIVERED,
210 stats::QUEEN_ANSWERED,
211 stats::QUEEN_SKIPPED,
212 stats::QUEEN_DISMISSED,
213 stats::ASSIMILATED,
214 stats::EARLY,
215 stats::QUEUED_FREE,
216 stats::MODEL_CALL,
217 stats::MSG_SENT,
220 stats::MSG_DROPPED,
221];
222
223#[derive(Deserialize)]
224struct FeedQuery {
225 c: Option<String>,
226 limit: Option<usize>,
227}
228
229const PAGE: usize = 200;
231const PAGE_MAX: usize = 2000;
234
235async fn feed(State(app): State<Arc<App>>, Query(q): Query<FeedQuery>) -> Response {
237 let Some(m) = app.mind(q.c.as_deref()) else { return not_found(q.c.as_deref().unwrap_or_default()) };
238 stats::backfill(&m.home);
239 let aliases = collective::aliases();
240 let mut evs: Vec<stats::Event> = stats::events(&m.home).into_iter().filter(|e| NOTABLE.contains(&e.kind.as_str())).collect();
241 evs.sort_by_key(|e| std::cmp::Reverse(e.ms));
242 evs.truncate(q.limit.unwrap_or(PAGE).min(PAGE_MAX));
243 let out: Vec<Value> = evs
244 .into_iter()
245 .map(|e| {
246 json!({"ms": e.ms, "kind": e.kind, "engineer": aliases.display(&e.engineer), "repo": e.repo, "lessons": e.lessons,
247 "n": e.n, "usd": e.usd, "dur_ms": e.dur_ms, "why": e.why, "text": e.text})
248 })
249 .collect();
250 Json(json!({"collective": m.name, "events": out})).into_response()
251}
252
253async fn queen(State(app): State<Arc<App>>, Query(p): Query<Pick>) -> Response {
254 let Some(m) = app.mind(p.c.as_deref()) else { return not_found(p.c.as_deref().unwrap_or_default()) };
255 let mut entries = borg::queen::entries(&m.home);
256 entries.sort_by_key(|e| std::cmp::Reverse(e.ms));
257 let attention = borg::escalation::team(&m.home);
259 Json(json!({"collective": m.name, "entries": entries, "attention": attention})).into_response()
260}
261
262async fn drones(State(app): State<Arc<App>>, Query(p): Query<Pick>) -> Response {
263 let Some(m) = app.mind(p.c.as_deref()) else { return not_found(p.c.as_deref().unwrap_or_default()) };
264 let now = learn::now_ms();
265 let mut ds = top::drones(&m.home, m.share == Share::Team, now);
266 for d in &mut ds {
267 d.p.engineer = collective::aliases().display(&d.p.engineer);
268 }
269 let inbox = top::inbox(&m.home, &ds, now);
270 Json(top::to_json_with(&m.name, m.share, &ds, &inbox)).into_response()
271}
272
273async fn messages(State(app): State<Arc<App>>, Query(q): Query<FeedQuery>) -> Response {
280 let Some(m) = app.mind(q.c.as_deref()) else { return not_found(q.c.as_deref().unwrap_or_default()) };
281 let now = learn::now_ms();
282 let ds = top::drones(&m.home, m.share == Share::Team, now);
283 let shown = msg::seen_by(&m.home);
284 let here = msg::node(&m.home);
285 let mut all: Vec<msg::Message> = msg::read(&m.home).into_iter().filter(|x| now - x.ms <= msg::TTL_MS).collect();
286 let unread: usize = top::unread_of(&m.home, &ds, &here, &all, now).values().sum();
287 all.sort_by(|a, b| (b.ms, &b.id).cmp(&(a.ms, &a.id)));
288 all.truncate(q.limit.unwrap_or(PAGE).min(PAGE_MAX));
289 let out: Vec<Value> = all
290 .iter()
291 .map(|x| {
292 let to = shown.get(&presence::safe(&x.id)).map_or(0, Vec::len);
293 let to_kind = match msg::Address::parse(&x.to) {
294 Some(a) if a.broadcast() => "all",
295 Some(msg::Address::Engineer { .. }) => "engineer",
296 _ => "session",
297 };
298 json!({"id": x.id, "kind": x.kind, "by": x.by, "from": top::sender(&ds, &here, x), "harness": x.harness, "node": x.node,
299 "to": x.to, "to_name": top::recipient(&ds, &here, &x.to), "to_kind": to_kind, "re": x.re, "ms": x.ms, "text": x.text, "shown_to": to})
300 })
301 .collect();
302 Json(json!({"collective": m.name, "messages": out, "unread": unread})).into_response()
303}
304
305#[derive(Deserialize)]
306struct Shown {
307 ids: Vec<String>,
308}
309
310async fn messages_seen(State(app): State<Arc<App>>, Query(p): Query<Pick>, Json(shown): Json<Shown>) -> Response {
316 let Some(m) = app.mind(p.c.as_deref()) else { return not_found(p.c.as_deref().unwrap_or_default()) };
317 let now = learn::now_ms();
318 let ids: std::collections::HashSet<&str> = shown.ids.iter().take(PAGE_MAX).map(String::as_str).collect();
319 let all: Vec<msg::Message> = msg::read(&m.home).into_iter().filter(|x| now - x.ms <= msg::TTL_MS && ids.contains(x.id.as_str())).collect();
320 let shown: Vec<&msg::Message> = all.iter().collect();
321 match msg::ack_engineer(&m.home, &app.cwd, &shown, now) {
322 Ok(()) => Json(json!({"collective": m.name})).into_response(),
323 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("borg web: acknowledging: {e}")).into_response(),
324 }
325}
326
327async fn select_of(State(app): State<Arc<App>>, Query(p): Query<Pick>) -> Response {
332 let Some(m) = app.mind(p.c.as_deref()) else { return not_found(p.c.as_deref().unwrap_or_default()) };
333 let s = top::selection(&m.home.join("lessons.jsonl"), borg::select::policy());
334 Json(json!({"collective": m.name, "selection": s})).into_response()
335}
336
337async fn learning_of(State(app): State<Arc<App>>, Query(p): Query<Pick>) -> Response {
339 let Some(m) = app.mind(p.c.as_deref()) else { return not_found(p.c.as_deref().unwrap_or_default()) };
340 Json(learning::view(&m.name, &m.home, m.share)).into_response()
341}
342
343fn stamp(path: &Path) -> Vec<(u64, u128)> {
345 let one = |p: &Path| std::fs::metadata(p).ok().map(|m| (m.len(), mtime(&m)));
346 match std::fs::read_dir(path) {
347 Ok(rd) => {
348 let mut v: Vec<(u64, u128)> = rd.filter_map(|e| e.ok()).filter_map(|e| one(&e.path())).collect();
349 v.sort();
350 v
351 }
352 Err(_) => one(path).into_iter().collect(),
353 }
354}
355
356fn mtime(m: &std::fs::Metadata) -> u128 {
357 m.modified().ok().and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()).map_or(0, |d| d.as_nanos())
358}
359
360fn watched(home: &Path) -> Vec<(&'static str, Vec<PathBuf>)> {
362 vec![
363 (
364 "lessons",
365 vec![home.join("lessons.jsonl"), home.join("outcomes.jsonl"), home.join(outcome::TEAM_FILE), home.join("stats.jsonl"), stats::events_path(home)],
366 ),
367 ("stats", vec![home.join("stats.jsonl"), stats::events_path(home)]),
368 ("drones", vec![home.join("presence"), presence::remote_path(home), msg::path(home), msg::seen_path(home)]),
370 ("messages", vec![msg::path(home), msg::seen_path(home)]),
371 ("feed", vec![stats::events_path(home)]),
372 ("queen", vec![home.join("queen"), home.join("stats.jsonl"), stats::events_path(home)]),
373 ("learning", vec![home.join("lessons.jsonl"), home.join("stats.jsonl"), stats::events_path(home)]),
375 ("select", vec![home.join(borg::select::summary::CACHE), home.join("outcomes.jsonl")]),
377 ]
378}
379
380type Fingerprint = Vec<(&'static str, Vec<Vec<(u64, u128)>>)>;
382
383fn fingerprint(home: &Path) -> Fingerprint {
384 watched(home).into_iter().map(|(view, paths)| (view, paths.iter().map(|p| stamp(p)).collect())).collect()
385}
386
387async fn events(State(app): State<Arc<App>>, Query(p): Query<Pick>) -> Response {
388 let Some(m) = app.mind(p.c.as_deref()) else { return not_found(p.c.as_deref().unwrap_or_default()) };
389 let home = m.home.clone();
390 let start = fingerprint(&home);
391 let stream = futures_util::stream::unfold((home, start), |(home, mut last)| async move {
392 loop {
393 tokio::time::sleep(Duration::from_secs(1)).await;
394 let now = fingerprint(&home);
395 let changed: Vec<&str> = now.iter().zip(&last).filter(|(a, b)| a != b).map(|(a, _)| a.0).collect();
396 if !changed.is_empty() {
397 let ev = Event::default().event("change").data(json!({"changed": changed}).to_string());
398 last = now;
399 return Some((Ok::<Event, Infallible>(ev), (home, last)));
400 }
401 last = now;
402 }
403 });
404 let mut stop = app.stop.clone();
405 let stream = stream.take_until(async move {
406 let _ = stop.wait_for(|s| *s).await;
407 });
408 Sse::new(Box::pin(stream) as std::pin::Pin<Box<dyn Stream<Item = Result<Event, Infallible>> + Send>>).keep_alive(KeepAlive::default()).into_response()
409}
410
411const INDEX: &str = include_str!("../dist/index.html");
413const JS: &[u8] = include_bytes!("../dist/borg-web-ui.js");
414const WASM: &[u8] = include_bytes!("../dist/borg-web-ui_bg.wasm");
415
416fn asset(kind: &'static str, body: &'static [u8]) -> Response {
417 ([(header::CONTENT_TYPE, kind), (header::CACHE_CONTROL, "no-cache")], body).into_response()
418}
419
420pub fn router(app: Arc<App>) -> Router {
421 Router::new()
422 .route("/", get(|| async { ([(header::CONTENT_TYPE, "text/html; charset=utf-8")], INDEX).into_response() }))
423 .route("/borg-web-ui.js", get(|| async { asset("text/javascript", JS) }))
424 .route("/borg-web-ui_bg.wasm", get(|| async { asset("application/wasm", WASM) }))
425 .route("/api/collectives", get(collectives))
426 .route("/api/stats", get(stats_of))
427 .route("/api/lessons", get(lessons))
428 .route("/api/drones", get(drones))
429 .route("/api/feed", get(feed))
430 .route("/api/queen", get(queen))
431 .route("/api/messages", get(messages))
432 .route("/api/messages/seen", post(messages_seen))
433 .route("/api/learning", get(learning_of))
434 .route("/api/select", get(select_of))
435 .route("/api/events", get(events))
436 .layer(axum::middleware::from_fn(loopback_only))
437 .with_state(app)
438}
439
440async fn loopback_only(req: axum::extract::Request, next: axum::middleware::Next) -> Response {
445 match req.headers().get(header::HOST).and_then(|h| h.to_str().ok()).is_some_and(loopback) {
446 true => next.run(req).await,
447 false => (StatusCode::FORBIDDEN, "borg web: served to a loopback Host only (127.0.0.1, localhost, [::1])\n").into_response(),
448 }
449}
450
451fn loopback(host: &str) -> bool {
453 let (name, port) = match host.strip_prefix('[') {
454 Some(v6) => match v6.split_once(']') {
455 Some(("::1", port)) => ("::1", port),
456 _ => return false,
457 },
458 None => host.find(':').map_or((host, ""), |i| host.split_at(i)),
459 };
460 let port = port.is_empty() || port.strip_prefix(':').is_some_and(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit()));
461 port && (name == "::1" || name == "127.0.0.1" || name.eq_ignore_ascii_case("localhost"))
462}
463
464pub async fn publish(minds: Vec<Mind>) {
467 loop {
468 for m in &minds {
469 stats::rollup(&m.home);
470 outcome::publish(&m.home);
471 }
472 tokio::time::sleep(Duration::from_secs(60)).await;
473 }
474}
475
476#[cfg(test)]
477mod tests {
478 use super::*;
479
480 #[test]
483 fn the_selection_view_follows_its_summary_and_the_outcomes() {
484 let home = std::env::temp_dir().join(format!("borg-web-watch-select-{}", std::process::id()));
485 let _ = std::fs::remove_dir_all(&home);
486 std::fs::create_dir_all(&home).unwrap();
487 let changed = |a: &Fingerprint, b: &Fingerprint| a.iter().zip(b).filter(|(x, y)| x != y).map(|(x, _)| x.0).collect::<Vec<_>>();
488 let before = fingerprint(&home);
489 std::fs::write(home.join(borg::select::summary::CACHE), "{\"sv\":1,\"at\":0}").unwrap();
490 let cached = fingerprint(&home);
491 assert_eq!(changed(&before, &cached), ["select"]);
492 std::fs::write(home.join("outcomes.jsonl"), "{}\n").unwrap();
493 assert!(changed(&cached, &fingerprint(&home)).contains(&"select"));
494 let _ = std::fs::remove_dir_all(&home);
495 }
496}