magi-cli 0.80.0

Blind multi-agent implementation competition: N agents implement, M judges rank blind, deliberate, vote privately, winner survives double review + E2E gate
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
//! The waiter: something that keeps waiting on a question after the process
//! that asked it is gone.
//!
//! `magi ask` blocks in the agent's own shell tool, and that tool kills a wait
//! that runs long (see `ask::WAIT_SLICE`). An agent can also background the
//! call and simply finish. Either way the question outlives the only process
//! that would have read the owner's answer, and the owner's reply lands in a
//! file nobody is polling - the phone then says "waiting for the agent" while
//! there is no agent to wait for.
//!
//! This module is the guarantee that something is always on the other end. It
//! runs inside `magi serve`, on its own cadence rather than the queue's, and
//! for every question a `magi ask` filed it decides one of three things:
//!
//! - **the asker is still there** (its [`Lease`] is fresh): do nothing. A
//!   second agent must never be started while the first is blocked;
//! - **the asker is gone and the owner has said or answered something the agent
//!   has not read**: resume the *same seat's* CLI session with the question,
//!   the thread so far and the owner's words, so the agent that holds the
//!   context carries on;
//! - **nobody answered before `answer_timeout`**: abandon it, in the same words
//!   the asker would have used.
//!
//! # Shape
//!
//! Same split as [`crate::queue`] and [`crate::ask`]: [`decide`] is pure - a
//! question, a lease, a clock and a bool in, an [`Action`] out - and every
//! filesystem or process effect is in [`Waiter::tick`]. State lives on disk (the
//! question record and its lease), so a restarted daemon picks up exactly where
//! the last one stopped, including an answer nobody has read yet.
//!
//! # What it will not do
//!
//! It never starts a fresh consultant. If the seat's session cannot be resumed
//! (`agent::has_session` says no, the working directory is gone, the run is
//! over, the agent left the roster) the question stays as it is, the owner
//! gets a notification saying why, and nothing else runs. A new agent would
//! have to be caught up on everything the first one knew, and would act on the
//! owner's words without the context they were written against.
//!
//! It never assumes a resume worked either: a turn that fails or times out
//! leaves the word undelivered, and a later tick tries again.

use std::collections::HashMap;
use std::path::PathBuf;
use std::time::{Duration, Instant};

use anyhow::Result;
use jiff::Timestamp;

use crate::agent::{self, Invocation, SeatState};
use crate::ask::{Lease, Question, QuestionStatus, Questions, Waiter as Note, WaiterKind, Who};
use crate::config::{AgentSpec, Config};
use crate::notices::{self, Notice};
use crate::prompt;
use crate::run::RunState;

/// How often the waiter looks at the store.
pub const TICK: Duration = Duration::from_secs(5);

/// Wall-clock limit for one resumed turn. A resumed agent may go on to do real
/// work once it has the owner's word, but it is one conversation turn, not a
/// whole node.
const DELIVERY_TIMEOUT: Duration = Duration::from_secs(15 * 60);

/// How long a failed or refused delivery is left alone before the same input
/// is tried again. Long enough not to hammer a seat that is out of quota,
/// short enough that a fixed problem does not wait a day.
const RETRY_AFTER: Duration = Duration::from_secs(10 * 60);

/// What the agent has not read yet.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Word {
    /// The owner spoke back without deciding.
    Said(String),
    /// The owner decided.
    Answered(String),
}

/// What to do about one question, this tick.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Action {
    /// Nothing: not ours, someone is waiting, or nothing new to hand over.
    Idle,
    /// The deadline passed with no answer and nobody waiting.
    Expire,
    /// Resume the asking seat with this.
    Deliver(Word),
}

/// The whole decision, with no I/O.
///
/// `seat_busy` is whether the asking seat's CLI is still mid-turn - the agent
/// that exited its `magi ask` may not have exited itself, and resuming a
/// conversation that is running would fork it.
///
/// Order matters: a word the owner gave before the deadline is delivered even
/// if the deadline has passed by the time the waiter looks, and only then does
/// an unanswered question expire.
pub fn decide(
    q: &Question,
    lease: Option<&Lease>,
    seat_busy: bool,
    default_timeout: u64,
    now: Timestamp,
) -> Action {
    decide_owned(q, lease, seat_busy, default_timeout, now, false)
}

/// [`decide`], told whether the daemon owns the answer's action.
///
/// `daemon_owns_action` is true when the daemon will apply the action (the
/// task exists and is in a state `daemon::decide_action` acts on). An answer
/// that carries a [`crate::ask::ChoiceAction`] is then the daemon's alone
/// (`daemon::apply_choice_actions`, idempotent through `Task::actions_applied`
/// under the task's claim): resuming the dead seat here as well would run the
/// same seat twice when the action is a resume or a requeue. With no task
/// there is no one to apply it, so the words are delivered as before.
pub fn decide_owned(
    q: &Question,
    lease: Option<&Lease>,
    seat_busy: bool,
    default_timeout: u64,
    now: Timestamp,
    daemon_owns_action: bool,
) -> Action {
    if q.cwd.is_none() || lease.is_some_and(|l| l.fresh(now)) {
        return Action::Idle;
    }
    match q.status {
        QuestionStatus::Abandoned => Action::Idle,
        QuestionStatus::Answered => match q.resolution() {
            _ if daemon_owns_action && q.chosen_action().is_some() => Action::Idle,
            Some(a) if !q.answer_delivered && !seat_busy => Action::Deliver(Word::Answered(a)),
            _ => Action::Idle,
        },
        QuestionStatus::Open => {
            if let Some(said) = q.unread_from_owner() {
                if seat_busy {
                    return Action::Idle;
                }
                return Action::Deliver(Word::Said(said));
            }
            let secs = if q.answer_timeout > 0 {
                q.answer_timeout
            } else {
                default_timeout
            };
            let deadline = q.last_activity().saturating_add(secs as i64);
            if now.as_second() > deadline {
                Action::Expire
            } else {
                Action::Idle
            }
        }
    }
}

/// Everything needed to resume one seat.
struct Target {
    spec: AgentSpec,
    seat: SeatState,
    cwd: PathBuf,
    sessions: bool,
    allow_write: bool,
    language: String,
}

/// The waiter's state: the store it watches and what it must remember between
/// ticks (only which inputs it has already failed on).
pub struct Waiter {
    store: Questions,
    home: PathBuf,
    /// The config the conductor runs under, for a question the conductor asked
    /// (it has no run whose snapshot could say).
    conduct_cfg: Option<Config>,
    /// `answer_timeout` for a question that did not record its own.
    default_timeout: u64,
    /// Inputs already tried and failed, by question id: `(key, until)`.
    memo: HashMap<String, (String, Instant)>,
}

impl Waiter {
    /// A waiter over `store`, with `home` as the magi home its runs live under.
    pub fn new(store: Questions, home: PathBuf, conduct_cfg: Option<Config>) -> Self {
        let default_timeout = conduct_cfg
            .as_ref()
            .map_or(Config::default().graph.answer_timeout, |c| {
                c.graph.answer_timeout
            });
        Self {
            store,
            home,
            conduct_cfg,
            default_timeout,
            memo: HashMap::new(),
        }
    }

    /// Is this question's asking seat still mid-turn?
    fn seat_busy(&self, q: &Question, now: Timestamp) -> bool {
        if q.run == crate::conduct::NODE {
            return crate::conduct::busy(&self.home);
        }
        RunState::load_under(&q.run, &self.home).is_ok_and(|s| {
            s.seats_active()
                .any(|(k, a)| *k == q.seat && a.remaining_secs(now) > 0)
        })
    }

    /// Whether the daemon applies this question's chosen action, so the
    /// waiter must not deliver the answer (see [`decide_owned`]).
    fn daemon_owns_action(&self, q: &Question) -> bool {
        crate::daemon::task_of_question(&crate::queue::Queue::at(self.home.join("queue")).list(), q)
            .is_some_and(|t| crate::daemon::action_standing(t, q).daemon_owns())
    }

    /// Look at every question once and act on what needs acting on. `halt` is
    /// asked between questions and during a delivery: true means the daemon is
    /// parking, and whatever is in flight is dropped, undelivered and
    /// resumable.
    pub async fn tick(&mut self, now: Timestamp, halt: &(dyn Fn() -> bool + Sync)) {
        for q in self.store.list() {
            if halt() {
                return;
            }
            let lease = self.store.read_lease(&q.id);
            let owned = self.daemon_owns_action(&q);
            match decide_owned(&q, lease.as_ref(), false, self.default_timeout, now, owned) {
                Action::Idle => {}
                Action::Expire => self.expire(&q),
                Action::Deliver(word) => {
                    // A question with a deputy is the deputy's to read
                    // (`crate::deputy`); resuming the conductor's own seat
                    // here would fork a conversation that is not its own.
                    if q.deputy.is_some() {
                        // Unless that deputy is spent: then nobody will read
                        // the word, and the deadline still has to retire the
                        // question.
                        if crate::deputy::exhausted_past_deadline(
                            &q,
                            crate::deputy::can_start(
                                self.conduct_cfg.as_ref(),
                                crate::deputy::agent_of(&q),
                            ),
                            self.default_timeout,
                            now,
                        ) {
                            self.expire(&q);
                        }
                        continue;
                    }
                    if self.seat_busy(&q, now) {
                        continue;
                    }
                    let key = format!("{}:{:?}", q.thread.len(), word);
                    if matches!(self.memo.get(&q.id), Some((k, until)) if *k == key && Instant::now() < *until)
                    {
                        continue;
                    }
                    if let Err(e) = self.deliver(&q, &word, halt).await {
                        tracing::warn!("question {}: {e:#}", q.short());
                    }
                }
            }
        }
    }

    /// Abandon an unanswered question whose deadline passed with nobody
    /// waiting - the words the asker's own wait would have used.
    fn expire(&self, q: &Question) {
        let secs = if q.answer_timeout > 0 {
            q.answer_timeout
        } else {
            self.default_timeout
        };
        let startable =
            crate::deputy::can_start(self.conduct_cfg.as_ref(), crate::deputy::agent_of(q));
        let why = format!("no answer within {}s of asking", secs.max(1));
        let done = self.store.update(&q.id, |r| {
            // Decided again on the record as it is now: the owner may have
            // spoken since the tick read it, and a word given in time is
            // delivered, not abandoned.
            let now = Timestamp::now();
            let lease = self.store.read_lease(&r.id);
            if decide(r, lease.as_ref(), false, self.default_timeout, now) != Action::Expire
                && !(lease.as_ref().is_none_or(|l| !l.fresh(now))
                    && crate::deputy::exhausted_past_deadline(
                        r,
                        startable,
                        self.default_timeout,
                        now,
                    ))
            {
                return Ok(false);
            }
            r.abandon(&why);
            r.waiter = None;
            Ok(true)
        });
        match done {
            Ok((_, false)) => {}
            Ok((_, true)) => {
                self.store.drop_lease(&q.id);
                tracing::warn!(
                    "question {} went unanswered for {secs}s with nobody waiting; \
                     it stays as the record of it",
                    q.short()
                );
            }
            Err(e) => tracing::warn!("could not expire question {}: {e:#}", q.short()),
        }
    }

    /// Work out how to resume the seat, or why it cannot be.
    fn target(&self, q: &Question) -> std::result::Result<Target, String> {
        let cwd = q
            .cwd
            .as_deref()
            .map(PathBuf::from)
            .filter(|p| p.is_dir())
            .ok_or("the directory the agent was working in is gone")?;
        let (cfg, seat, allow_write) = if q.run == crate::conduct::NODE {
            let cfg = self
                .conduct_cfg
                .clone()
                .ok_or("the conductor's configuration is not available")?;
            let seat = crate::conduct::load_seat(&self.home)
                .ok_or("the conductor has no recorded session")?;
            (cfg, seat, false)
        } else {
            let state = RunState::load_under(&q.run, &self.home)
                .map_err(|_| "the run that asked is gone".to_owned())?;
            if !state.status.resumable() {
                return Err(format!(
                    "the run that asked has already {}",
                    state.status.as_str()
                ));
            }
            let seat = state
                .seats
                .get(&q.seat)
                .cloned()
                .ok_or("the run has no record of the asking seat")?;
            let write = matches!(q.node.as_str(), "implement" | "fix");
            (state.config, seat, write)
        };
        let spec = cfg
            .agent(&seat.agent)
            .map_err(|_| format!("agent `{}` is no longer in the roster", seat.agent))?
            .clone();
        if !agent::has_session(spec.kind, &seat, cfg.graph.sessions) {
            return Err("the seat has no session to resume".to_owned());
        }
        Ok(Target {
            spec,
            seat,
            cwd,
            sessions: cfg.graph.sessions,
            allow_write,
            language: cfg.graph.language.clone(),
        })
    }

    /// Say why the owner's word cannot reach anybody, once per input.
    fn refuse(&mut self, q: &Question, key: String, why: &str) {
        tracing::warn!("question {} cannot be delivered: {why}", q.short());
        notices::raise_in(
            &self.home,
            Notice::warn(
                &format!("question:{}", q.id),
                format!(
                    "Question {} \"{}\": what you said cannot reach the agent \
                     ({why}). It is recorded, but nothing will read it.",
                    q.short(),
                    q.summary
                ),
            ),
        );
        let _ = self.store.update(&q.id, |r| {
            r.waiter = None;
            Ok(())
        });
        self.memo
            .insert(q.id.clone(), (key, Instant::now() + RETRY_AFTER));
    }

    /// Resume the asking seat with the owner's word.
    async fn deliver(
        &mut self,
        q: &Question,
        word: &Word,
        halt: &(dyn Fn() -> bool + Sync),
    ) -> Result<()> {
        let key = format!("{}:{:?}", q.thread.len(), word);
        let mut target = match self.target(q) {
            Ok(t) => t,
            Err(why) => {
                self.refuse(q, key, &why);
                return Ok(());
            }
        };

        let claim = self.store.root().join(format!("{}.claim", q.id));
        if !take_claim(&claim, DELIVERY_TIMEOUT + Duration::from_secs(60)) {
            return Ok(());
        }
        let _release = Release(claim);

        // Re-read under the claim: the asker may have come back, or the owner
        // may have said more, since the tick's copy was taken.
        let q = self.store.get(&q.id)?;
        let now = Timestamp::now();
        let lease = self.store.read_lease(&q.id);
        let owned = self.daemon_owns_action(&q);
        let Action::Deliver(word) =
            decide_owned(&q, lease.as_ref(), false, self.default_timeout, now, owned)
        else {
            return Ok(());
        };
        let snapshot = q.thread.len();

        // Ownership of an action answer is settled against the daemon's task
        // claim: hold it while the lease goes up, so the daemon either sees the
        // fresh lease or has already applied the action (and we step aside).
        // A claim held by a running competition is not the daemon applying an
        // action; that case is delivered, with the re-check below.
        let queue = crate::queue::Queue::at(self.home.join("queue"));
        let task_id = crate::daemon::task_of_question(&queue.list(), &q).map(|t| t.id.clone());
        // Never go on without the claim: a failed claim means an attempt (or
        // the daemon) holds the task, and whatever it does next decides who
        // owns this answer. Step aside; the next tick judges it afresh.
        let _task_claim = match (&task_id, q.chosen_action()) {
            (Some(id), Some(_)) => match queue.claim(id) {
                Ok(c) => Some(c),
                Err(e) => {
                    tracing::debug!(
                        "question {}: task claim not available ({e:#}), deferring delivery",
                        q.short()
                    );
                    return Ok(());
                }
            },
            _ => None,
        };
        // Applied, stale and pending are all stop conditions; only a busy task
        // (or no action at all) is delivered to.
        let task_still_ours = || match &task_id {
            Some(id) => queue
                .get(id)
                .map(|t| !crate::daemon::action_standing(&t, &q).daemon_owns())
                .unwrap_or(true),
            None => true,
        };
        if !task_still_ours() {
            return Ok(());
        }

        self.store.beat(&q.id, WaiterKind::Daemon);
        // Re-check after the lease is up: a daemon that decided before it
        // existed has by now written its mark (or is blocked on our claim).
        if !task_still_ours() {
            return Ok(());
        }
        self.store.update(&q.id, |r| {
            r.waiter = Some(Note {
                kind: WaiterKind::Daemon,
                since: now,
            });
            Ok(())
        })?;
        tracing::info!(
            "question {}: the asker is gone, resuming seat {} with the owner's word",
            q.short(),
            q.seat
        );

        let thread: Vec<(&str, &str)> = q
            .thread
            .iter()
            .map(|t| {
                (
                    if t.who == Who::Operator {
                        "operator"
                    } else {
                        "agent"
                    },
                    t.body.as_str(),
                )
            })
            .collect();
        // What the owner said and the agent has not read has its own section in
        // the prompt, so the recap stops at what it already read.
        let shown = match word {
            Word::Said(_) => &thread[..q.delivered_turns.min(thread.len())],
            Word::Answered(_) => &thread[..],
        };
        let owner_word = match &word {
            Word::Said(s) => prompt::OwnerWord::Said(s),
            Word::Answered(a) => prompt::OwnerWord::Answered(a),
        };
        let body = prompt::question_resumed(
            &q.id,
            &q.summary,
            &q.detail,
            shown,
            &owner_word,
            &target.language,
        );

        let artifacts = self.store.root().join(format!("{}.delivery", q.id));
        let stem = format!("deliver-{}", now.as_second());
        let cache_dir = None;
        let inv = Invocation {
            cwd: &target.cwd,
            prompt: &body,
            timeout: DELIVERY_TIMEOUT,
            allow_write: target.allow_write,
            sessions: target.sessions,
            artifacts: &artifacts,
            stem: &stem,
            run: &q.run,
            node: &q.node,
            cache_dir,
            attachments: &[],
            writable: &[],
        };

        let store = self.store.clone();
        let id = q.id.clone();
        let out = {
            let fut = agent::invoke(&target.spec, &mut target.seat, &inv);
            tokio::pin!(fut);
            let mut beat = tokio::time::interval(Duration::from_secs(1));
            let mut beats = 0u32;
            loop {
                tokio::select! {
                    r = &mut fut => break Some(r),
                    _ = beat.tick() => {
                        if halt() {
                            break None;
                        }
                        beats += 1;
                        if beats % 20 == 0 {
                            store.beat(&id, WaiterKind::Daemon);
                        }
                    }
                }
            }
        };

        let Some(out) = out else {
            // Parking: leave everything as it was, resumable.
            let _ = self.store.update(&q.id, |r| {
                r.waiter = None;
                Ok(())
            });
            return Ok(());
        };
        let out = match out {
            Ok(o) if o.usable() => o,
            other => {
                let why = match other {
                    Ok(o) if o.timed_out => "the resumed turn timed out".to_owned(),
                    Ok(o) if o.quota_exhausted() => "the agent is out of quota".to_owned(),
                    Ok(o) => format!("the resumed turn failed (exit {:?})", o.exit_code),
                    Err(e) => format!("the agent could not be started: {e:#}"),
                };
                self.refuse(&q, key, &why);
                return Ok(());
            }
        };

        let text = out.text.trim().to_owned();
        self.store.update(&q.id, |r| {
            r.waiter = None;
            match &word {
                Word::Answered(_) => r.answer_delivered = true,
                Word::Said(_) => {
                    r.delivered_turns = r.delivered_turns.max(snapshot);
                    // An agent that answered in prose instead of calling
                    // `magi ask --thread` still spoke; keep it on the record so
                    // the owner reads it. If it did use --thread, or the owner
                    // said more meanwhile, the record already moved on.
                    let agent_spoke = r.thread[snapshot.min(r.thread.len())..]
                        .iter()
                        .any(|t| t.who == Who::Agent);
                    if !agent_spoke && r.thread.len() == snapshot && r.status.open() {
                        let choices = r.choices.clone();
                        r.reply(text, choices)?;
                    }
                }
            }
            Ok(())
        })?;
        self.memo.remove(&q.id);
        Ok(())
    }
}

/// Take the delivery claim, so two waiters (`magi serve` twice, or a stray
/// second daemon) cannot resume the same seat at once. A claim older than a
/// whole delivery plus slack was left by a process that died.
pub(crate) fn take_claim(path: &std::path::Path, stale_after: Duration) -> bool {
    let attempt = || {
        std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(path)
            .is_ok()
    };
    if attempt() {
        return true;
    }
    let stale = std::fs::metadata(path)
        .and_then(|m| m.modified())
        .ok()
        .and_then(|t| t.elapsed().ok())
        .is_some_and(|age| age > stale_after);
    if stale {
        let _ = std::fs::remove_file(path);
        return attempt();
    }
    false
}

/// Removes the delivery claim when the delivery ends, however it ends.
pub(crate) struct Release(pub(crate) PathBuf);

impl Drop for Release {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.0);
    }
}

/// The waiter's loop, until `stop` is asked for.
///
/// A stop ends the loop between ticks; a park also drops a delivery in flight.
/// Neither writes anything to a question: the lease just ages out and the next
/// waiter - this daemon restarted, or another - reads the same state off disk.
pub async fn run(mut waiter: Waiter, stop: crate::daemon::Stop) {
    let halt = {
        let stop = stop.clone();
        move || stop.parking()
    };
    while !stop.stopped() {
        waiter.tick(Timestamp::now(), &halt).await;
        tokio::time::sleep(TICK).await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn ts(secs: i64) -> Timestamp {
        Timestamp::from_second(secs).unwrap()
    }

    fn asked(at: i64) -> Question {
        let mut q = Question::new(
            "run".into(),
            "implement".into(),
            "impl-A".into(),
            "which?".into(),
            String::new(),
            vec![],
        );
        q.cwd = Some("/tmp".into());
        q.asked_at = ts(at);
        q.answer_timeout = 1000;
        q
    }

    fn lease(at: i64) -> Lease {
        Lease {
            kind: WaiterKind::Asker,
            pid: 1,
            beat_at: ts(at),
        }
    }

    #[test]
    fn a_fresh_lease_means_nobody_else_acts() {
        let mut q = asked(0);
        q.say("why?").unwrap();
        assert_eq!(
            decide(&q, Some(&lease(95)), false, 86_400, ts(100)),
            Action::Idle
        );
    }

    #[test]
    fn a_stale_lease_delivers_what_the_owner_said() {
        let mut q = asked(0);
        q.say("why?").unwrap();
        assert_eq!(
            decide(&q, Some(&lease(0)), false, 86_400, ts(500)),
            Action::Deliver(Word::Said("why?".into()))
        );
        assert_eq!(
            decide(&q, None, true, 86_400, ts(500)),
            Action::Idle,
            "a seat still mid-turn is not resumed"
        );
    }

    #[test]
    fn an_answer_nobody_read_is_delivered_once() {
        let mut q = asked(0);
        q.choices = vec!["A".into()];
        q.answer(crate::ask::Answer::Choice("A".into())).unwrap();
        assert_eq!(
            decide(&q, None, false, 86_400, ts(10)),
            Action::Deliver(Word::Answered("A".into()))
        );
        q.answer_delivered = true;
        assert_eq!(decide(&q, None, false, 86_400, ts(10)), Action::Idle);
    }

    #[test]
    fn only_asked_questions_and_only_before_the_deadline() {
        let mut q = asked(0);
        assert_eq!(decide(&q, None, false, 86_400, ts(999)), Action::Idle);
        assert_eq!(decide(&q, None, false, 86_400, ts(1001)), Action::Expire);
        q.cwd = None;
        assert_eq!(decide(&q, None, false, 86_400, ts(1001)), Action::Idle);
    }

    #[test]
    fn the_deadline_runs_from_the_last_turn_not_from_asking() {
        let mut q = asked(0);
        q.thread.push(crate::ask::Turn {
            who: Who::Agent,
            body: "context".into(),
            at: ts(4000),
        });
        q.delivered_turns = 1;
        assert_eq!(decide(&q, None, false, 86_400, ts(4500)), Action::Idle);
        assert_eq!(decide(&q, None, false, 86_400, ts(5001)), Action::Expire);
    }

    #[test]
    fn a_word_given_in_time_is_delivered_after_the_deadline() {
        let mut q = asked(0);
        q.say("wait").unwrap();
        assert!(matches!(
            decide(&q, None, false, 86_400, ts(5000)),
            Action::Deliver(_)
        ));
    }

    #[test]
    fn an_action_answer_is_left_to_the_daemon() {
        let mut q = asked(0);
        q.choices = vec!["A".into()];
        q.actions
            .insert("A".into(), crate::ask::ChoiceAction::Requeue);
        q.answer(crate::ask::Answer::Choice("A".into())).unwrap();
        assert_eq!(
            decide_owned(&q, None, false, 86_400, ts(10), true),
            Action::Idle,
            "the daemon applies the action; the waiter must not also resume the seat"
        );
        // Delivered or not, a stale lease changes nothing.
        q.answer_delivered = true;
        assert_eq!(
            decide_owned(&q, None, false, 86_400, ts(10), true),
            Action::Idle
        );
        // No task to apply it: the words are delivered as before.
        q.answer_delivered = false;
        assert_eq!(
            decide_owned(&q, None, false, 86_400, ts(10), false),
            Action::Deliver(Word::Answered("A".into()))
        );
        // No action: delivered even when a task exists.
        let mut plain = asked(0);
        plain.choices = vec!["A".into()];
        plain
            .answer(crate::ask::Answer::Choice("A".into()))
            .unwrap();
        assert_eq!(
            decide_owned(&plain, None, false, 86_400, ts(10), true),
            Action::Deliver(Word::Answered("A".into()))
        );
    }
}