1use 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#[derive(Debug, Clone)]
31pub struct Filing {
32 pub instruction: String,
34 pub title: String,
36 pub repo: PathBuf,
38 pub source: Source,
40 pub solo: bool,
42 pub overrides: RunOverrides,
44 pub review_of: Option<String>,
46 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#[derive(Debug)]
65pub enum Filed {
66 Daemon {
69 task: Task,
71 pid: Option<u32>,
73 },
74 Standalone {
77 task: Task,
79 claim: Claim,
81 },
82}
83
84pub 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 let claim = queue.file_claimed(&mut task).context("file the task")?;
112 Ok(Filed::Standalone { task, claim })
113 }
114 }
115}
116
117pub 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#[derive(Debug)]
133pub struct Followed {
134 pub task: Task,
136 pub finished: bool,
139}
140
141impl Followed {
142 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
158pub 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#[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
238pub 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#[derive(Debug)]
273pub struct Adopted {
274 queue: Queue,
275 task: Option<Task>,
278 claim: Option<Claim>,
279 quota_before: Vec<crate::run::QuotaLoss>,
280}
281
282pub 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 return Ok(match queue.claim(&id) {
311 Ok(claim) => {
312 let started = queue.get(&id).and_then(|mut task| {
313 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 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 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 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}