Skip to main content

magi/
direct.rs

1//! Runs started by hand are tasks.
2//!
3//! `magi run` and `magi review` used to mint a run no task knew about: no
4//! detail page, no attempt history, invisible to duplicate detection, and an
5//! orphan the moment anything cleaned up after it. They now file a task first,
6//! through the queue like `magi task add`, and then either hand it to a live
7//! loop or execute it here through the daemon's own attempt
8//! ([`crate::daemon::run_claimed`]). There is no second dispatcher in this
9//! module: it decides *who* runs the task and watches it, nothing more.
10//!
11//! * A live loop in another process (a fresh `daemon.json` heartbeat, see
12//!   [`crate::daemon::foreign_loop`]) gets the task as `urgent`, runs it in
13//!   its own urgent slot, and the caller [`follow`]s it. The caller never
14//!   claims or executes it: two processes must not drive one task.
15//! * Otherwise the caller claims the task **before it is written**, so a loop
16//!   that starts mid-run finds the claim and leaves it alone, and executes it.
17
18use std::path::{Path, PathBuf};
19use std::time::{Duration, Instant};
20
21use anyhow::{Context, Result, bail};
22use jiff::Timestamp;
23
24use crate::config::Config;
25use crate::queue::{Claim, Queue, RunOverrides, Source, Task, TaskStatus};
26
27/// What to file. Everything `magi run` / `magi review` take that changes what
28/// the run does rides on the task, so a loop that executes it behaves the same
29/// as this process would have.
30#[derive(Debug, Clone)]
31pub struct Filing {
32    /// The task text; for a review, a line saying which branch.
33    pub instruction: String,
34    /// The task's title.
35    pub title: String,
36    /// Absolute repository path.
37    pub repo: PathBuf,
38    /// Who asked: the operator, or the agent seat named by `MAGI_RUN` / `MAGI_NODE`.
39    pub source: Source,
40    /// `--solo`.
41    pub solo: bool,
42    /// Command-line choices carried onto the task.
43    pub overrides: RunOverrides,
44    /// `magi review <branch>`: the branch to review.
45    pub review_of: Option<String>,
46    /// Explicit source inputs, pinned when the CLI files the task.
47    pub sources: crate::refs::Sources,
48}
49
50impl Filing {
51    fn into_task(self) -> Task {
52        let mut task = Task::new(self.title, self.instruction, self.repo, self.source);
53        task.sources = self.sources;
54        task.solo = self.solo;
55        task.review_of = self.review_of;
56        let mut overrides = self.overrides;
57        overrides.inherit_daemon_options = false;
58        task.overrides = Some(overrides);
59        task
60    }
61}
62
63/// Who will execute a filed task.
64#[derive(Debug)]
65pub enum Filed {
66    /// A loop in another process owns the queue; the task is `urgent` and
67    /// unclaimed. `pid` is the loop's, when it published one.
68    Daemon {
69        /// The filed task.
70        task: Task,
71        /// The loop's pid.
72        pid: Option<u32>,
73    },
74    /// Nobody else is serving: the task is claimed for this process, and the
75    /// claim lives as long as this value.
76    Standalone {
77        /// The filed task.
78        task: Task,
79        /// Held until dropped, which is what keeps a later loop away.
80        claim: Claim,
81    },
82}
83
84/// File `filing` and decide who runs it. `home` is the magi home whose
85/// `daemon.json` is judged; `own_pid` is this process (a loop of our own is
86/// not "another" one). `force_local` (`--dry-run`) never hands over: a dry run
87/// spends no agent call and a loop would not honour that.
88pub fn file(
89    queue: &Queue,
90    home: &Path,
91    now: Timestamp,
92    own_pid: u32,
93    filing: Filing,
94    force_local: bool,
95) -> Result<Filed> {
96    let mut task = filing.into_task();
97    let owner = if force_local {
98        None
99    } else {
100        crate::daemon::foreign_loop(crate::daemon::read_status(home).as_ref(), now, own_pid)
101    };
102    match owner {
103        Some(pid) => {
104            task.urgent = true;
105            queue.put(&mut task).context("file the task")?;
106            Ok(Filed::Daemon { task, pid })
107        }
108        None => {
109            // Claimed as the task is filed, on the id that was reserved:
110            // between `put` and the claim a loop could take it.
111            let claim = queue.file_claimed(&mut task).context("file the task")?;
112            Ok(Filed::Standalone { task, claim })
113        }
114    }
115}
116
117/// The config a task's run is built with: the task's `--config` layer, then
118/// its overrides and `solo`. What the loop's `attempt` does, for the dry run.
119pub fn config_for(task: &Task, repo: &Path) -> Result<Config> {
120    let path = task.overrides.as_ref().and_then(|o| o.config.as_deref());
121    let (mut cfg, _) = Config::discover(repo, path)?;
122    if let Some(o) = &task.overrides {
123        o.apply(&mut cfg);
124    }
125    if task.solo {
126        cfg.graph.implementers = 1;
127    }
128    Ok(cfg)
129}
130
131/// How a followed task ended up.
132#[derive(Debug)]
133pub struct Followed {
134    /// The task as last read.
135    pub task: Task,
136    /// False when `max_wait` ran out first: the loop still owns the task and
137    /// nothing is known of its outcome yet.
138    pub finished: bool,
139}
140
141impl Followed {
142    /// The error a caller must exit with when the wait ran out before the
143    /// task finished: that is not a success, and the outcome is not known.
144    /// `None` once the task finished (whatever its outcome).
145    pub fn unfinished(&self) -> Option<anyhow::Error> {
146        if self.finished {
147            return None;
148        }
149        Some(anyhow::anyhow!(
150            "task {id} has not finished: it is still running under the loop. \
151             This is not a success; the outcome is not known yet. Read it with \
152             `magi task show {id}`. The loop keeps the task, so do not run it again.",
153            id = self.task.short()
154        ))
155    }
156}
157
158/// Watch `id` until the loop that owns it is done with it, calling `seen` with
159/// each run id the first time it appears on the task, and `progress` with a
160/// line whenever the newest run's status changes.
161///
162/// Finished is `done`, `held` or `blocked`: a `failed` or `queued` task is the
163/// loop's to retry. A task that stays unfinished while neither a scheduler nor
164/// its independently leased task worker is alive is an error, never taken over
165/// here. Scheduler replacement does not end a worker's ownership or this wait.
166/// Recovery remains the scheduler's own claim reclaim and `Runner::resume`.
167/// `max_wait` is not an error: it returns with `finished: false`.
168pub async fn follow(
169    queue: &Queue,
170    home: &Path,
171    id: &str,
172    poll: Duration,
173    max_wait: Option<Duration>,
174    mut seen: impl FnMut(&str),
175    mut progress: impl FnMut(&str),
176) -> Result<Followed> {
177    let began = Instant::now();
178    let mut announced = 0usize;
179    let mut last_status = String::new();
180    loop {
181        let task = queue.get(id)?;
182        for run in task.runs.iter().skip(announced) {
183            seen(run);
184        }
185        announced = task.runs.len();
186        if let Some(run) = task.runs.last()
187            && crate::run::try_home().is_some()
188            && let Ok(state) = crate::run::RunState::load(run)
189        {
190            let line = format!("run {}: {}", state.short(), state.status.as_str());
191            if line != last_status {
192                progress(&line);
193                last_status = line;
194            }
195        }
196        if matches!(
197            task.status,
198            TaskStatus::Done | TaskStatus::Held | TaskStatus::Blocked
199        ) {
200            return Ok(Followed {
201                task,
202                finished: true,
203            });
204        }
205        let reading = crate::daemon::read_status(home);
206        if crate::daemon::foreign_loop(reading.as_ref(), Timestamp::now(), std::process::id())
207            .is_none()
208            && !crate::daemon::is_working_on_task(home, id, Timestamp::now())
209        {
210            bail!(
211                "no magi loop is serving the queue any more, and task {} is still {}; it was \
212                 not taken over here (the loop may only be slow to heartbeat). Start `magi \
213                 serve` to carry on, or follow it with `magi task show {}`",
214                task.short(),
215                task.status.as_str(),
216                task.short()
217            );
218        }
219        if max_wait.is_some_and(|m| began.elapsed() >= m) {
220            return Ok(Followed {
221                task,
222                finished: false,
223            });
224        }
225        tokio::time::sleep(poll).await;
226    }
227}
228
229/// Read-only lookup of the task already owning a run.
230#[must_use]
231pub fn resume_owner(queue: &Queue, run: &str) -> Option<Task> {
232    queue
233        .list()
234        .into_iter()
235        .find(|task| task.runs.iter().any(|id| id == run))
236}
237
238/// Find and claim an existing owner, or file a new owner while already claimed.
239/// The caller serializes find-or-create with the run's `resume-<id>` claim and
240/// retains this returned claim until its pinned request is handed to a worker.
241pub fn claim_resume_owner(
242    queue: &Queue,
243    run: &str,
244    instruction: &str,
245    repo: &Path,
246    source: Source,
247    sources: &crate::refs::Sources,
248) -> Result<(Task, Claim)> {
249    if let Some(listed) = resume_owner(queue, run) {
250        let claim = queue.claim(&listed.id)?;
251        let task = queue.get(&listed.id)?;
252        anyhow::ensure!(
253            task.runs.iter().any(|id| id == run),
254            "run queue owner changed"
255        );
256        return Ok((task, claim));
257    }
258    let mut task = Task::new(
259        crate::queue::title_from(instruction, 72),
260        instruction.to_owned(),
261        repo.to_path_buf(),
262        source,
263    );
264    task.sources = sources.clone();
265    task.link_run(run);
266    let claim = queue.file_claimed(&mut task)?;
267    Ok((task, claim))
268}
269
270/// A run opened inside this process (the follow-up review of `magi fix`) and
271/// the task that owns it. See [`adopt`].
272#[derive(Debug)]
273pub struct Adopted {
274    queue: Queue,
275    /// The task this process claimed and started for the run; `None` when the
276    /// claim could not be taken and the run was only linked.
277    task: Option<Task>,
278    claim: Option<Claim>,
279    quota_before: Vec<crate::run::QuotaLoss>,
280}
281
282/// Give a run this process is about to execute an owning task. A parent task
283/// that exists and is queued, failed or held is claimed, started with the
284/// run and settled by [`Adopted::finish`] exactly like an ownerless run's; if
285/// somebody else holds its claim, or the task is Done, blocked or running, the run is only linked and nothing is settled. With no such
286/// parent, a task is filed and claimed for the duration. `Ok(None)` when no
287/// magi home is pinned, or when the run was only linked to its parent. An
288/// error means the run has no owner and must not be executed: filing or
289/// claiming the new task failed.
290pub fn adopt(
291    state: &crate::run::RunState,
292    parent_task: Option<&str>,
293) -> anyhow::Result<Option<Adopted>> {
294    if crate::run::try_home().is_none() {
295        return Ok(None);
296    }
297    adopt_in(Queue::open(), state, parent_task)
298}
299
300fn adopt_in(
301    queue: Queue,
302    state: &crate::run::RunState,
303    parent_task: Option<&str>,
304) -> anyhow::Result<Option<Adopted>> {
305    use anyhow::Context as _;
306    if let Some(parent) = parent_task
307        && let Ok(id) = queue.resolve_id(parent)
308    {
309        // The parent exists: never file a second owner for this run.
310        return Ok(match queue.claim(&id) {
311            Ok(claim) => {
312                let started = queue.get(&id).and_then(|mut task| {
313                    // A task the daemon could pick up, or one a failed
314                    // hand-started run left held (that is what a requested
315                    // follow-up review re-verifies), is this run's to settle.
316                    // A Done, blocked or running one is settled already or
317                    // owned by somebody else: restarting it would let a
318                    // follow-up review overwrite its outcome, so it only
319                    // gains the run.
320                    if !(task.status.runnable() || task.status == TaskStatus::Held) {
321                        return Ok(None);
322                    }
323                    task.start(state.id.clone());
324                    queue.put(&mut task)?;
325                    Ok(Some(task))
326                });
327                match started {
328                    Ok(None) => {
329                        drop(claim);
330                        let _ = queue.link_run(&id, &state.id);
331                        None
332                    }
333                    Ok(Some(task)) => Some(Adopted {
334                        queue,
335                        task: Some(task),
336                        claim: Some(claim),
337                        quota_before: state.quota.clone(),
338                    }),
339                    Err(e) => {
340                        tracing::warn!("could not start task {id} for run {}: {e:#}", state.id);
341                        drop(claim);
342                        let _ = queue.link_run(&id, &state.id);
343                        None
344                    }
345                }
346            }
347            Err(e) => {
348                // Another driver owns the task; do not settle over its result.
349                tracing::warn!("task {id} is claimed elsewhere ({e:#}); linking run only");
350                let _ = queue.link_run(&id, &state.id);
351                None
352            }
353        });
354    }
355    let mut task = Task::new(
356        crate::queue::title_from(&state.instruction, 72),
357        state.instruction.clone(),
358        state.repo.clone(),
359        Source::Human,
360    );
361    task.start(state.id.clone());
362    let claim = queue
363        .file_claimed(&mut task)
364        .with_context(|| format!("could not file a new task for run {}", state.id))?;
365    Ok(Some(Adopted {
366        queue,
367        task: Some(task),
368        claim: Some(claim),
369        quota_before: state.quota.clone(),
370    }))
371}
372
373impl Adopted {
374    /// Settle the task of an adopted ownerless run after it executed.
375    pub fn finish(mut self, state: &crate::run::RunState, result: Result<()>) {
376        if let Some(task) = &mut self.task {
377            crate::daemon::finish_attempt(
378                crate::daemon::Opts::default().max_attempts,
379                &self.queue,
380                task,
381                state,
382                &self.quota_before,
383                result,
384            );
385            crate::daemon::hold_if_runnable(&self.queue, task);
386        }
387        drop(self.claim.take());
388    }
389}
390
391#[cfg(test)]
392mod tests {
393    use super::*;
394    use crate::daemon::Status;
395
396    fn filing(repo: &Path) -> Filing {
397        Filing {
398            instruction: "review the branch".to_owned(),
399            title: "review x".to_owned(),
400            repo: repo.to_path_buf(),
401            source: Source::Human,
402            solo: false,
403            overrides: RunOverrides {
404                merge: Some("none".to_owned()),
405                ..RunOverrides::default()
406            },
407            sources: Default::default(),
408            review_of: Some("feat/x".to_owned()),
409        }
410    }
411
412    fn heartbeat(home: &Path, pid: u32, age_secs: i64) {
413        let mut status = Status::new();
414        status.pid = pid;
415        status.updated_at = Timestamp::now()
416            .checked_sub(jiff::SignedDuration::from_secs(age_secs))
417            .unwrap();
418        crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
419    }
420
421    #[test]
422    fn explicit_sources_survive_filing_queue_storage_and_release() {
423        let dir = tempfile::tempdir().unwrap();
424        let queue = Queue::at(dir.path().join("queue"));
425        let mut direct = filing(dir.path());
426        direct.review_of = None;
427        direct.instruction = "Do not import the historical example abc1234".into();
428        direct.sources = crate::refs::Sources {
429            start_from: Some(crate::refs::Input {
430                token: "feature/selected".into(),
431                sha: "1".repeat(40),
432                provenance: crate::refs::Provenance::ExplicitCli,
433            }),
434            pick_commits: vec![crate::refs::Input {
435                token: "abc5678".into(),
436                sha: "2".repeat(40),
437                provenance: crate::refs::Provenance::ExplicitCli,
438            }],
439        };
440        let selected = direct.sources.clone();
441        let mut task = direct.into_task();
442        queue.put(&mut task).unwrap();
443        let mut restored = queue.get(&task.id).unwrap();
444        assert_eq!(restored.sources, selected);
445        restored.hold_manual(Some("retry preparation".into()));
446        restored.release_operator("test", None);
447        queue.put(&mut restored).unwrap();
448        assert_eq!(queue.get(&task.id).unwrap().sources, selected);
449        assert_eq!(restored.instruction, task.instruction);
450    }
451
452    #[test]
453    fn task_implementers_direct_filing_keeps_options_isolated() {
454        let dir = tempfile::tempdir().unwrap();
455        let mut direct = filing(dir.path());
456        direct.overrides.inherit_daemon_options = true;
457        let task = direct.into_task();
458        assert!(!task.overrides.unwrap().inherit_daemon_options);
459    }
460
461    #[test]
462    fn without_a_loop_the_task_is_claimed_before_anyone_else_can_take_it() {
463        let dir = tempfile::tempdir().unwrap();
464        let q = Queue::at(dir.path().join("queue"));
465        let filed = file(
466            &q,
467            dir.path(),
468            Timestamp::now(),
469            1,
470            filing(dir.path()),
471            false,
472        )
473        .unwrap();
474        let Filed::Standalone { task, claim } = filed else {
475            panic!("no loop is alive");
476        };
477        assert!(!task.urgent);
478        assert_eq!(task.review_of.as_deref(), Some("feat/x"));
479        assert_eq!(
480            task.overrides.as_ref().unwrap().merge.as_deref(),
481            Some("none")
482        );
483        assert!(
484            q.claim(&task.id).is_err(),
485            "a second process must not be able to claim a standalone run's task"
486        );
487        drop(claim);
488        assert!(q.claim(&task.id).is_ok(), "released with the claim");
489    }
490
491    #[test]
492    fn a_live_foreign_loop_gets_an_urgent_unclaimed_task() {
493        let dir = tempfile::tempdir().unwrap();
494        let q = Queue::at(dir.path().join("queue"));
495        heartbeat(dir.path(), 4242, 0);
496        let filed = file(
497            &q,
498            dir.path(),
499            Timestamp::now(),
500            1,
501            filing(dir.path()),
502            false,
503        )
504        .unwrap();
505        let Filed::Daemon { task, pid } = filed else {
506            panic!("a loop is alive");
507        };
508        assert_eq!(pid, Some(4242));
509        assert!(task.urgent);
510        let stored = q.get(&task.id).unwrap();
511        assert_eq!(stored.status, TaskStatus::Queued);
512        assert!(stored.runs.is_empty(), "nothing ran in this process");
513        assert!(q.claim(&task.id).is_ok(), "and this process holds no claim");
514    }
515
516    #[test]
517    fn a_stale_heartbeat_our_own_pid_and_a_dry_run_do_not_count_as_a_loop() {
518        let dir = tempfile::tempdir().unwrap();
519        let q = Queue::at(dir.path().join("queue"));
520        heartbeat(dir.path(), 4242, 3600);
521        assert!(matches!(
522            file(
523                &q,
524                dir.path(),
525                Timestamp::now(),
526                1,
527                filing(dir.path()),
528                false
529            )
530            .unwrap(),
531            Filed::Standalone { .. }
532        ));
533        heartbeat(dir.path(), 7, 0);
534        assert!(matches!(
535            file(
536                &q,
537                dir.path(),
538                Timestamp::now(),
539                7,
540                filing(dir.path()),
541                false
542            )
543            .unwrap(),
544            Filed::Standalone { .. }
545        ));
546        heartbeat(dir.path(), 4242, 0);
547        assert!(matches!(
548            file(
549                &q,
550                dir.path(),
551                Timestamp::now(),
552                1,
553                filing(dir.path()),
554                true
555            )
556            .unwrap(),
557            Filed::Standalone { .. }
558        ));
559    }
560
561    #[tokio::test]
562    async fn following_gives_up_on_a_task_nobody_is_serving() {
563        let dir = tempfile::tempdir().unwrap();
564        let q = Queue::at(dir.path().join("queue"));
565        let mut t = filing(dir.path()).into_task();
566        q.put(&mut t).unwrap();
567        let err = follow(
568            &q,
569            dir.path(),
570            &t.id,
571            Duration::from_millis(5),
572            None,
573            |_| {},
574            |_| {},
575        )
576        .await
577        .unwrap_err();
578        assert!(format!("{err}").contains("no magi loop"), "{err}");
579        assert!(q.claim(&t.id).is_ok(), "never taken over");
580    }
581
582    #[tokio::test]
583    async fn following_returns_when_the_loop_finishes_the_task_and_times_out_otherwise() {
584        let dir = tempfile::tempdir().unwrap();
585        let q = Queue::at(dir.path().join("queue"));
586        heartbeat(dir.path(), 4242, 0);
587        let mut t = filing(dir.path()).into_task();
588        q.put(&mut t).unwrap();
589        let waited = follow(
590            &q,
591            dir.path(),
592            &t.id,
593            Duration::from_millis(5),
594            Some(Duration::from_millis(30)),
595            |_| {},
596            |_| {},
597        )
598        .await
599        .unwrap();
600        assert!(!waited.finished, "the wait ran out, the loop still owns it");
601        let err = waited
602            .unfinished()
603            .expect("an unfinished wait is an error")
604            .to_string();
605        assert!(err.contains(t.short()), "{err}");
606        assert!(
607            err.contains("not finished") || err.contains("not a success"),
608            "{err}"
609        );
610        assert!(err.contains("magi task show"), "{err}");
611        assert_eq!(
612            q.get(&t.id).unwrap().status,
613            TaskStatus::Queued,
614            "the queue is untouched"
615        );
616        t.link_run("20260101-000000-abcd");
617        t.succeed();
618        q.put(&mut t).unwrap();
619        let mut runs = Vec::new();
620        let done = follow(
621            &q,
622            dir.path(),
623            &t.id,
624            Duration::from_millis(5),
625            Some(Duration::from_secs(5)),
626            |r| runs.push(r.to_owned()),
627            |_| {},
628        )
629        .await
630        .unwrap();
631        assert_eq!(done.task.status, TaskStatus::Done);
632        assert!(done.unfinished().is_none());
633        assert_eq!(runs, ["20260101-000000-abcd"]);
634    }
635
636    fn parent_in(q: &Queue, dir: &Path) -> Task {
637        let mut t = filing(dir).into_task();
638        q.put(&mut t).unwrap();
639        t
640    }
641
642    fn follow_up_state(dir: &Path) -> crate::run::RunState {
643        crate::run::RunState::new(
644            dir.to_path_buf(),
645            "main".to_owned(),
646            "abc1234".to_owned(),
647            "review the branch".to_owned(),
648            Config::default(),
649        )
650    }
651
652    #[test]
653    fn an_adopted_review_claims_starts_and_settles_a_runnable_task() {
654        let dir = tempfile::tempdir().unwrap();
655        let q = Queue::at(dir.path().join("queue"));
656        let parent = parent_in(&q, dir.path());
657        let state = follow_up_state(dir.path());
658        let adopted = adopt_in(q.clone(), &state, Some(&parent.id))
659            .unwrap()
660            .expect("adopted");
661        let running = q.get(&parent.id).unwrap();
662        assert_eq!(running.status, TaskStatus::Running);
663        assert_eq!(running.attempts, parent.attempts + 1);
664        assert_eq!(running.runs, std::slice::from_ref(&state.id));
665        assert!(q.claim(&parent.id).is_err(), "exclusive while it runs");
666        adopted.finish(&state, Err(anyhow::anyhow!("boom")));
667        let settled = q.get(&parent.id).unwrap();
668        assert_eq!(settled.status, TaskStatus::Held, "{settled:?}");
669        assert_eq!(
670            settled.runs,
671            std::slice::from_ref(&state.id),
672            "no duplicate run"
673        );
674        assert!(q.claim(&parent.id).is_ok(), "claim released");
675        assert_eq!(q.list().len(), 1, "no second owner was filed");
676    }
677
678    #[test]
679    fn an_ownerless_run_whose_task_cannot_be_filed_is_an_error() {
680        let dir = tempfile::tempdir().unwrap();
681        // The queue root sits under a regular file, so claiming cannot create it.
682        let blocker = dir.path().join("blocker");
683        std::fs::write(&blocker, "x").unwrap();
684        let q = Queue::at(blocker.join("q"));
685        let state = follow_up_state(dir.path());
686        let err = adopt_in(q, &state, None).expect_err("must not be swallowed");
687        assert!(format!("{err:#}").contains(&state.id), "{err:#}");
688    }
689
690    #[test]
691    fn a_task_claimed_elsewhere_only_gains_the_run() {
692        let dir = tempfile::tempdir().unwrap();
693        let q = Queue::at(dir.path().join("queue"));
694        let parent = parent_in(&q, dir.path());
695        let _theirs = q.claim(&parent.id).unwrap();
696        let state = follow_up_state(dir.path());
697        assert!(
698            adopt_in(q.clone(), &state, Some(&parent.id))
699                .unwrap()
700                .is_none()
701        );
702        let after = q.get(&parent.id).unwrap();
703        assert_eq!(after.status, parent.status);
704        assert_eq!(after.attempts, parent.attempts);
705        assert_eq!(after.runs, std::slice::from_ref(&state.id));
706        assert_eq!(q.list().len(), 1, "no second owner was filed");
707    }
708
709    #[test]
710    fn a_done_task_only_gains_the_run() {
711        let dir = tempfile::tempdir().unwrap();
712        let q = Queue::at(dir.path().join("queue"));
713        let mut parent = parent_in(&q, dir.path());
714        parent.succeed();
715        q.put(&mut parent).unwrap();
716        let state = follow_up_state(dir.path());
717        assert!(
718            adopt_in(q.clone(), &state, Some(&parent.id))
719                .unwrap()
720                .is_none()
721        );
722        let after = q.get(&parent.id).unwrap();
723        assert_eq!(after.status, TaskStatus::Done);
724        assert_eq!(after.attempts, parent.attempts);
725        assert_eq!(after.runs, std::slice::from_ref(&state.id));
726        assert!(q.claim(&parent.id).is_ok(), "claim released");
727    }
728
729    #[test]
730    fn a_held_task_is_claimed_and_settled_by_its_follow_up_review() {
731        let dir = tempfile::tempdir().unwrap();
732        let q = Queue::at(dir.path().join("queue"));
733        let mut parent = parent_in(&q, dir.path());
734        parent.hold_manual(Some("the run did not finish: stale".to_owned()));
735        q.put(&mut parent).unwrap();
736        let state = follow_up_state(dir.path());
737        let adopted = adopt_in(q.clone(), &state, Some(&parent.id))
738            .unwrap()
739            .expect("adopted");
740        assert_eq!(q.get(&parent.id).unwrap().status, TaskStatus::Running);
741        assert!(q.claim(&parent.id).is_err(), "claimed while it runs");
742        adopted.finish(&state, Err(anyhow::anyhow!("fresh failure")));
743        let after = q.get(&parent.id).unwrap();
744        assert_eq!(after.status, TaskStatus::Held);
745        assert!(
746            after
747                .hold_reason
748                .as_deref()
749                .unwrap()
750                .contains("fresh failure"),
751            "{after:?}"
752        );
753        assert_eq!(after.runs, std::slice::from_ref(&state.id));
754        assert!(q.claim(&parent.id).is_ok(), "claim released");
755    }
756}