Skip to main content

magi/
waiter.rs

1//! The waiter: something that keeps waiting on a question after the process
2//! that asked it is gone.
3//!
4//! `magi ask` blocks in the agent's own shell tool, and that tool kills a wait
5//! that runs long (see `ask::WAIT_SLICE`). An agent can also background the
6//! call and simply finish. Either way the question outlives the only process
7//! that would have read the owner's answer, and the owner's reply lands in a
8//! file nobody is polling - the phone then says "waiting for the agent" while
9//! there is no agent to wait for.
10//!
11//! This module is the guarantee that something is always on the other end. It
12//! runs inside `magi serve`, on its own cadence rather than the queue's, and
13//! for every question a `magi ask` filed it decides one of three things:
14//!
15//! - **the asker is still there** (its [`Lease`] is fresh): do nothing. A
16//!   second agent must never be started while the first is blocked;
17//! - **the asker is gone and the owner has said or answered something the agent
18//!   has not read**: resume the *same seat's* CLI session with the question,
19//!   the thread so far and the owner's words, so the agent that holds the
20//!   context carries on;
21//! - **nobody answered before `answer_timeout`**: abandon it, in the same words
22//!   the asker would have used.
23//!
24//! # Shape
25//!
26//! Same split as [`crate::queue`] and [`crate::ask`]: [`decide`] is pure - a
27//! question, a lease, a clock and a bool in, an [`Action`] out - and every
28//! filesystem or process effect is in [`Waiter::tick`]. State lives on disk (the
29//! question record and its lease), so a restarted daemon picks up exactly where
30//! the last one stopped, including an answer nobody has read yet.
31//!
32//! # What it will not do
33//!
34//! It never starts a fresh consultant. If the seat's session cannot be resumed
35//! (`agent::has_session` says no, the working directory is gone, the run is
36//! over, the agent left the roster) the question stays as it is, the owner
37//! gets a notification saying why, and nothing else runs. A new agent would
38//! have to be caught up on everything the first one knew, and would act on the
39//! owner's words without the context they were written against.
40//!
41//! It never assumes a resume worked either: a turn that fails or times out
42//! leaves the word undelivered, and a later tick tries again.
43
44use std::collections::HashMap;
45use std::path::PathBuf;
46use std::time::{Duration, Instant};
47
48use anyhow::Result;
49use jiff::Timestamp;
50
51use crate::agent::{self, Invocation, SeatState};
52use crate::ask::{Lease, Question, QuestionStatus, Questions, Waiter as Note, WaiterKind, Who};
53use crate::config::{AgentSpec, Config};
54use crate::notices::{self, Notice};
55use crate::prompt;
56use crate::run::RunState;
57
58/// How often the waiter looks at the store.
59pub const TICK: Duration = Duration::from_secs(5);
60
61/// Wall-clock limit for one resumed turn. A resumed agent may go on to do real
62/// work once it has the owner's word, but it is one conversation turn, not a
63/// whole node.
64const DELIVERY_TIMEOUT: Duration = Duration::from_secs(15 * 60);
65
66/// How long a failed or refused delivery is left alone before the same input
67/// is tried again. Long enough not to hammer a seat that is out of quota,
68/// short enough that a fixed problem does not wait a day.
69const RETRY_AFTER: Duration = Duration::from_secs(10 * 60);
70
71/// What the agent has not read yet.
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub enum Word {
74    /// The owner spoke back without deciding.
75    Said(String),
76    /// The owner decided.
77    Answered(String),
78}
79
80/// What to do about one question, this tick.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub enum Action {
83    /// Nothing: not ours, someone is waiting, or nothing new to hand over.
84    Idle,
85    /// The deadline passed with no answer and nobody waiting.
86    Expire,
87    /// Resume the asking seat with this.
88    Deliver(Word),
89}
90
91/// The whole decision, with no I/O.
92///
93/// `seat_busy` is whether the asking seat's CLI is still mid-turn - the agent
94/// that exited its `magi ask` may not have exited itself, and resuming a
95/// conversation that is running would fork it.
96///
97/// Order matters: a word the owner gave before the deadline is delivered even
98/// if the deadline has passed by the time the waiter looks, and only then does
99/// an unanswered question expire.
100pub fn decide(
101    q: &Question,
102    lease: Option<&Lease>,
103    seat_busy: bool,
104    default_timeout: u64,
105    now: Timestamp,
106) -> Action {
107    decide_owned(q, lease, seat_busy, default_timeout, now, false)
108}
109
110/// [`decide`], told whether the daemon owns the answer's action.
111///
112/// `daemon_owns_action` is true when the daemon will apply the action (the
113/// task exists and is in a state `daemon::decide_action` acts on). An answer
114/// that carries a [`crate::ask::ChoiceAction`] is then the daemon's alone
115/// (`daemon::apply_choice_actions`, idempotent through `Task::actions_applied`
116/// under the task's claim): resuming the dead seat here as well would run the
117/// same seat twice when the action is a resume or a requeue. With no task
118/// there is no one to apply it, so the words are delivered as before.
119pub fn decide_owned(
120    q: &Question,
121    lease: Option<&Lease>,
122    seat_busy: bool,
123    default_timeout: u64,
124    now: Timestamp,
125    daemon_owns_action: bool,
126) -> Action {
127    if q.cwd.is_none() || lease.is_some_and(|l| l.fresh(now)) {
128        return Action::Idle;
129    }
130    match q.status {
131        QuestionStatus::Abandoned => Action::Idle,
132        QuestionStatus::Answered => match q.resolution() {
133            _ if daemon_owns_action && q.chosen_action().is_some() => Action::Idle,
134            Some(a) if !q.answer_delivered && !seat_busy => Action::Deliver(Word::Answered(a)),
135            _ => Action::Idle,
136        },
137        QuestionStatus::Open => {
138            if let Some(said) = q.unread_from_owner() {
139                if seat_busy {
140                    return Action::Idle;
141                }
142                return Action::Deliver(Word::Said(said));
143            }
144            let secs = if q.answer_timeout > 0 {
145                q.answer_timeout
146            } else {
147                default_timeout
148            };
149            let deadline = q.last_activity().saturating_add(secs as i64);
150            if now.as_second() > deadline {
151                Action::Expire
152            } else {
153                Action::Idle
154            }
155        }
156    }
157}
158
159/// Everything needed to resume one seat.
160struct Target {
161    coordination: Option<crate::config::Coordination>,
162    fetched_base: Option<crate::run::FetchedBase>,
163    spec: AgentSpec,
164    seat: SeatState,
165    cwd: PathBuf,
166    sessions: bool,
167    allow_write: bool,
168    language: String,
169}
170
171/// The waiter's state: the store it watches and what it must remember between
172/// ticks (only which inputs it has already failed on).
173pub struct Waiter {
174    store: Questions,
175    home: PathBuf,
176    /// The config the conductor runs under, for a question the conductor asked
177    /// (it has no run whose snapshot could say).
178    conduct_cfg: Option<Config>,
179    /// `answer_timeout` for a question that did not record its own.
180    default_timeout: u64,
181    /// Inputs already tried and failed, by question id: `(key, until)`.
182    memo: HashMap<String, (String, Instant)>,
183}
184
185impl Waiter {
186    /// A waiter over `store`, with `home` as the magi home its runs live under.
187    pub fn new(store: Questions, home: PathBuf, conduct_cfg: Option<Config>) -> Self {
188        let default_timeout = conduct_cfg
189            .as_ref()
190            .map_or(Config::default().graph.answer_timeout, |c| {
191                c.graph.answer_timeout
192            });
193        Self {
194            store,
195            home,
196            conduct_cfg,
197            default_timeout,
198            memo: HashMap::new(),
199        }
200    }
201
202    /// Is this question's asking seat still mid-turn?
203    fn seat_busy(&self, q: &Question, now: Timestamp) -> bool {
204        if q.run == crate::conduct::NODE {
205            return crate::conduct::busy(&self.home);
206        }
207        RunState::load_under(&q.run, &self.home).is_ok_and(|s| {
208            s.seats_active()
209                .any(|(k, a)| *k == q.seat && a.remaining_secs(now) > 0)
210        })
211    }
212
213    /// Whether the daemon applies this question's chosen action, so the
214    /// waiter must not deliver the answer (see [`decide_owned`]).
215    fn daemon_owns_action(&self, q: &Question) -> bool {
216        crate::daemon::task_of_question(&crate::queue::Queue::at(self.home.join("queue")).list(), q)
217            .is_some_and(|t| crate::daemon::action_standing(t, q).daemon_owns())
218    }
219
220    /// Look at every question once and act on what needs acting on. `halt` is
221    /// asked between questions and during a delivery: true means the daemon is
222    /// parking, and whatever is in flight is dropped, undelivered and
223    /// resumable.
224    pub async fn tick(&mut self, now: Timestamp, halt: &(dyn Fn() -> bool + Sync)) {
225        for q in self.store.list() {
226            if halt() {
227                return;
228            }
229            let lease = self.store.read_lease(&q.id);
230            let owned = self.daemon_owns_action(&q);
231            match decide_owned(&q, lease.as_ref(), false, self.default_timeout, now, owned) {
232                Action::Idle => {}
233                Action::Expire => self.expire(&q),
234                Action::Deliver(word) => {
235                    // A question with a deputy is the deputy's to read
236                    // (`crate::deputy`); resuming the conductor's own seat
237                    // here would fork a conversation that is not its own.
238                    if q.deputy.is_some() {
239                        // Unless that deputy is spent: then nobody will read
240                        // the word, and the deadline still has to retire the
241                        // question.
242                        if crate::deputy::exhausted_past_deadline(
243                            &q,
244                            crate::deputy::can_start(
245                                self.conduct_cfg.as_ref(),
246                                crate::deputy::agent_of(&q),
247                            ),
248                            self.default_timeout,
249                            now,
250                        ) {
251                            self.expire(&q);
252                        }
253                        continue;
254                    }
255                    if self.seat_busy(&q, now) {
256                        continue;
257                    }
258                    let key = format!("{}:{:?}", q.thread.len(), word);
259                    if matches!(self.memo.get(&q.id), Some((k, until)) if *k == key && Instant::now() < *until)
260                    {
261                        continue;
262                    }
263                    if let Err(e) = self.deliver(&q, &word, halt).await {
264                        tracing::warn!("question {}: {e:#}", q.short());
265                    }
266                }
267            }
268        }
269    }
270
271    /// Abandon an unanswered question whose deadline passed with nobody
272    /// waiting - the words the asker's own wait would have used.
273    fn expire(&self, q: &Question) {
274        let secs = if q.answer_timeout > 0 {
275            q.answer_timeout
276        } else {
277            self.default_timeout
278        };
279        let startable =
280            crate::deputy::can_start(self.conduct_cfg.as_ref(), crate::deputy::agent_of(q));
281        let why = format!("no answer within {}s of asking", secs.max(1));
282        let done = self.store.update(&q.id, |r| {
283            // Decided again on the record as it is now: the owner may have
284            // spoken since the tick read it, and a word given in time is
285            // delivered, not abandoned.
286            let now = Timestamp::now();
287            let lease = self.store.read_lease(&r.id);
288            if decide(r, lease.as_ref(), false, self.default_timeout, now) != Action::Expire
289                && !(lease.as_ref().is_none_or(|l| !l.fresh(now))
290                    && crate::deputy::exhausted_past_deadline(
291                        r,
292                        startable,
293                        self.default_timeout,
294                        now,
295                    ))
296            {
297                return Ok(false);
298            }
299            r.abandon(&why);
300            r.waiter = None;
301            Ok(true)
302        });
303        match done {
304            Ok((_, false)) => {}
305            Ok((_, true)) => {
306                self.store.drop_lease(&q.id);
307                tracing::warn!(
308                    "question {} went unanswered for {secs}s with nobody waiting; \
309                     it stays as the record of it",
310                    q.short()
311                );
312            }
313            Err(e) => tracing::warn!("could not expire question {}: {e:#}", q.short()),
314        }
315    }
316
317    /// Admission uses the very same target resolution as actual delivery.
318    pub fn can_resume(&self, q: &Question) -> bool {
319        self.target(q).is_ok()
320    }
321
322    /// Work out how to resume the seat, or why it cannot be.
323    fn target(&self, q: &Question) -> std::result::Result<Target, String> {
324        let cwd = q
325            .cwd
326            .as_deref()
327            .map(PathBuf::from)
328            .filter(|p| p.is_dir())
329            .ok_or("the directory the agent was working in is gone")?;
330        let (cfg, seat, allow_write, fetched_base) = if q.run == crate::conduct::NODE {
331            let cfg = self
332                .conduct_cfg
333                .clone()
334                .ok_or("the conductor's configuration is not available")?;
335            let seat = crate::conduct::load_seat(&self.home)
336                .ok_or("the conductor has no recorded session")?;
337            (cfg, seat, false, None)
338        } else {
339            let state = RunState::load_under(&q.run, &self.home)
340                .map_err(|_| "the run that asked is gone".to_owned())?;
341            if !state.status.resumable() {
342                return Err(format!(
343                    "the run that asked has already {}",
344                    state.status.as_str()
345                ));
346            }
347            let seat = state
348                .seats
349                .get(&q.seat)
350                .cloned()
351                .ok_or("the run has no record of the asking seat")?;
352            let write = matches!(q.node.as_str(), "implement" | "fix");
353            (state.config, seat, write, state.fetched_base)
354        };
355        let spec = cfg
356            .agent(&seat.agent)
357            .map_err(|_| format!("agent `{}` is no longer in the roster", seat.agent))?
358            .clone();
359        if !agent::has_session(spec.kind, &seat, cfg.graph.sessions) {
360            return Err("the seat has no session to resume".to_owned());
361        }
362        Ok(Target {
363            coordination: cfg.coordination.clone(),
364            fetched_base,
365            spec,
366            seat,
367            cwd,
368            sessions: cfg.graph.sessions,
369            allow_write,
370            language: cfg.graph.language.clone(),
371        })
372    }
373
374    /// Say why the owner's word cannot reach anybody, once per input.
375    fn refuse(&mut self, q: &Question, key: String, why: &str) {
376        tracing::warn!("question {} cannot be delivered: {why}", q.short());
377        notices::raise_in(
378            &self.home,
379            Notice::warn(
380                &format!("question:{}", q.id),
381                format!(
382                    "Question {} \"{}\": what you said cannot reach the agent \
383                     ({why}). It is recorded, but nothing will read it.",
384                    q.short(),
385                    q.summary
386                ),
387            ),
388        );
389        let _ = self.store.update(&q.id, |r| {
390            r.waiter = None;
391            Ok(())
392        });
393        self.memo
394            .insert(q.id.clone(), (key, Instant::now() + RETRY_AFTER));
395    }
396
397    /// Resume the asking seat with the owner's word.
398    async fn deliver(
399        &mut self,
400        q: &Question,
401        word: &Word,
402        halt: &(dyn Fn() -> bool + Sync),
403    ) -> Result<()> {
404        let key = format!("{}:{:?}", q.thread.len(), word);
405        let mut target = match self.target(q) {
406            Ok(t) => t,
407            Err(why) => {
408                self.refuse(q, key, &why);
409                return Ok(());
410            }
411        };
412
413        let claim = self.store.root().join(format!("{}.claim", q.id));
414        if !take_claim(&claim, DELIVERY_TIMEOUT + Duration::from_secs(60)) {
415            return Ok(());
416        }
417        let _release = Release(claim);
418
419        // Re-read under the claim: the asker may have come back, or the owner
420        // may have said more, since the tick's copy was taken.
421        let q = self.store.get(&q.id)?;
422        let now = Timestamp::now();
423        let lease = self.store.read_lease(&q.id);
424        let owned = self.daemon_owns_action(&q);
425        let Action::Deliver(word) =
426            decide_owned(&q, lease.as_ref(), false, self.default_timeout, now, owned)
427        else {
428            return Ok(());
429        };
430        let snapshot = q.thread.len();
431
432        // Ownership of an action answer is settled against the daemon's task
433        // claim: hold it while the lease goes up, so the daemon either sees the
434        // fresh lease or has already applied the action (and we step aside).
435        // A claim held by a running competition is not the daemon applying an
436        // action; that case is delivered, with the re-check below.
437        let queue = crate::queue::Queue::at(self.home.join("queue"));
438        let task_id = crate::daemon::task_of_question(&queue.list(), &q).map(|t| t.id.clone());
439        // Never go on without the claim: a failed claim means an attempt (or
440        // the daemon) holds the task, and whatever it does next decides who
441        // owns this answer. Step aside; the next tick judges it afresh.
442        let _task_claim = match (&task_id, q.chosen_action()) {
443            (Some(id), Some(_)) => match queue.claim(id) {
444                Ok(c) => Some(c),
445                Err(e) => {
446                    tracing::debug!(
447                        "question {}: task claim not available ({e:#}), deferring delivery",
448                        q.short()
449                    );
450                    return Ok(());
451                }
452            },
453            _ => None,
454        };
455        // Applied, stale and pending are all stop conditions; only a busy task
456        // (or no action at all) is delivered to.
457        let task_still_ours = || match &task_id {
458            Some(id) => queue
459                .get(id)
460                .map(|t| !crate::daemon::action_standing(&t, &q).daemon_owns())
461                .unwrap_or(true),
462            None => true,
463        };
464        if !task_still_ours() {
465            return Ok(());
466        }
467
468        self.store.beat(&q.id, WaiterKind::Daemon);
469        // Re-check after the lease is up: a daemon that decided before it
470        // existed has by now written its mark (or is blocked on our claim).
471        if !task_still_ours() {
472            return Ok(());
473        }
474        self.store.update(&q.id, |r| {
475            r.waiter = Some(Note {
476                kind: WaiterKind::Daemon,
477                since: now,
478            });
479            Ok(())
480        })?;
481        tracing::info!(
482            "question {}: the asker is gone, resuming seat {} with the owner's word",
483            q.short(),
484            q.seat
485        );
486
487        let thread: Vec<(&str, &str)> = q
488            .thread
489            .iter()
490            .map(|t| {
491                (
492                    if t.who == Who::Operator {
493                        "operator"
494                    } else {
495                        "agent"
496                    },
497                    t.body.as_str(),
498                )
499            })
500            .collect();
501        // What the owner said and the agent has not read has its own section in
502        // the prompt, so the recap stops at what it already read.
503        let shown = match word {
504            Word::Said(_) => &thread[..q.delivered_turns.min(thread.len())],
505            Word::Answered(_) => &thread[..],
506        };
507        let owner_word = match &word {
508            Word::Said(s) => prompt::OwnerWord::Said(s),
509            Word::Answered(a) => prompt::OwnerWord::Answered(a),
510        };
511        let body = prompt::question_resumed(
512            &q.id,
513            &q.summary,
514            &q.detail,
515            shown,
516            &owner_word,
517            &target.language,
518        );
519
520        let artifacts = self.store.root().join(format!("{}.delivery", q.id));
521        let stem = format!("deliver-{}", now.as_second());
522        let cache_dir = None;
523        let inv = Invocation {
524            coordination: target.coordination.as_ref(),
525            fetched_base: target.fetched_base.as_ref(),
526            cwd: &target.cwd,
527            prompt: &body,
528            timeout: DELIVERY_TIMEOUT,
529            allow_write: target.allow_write,
530            unsandboxed: false,
531            sessions: target.sessions,
532            artifacts: &artifacts,
533            stem: &stem,
534            run: &q.run,
535            node: &q.node,
536            cache_dir,
537            attachments: &[],
538            writable: &[],
539        };
540
541        let store = self.store.clone();
542        let id = q.id.clone();
543        let out = {
544            let fut = agent::invoke(&target.spec, &mut target.seat, &inv);
545            tokio::pin!(fut);
546            let mut beat = tokio::time::interval(Duration::from_secs(1));
547            let mut beats = 0u32;
548            loop {
549                tokio::select! {
550                    r = &mut fut => break Some(r),
551                    _ = beat.tick() => {
552                        if halt() {
553                            break None;
554                        }
555                        beats += 1;
556                        if beats % 20 == 0 {
557                            store.beat(&id, WaiterKind::Daemon);
558                        }
559                    }
560                }
561            }
562        };
563
564        let Some(out) = out else {
565            // Parking: leave everything as it was, resumable.
566            let _ = self.store.update(&q.id, |r| {
567                r.waiter = None;
568                Ok(())
569            });
570            return Ok(());
571        };
572        let out = match out {
573            Ok(o) if o.usable() => o,
574            other => {
575                let why = match other {
576                    Ok(o) if o.timed_out => "the resumed turn timed out".to_owned(),
577                    Ok(o) if o.quota_exhausted() => "the agent is out of quota".to_owned(),
578                    Ok(o) => format!(
579                        "the resumed turn failed ({})",
580                        notices::exit_status(o.exit_code)
581                    ),
582                    Err(e) => format!("the agent could not be started: {e:#}"),
583                };
584                self.refuse(&q, key, &why);
585                return Ok(());
586            }
587        };
588
589        let text = out.text.trim().to_owned();
590        self.store.update(&q.id, |r| {
591            r.waiter = None;
592            match &word {
593                Word::Answered(_) => r.answer_delivered = true,
594                Word::Said(_) => {
595                    r.delivered_turns = r.delivered_turns.max(snapshot);
596                    // An agent that answered in prose instead of calling
597                    // `magi ask --thread` still spoke; keep it on the record so
598                    // the owner reads it. If it did use --thread, or the owner
599                    // said more meanwhile, the record already moved on.
600                    let agent_spoke = r.thread[snapshot.min(r.thread.len())..]
601                        .iter()
602                        .any(|t| t.who == Who::Agent);
603                    if !agent_spoke && r.thread.len() == snapshot && r.status.open() {
604                        let choices = r.choices.clone();
605                        r.reply(text, choices)?;
606                    }
607                }
608            }
609            Ok(())
610        })?;
611        self.memo.remove(&q.id);
612        Ok(())
613    }
614}
615
616/// Take the delivery claim, so two waiters (`magi serve` twice, or a stray
617/// second daemon) cannot resume the same seat at once. A claim older than a
618/// whole delivery plus slack was left by a process that died.
619pub(crate) fn take_claim(path: &std::path::Path, stale_after: Duration) -> bool {
620    let attempt = || {
621        std::fs::OpenOptions::new()
622            .write(true)
623            .create_new(true)
624            .open(path)
625            .is_ok()
626    };
627    if attempt() {
628        return true;
629    }
630    let stale = std::fs::metadata(path)
631        .and_then(|m| m.modified())
632        .ok()
633        .and_then(|t| t.elapsed().ok())
634        .is_some_and(|age| age > stale_after);
635    if stale {
636        let _ = std::fs::remove_file(path);
637        return attempt();
638    }
639    false
640}
641
642/// Removes the delivery claim when the delivery ends, however it ends.
643pub(crate) struct Release(pub(crate) PathBuf);
644
645impl Drop for Release {
646    fn drop(&mut self) {
647        let _ = std::fs::remove_file(&self.0);
648    }
649}
650
651/// The waiter's loop, until `stop` is asked for.
652///
653/// A stop ends the loop between ticks; a park also drops a delivery in flight.
654/// Neither writes anything to a question: the lease just ages out and the next
655/// waiter - this daemon restarted, or another - reads the same state off disk.
656pub async fn run(mut waiter: Waiter, stop: crate::daemon::Stop) {
657    let halt = {
658        let stop = stop.clone();
659        move || stop.parking()
660    };
661    while !stop.stopped() {
662        waiter.tick(Timestamp::now(), &halt).await;
663        tokio::time::sleep(TICK).await;
664    }
665}
666
667#[cfg(test)]
668mod tests {
669    use super::*;
670
671    fn ts(secs: i64) -> Timestamp {
672        Timestamp::from_second(secs).unwrap()
673    }
674
675    fn asked(at: i64) -> Question {
676        let mut q = Question::new(
677            "run".into(),
678            "implement".into(),
679            "impl-A".into(),
680            "which?".into(),
681            String::new(),
682            vec![],
683        );
684        q.cwd = Some("/tmp".into());
685        q.asked_at = ts(at);
686        q.answer_timeout = 1000;
687        q
688    }
689
690    fn lease(at: i64) -> Lease {
691        Lease {
692            kind: WaiterKind::Asker,
693            pid: 1,
694            beat_at: ts(at),
695        }
696    }
697
698    #[test]
699    fn a_fresh_lease_means_nobody_else_acts() {
700        let mut q = asked(0);
701        q.say("why?").unwrap();
702        assert_eq!(
703            decide(&q, Some(&lease(95)), false, 86_400, ts(100)),
704            Action::Idle
705        );
706    }
707
708    #[test]
709    fn a_stale_lease_delivers_what_the_owner_said() {
710        let mut q = asked(0);
711        q.say("why?").unwrap();
712        assert_eq!(
713            decide(&q, Some(&lease(0)), false, 86_400, ts(500)),
714            Action::Deliver(Word::Said("why?".into()))
715        );
716        assert_eq!(
717            decide(&q, None, true, 86_400, ts(500)),
718            Action::Idle,
719            "a seat still mid-turn is not resumed"
720        );
721    }
722
723    #[test]
724    fn an_answer_nobody_read_is_delivered_once() {
725        let mut q = asked(0);
726        q.choices = vec!["A".into()];
727        q.answer(crate::ask::Answer::Choice("A".into())).unwrap();
728        assert_eq!(
729            decide(&q, None, false, 86_400, ts(10)),
730            Action::Deliver(Word::Answered("A".into()))
731        );
732        q.answer_delivered = true;
733        assert_eq!(decide(&q, None, false, 86_400, ts(10)), Action::Idle);
734    }
735
736    #[test]
737    fn only_asked_questions_and_only_before_the_deadline() {
738        let mut q = asked(0);
739        assert_eq!(decide(&q, None, false, 86_400, ts(999)), Action::Idle);
740        assert_eq!(decide(&q, None, false, 86_400, ts(1001)), Action::Expire);
741        q.cwd = None;
742        assert_eq!(decide(&q, None, false, 86_400, ts(1001)), Action::Idle);
743    }
744
745    #[test]
746    fn the_deadline_runs_from_the_last_turn_not_from_asking() {
747        let mut q = asked(0);
748        q.thread.push(crate::ask::Turn {
749            who: Who::Agent,
750            body: "context".into(),
751            at: ts(4000),
752            note: None,
753        });
754        q.delivered_turns = 1;
755        assert_eq!(decide(&q, None, false, 86_400, ts(4500)), Action::Idle);
756        assert_eq!(decide(&q, None, false, 86_400, ts(5001)), Action::Expire);
757    }
758
759    #[test]
760    fn a_word_given_in_time_is_delivered_after_the_deadline() {
761        let mut q = asked(0);
762        q.say("wait").unwrap();
763        assert!(matches!(
764            decide(&q, None, false, 86_400, ts(5000)),
765            Action::Deliver(_)
766        ));
767    }
768
769    #[test]
770    fn an_action_answer_is_left_to_the_daemon() {
771        let mut q = asked(0);
772        q.choices = vec!["A".into()];
773        q.actions
774            .insert("A".into(), crate::ask::ChoiceAction::Requeue);
775        q.answer(crate::ask::Answer::Choice("A".into())).unwrap();
776        assert_eq!(
777            decide_owned(&q, None, false, 86_400, ts(10), true),
778            Action::Idle,
779            "the daemon applies the action; the waiter must not also resume the seat"
780        );
781        // Delivered or not, a stale lease changes nothing.
782        q.answer_delivered = true;
783        assert_eq!(
784            decide_owned(&q, None, false, 86_400, ts(10), true),
785            Action::Idle
786        );
787        // No task to apply it: the words are delivered as before.
788        q.answer_delivered = false;
789        assert_eq!(
790            decide_owned(&q, None, false, 86_400, ts(10), false),
791            Action::Deliver(Word::Answered("A".into()))
792        );
793        // No action: delivered even when a task exists.
794        let mut plain = asked(0);
795        plain.choices = vec!["A".into()];
796        plain
797            .answer(crate::ask::Answer::Choice("A".into()))
798            .unwrap();
799        assert_eq!(
800            decide_owned(&plain, None, false, 86_400, ts(10), true),
801            Action::Deliver(Word::Answered("A".into()))
802        );
803    }
804}