1use std::collections::HashMap;
34use std::path::{Path, PathBuf};
35
36use anyhow::{Context, Result, bail};
37use jiff::Timestamp;
38use serde::{Deserialize, Serialize};
39
40use crate::ask::Questions;
41
42pub const SCHEMA: u32 = 13;
115
116#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
118#[serde(rename_all = "lowercase")]
119pub enum HoldSource {
120 Manual,
122 Machine,
124}
125
126impl HoldSource {
127 pub fn label(self) -> &'static str {
129 match self {
130 Self::Manual => "manual",
131 Self::Machine => "machine",
132 }
133 }
134}
135
136#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
139#[serde(tag = "kind", rename_all = "lowercase")]
140pub enum Source {
141 Human,
143 Agent {
146 run: String,
148 node: String,
150 },
151 Issue {
153 number: u64,
155 repo: String,
157 },
158}
159
160pub const CHAT_NODE: &str = "chat";
163
164impl Source {
165 pub fn label(&self) -> String {
167 match self {
168 Self::Human => "human".to_owned(),
169 Self::Agent { run, node } => format!("{node}@{}", short(run)),
170 Self::Issue { number, .. } => format!("issue #{number}"),
171 }
172 }
173}
174
175#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
177#[serde(rename_all = "lowercase")]
178pub enum TaskStatus {
179 Queued,
181 Running,
183 Done,
185 Failed,
187 Held,
189 Blocked,
193 Parked,
198}
199
200impl TaskStatus {
201 pub fn runnable(self) -> bool {
203 matches!(self, Self::Queued | Self::Failed | Self::Parked)
204 }
205
206 pub fn as_str(self) -> &'static str {
208 match self {
209 Self::Queued => "queued",
210 Self::Running => "running",
211 Self::Done => "done",
212 Self::Failed => "failed",
213 Self::Held => "held",
214 Self::Blocked => "blocked",
215 Self::Parked => "parked",
216 }
217 }
218}
219
220#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
223pub struct TaskCounts {
224 pub queued: usize,
226 pub running: usize,
228 pub done: usize,
230 pub failed: usize,
232 pub held: usize,
234 pub blocked: usize,
236 pub parked: usize,
238}
239
240impl TaskCounts {
241 pub fn of(tasks: &[Task]) -> Self {
245 let mut counts = Self::default();
246 for t in tasks {
247 match t.status {
248 TaskStatus::Queued => counts.queued += 1,
249 TaskStatus::Running => counts.running += 1,
250 TaskStatus::Done => counts.done += 1,
251 TaskStatus::Failed => counts.failed += 1,
252 TaskStatus::Held => counts.held += 1,
253 TaskStatus::Blocked => counts.blocked += 1,
254 TaskStatus::Parked => counts.parked += 1,
255 }
256 }
257 counts
258 }
259}
260
261#[derive(Debug, Clone, Serialize, Deserialize)]
263#[serde(deny_unknown_fields)]
264pub struct Task {
265 pub schema: u32,
267 pub id: String,
269 pub title: String,
271 pub instruction: String,
273 pub repo: PathBuf,
275 pub source: Source,
277 #[serde(default)]
279 pub priority: i32,
280 #[serde(default)]
291 pub solo: bool,
292 pub status: TaskStatus,
294 #[serde(default)]
296 pub attempts: usize,
297 #[serde(default)]
299 pub runs: Vec<String>,
300 #[serde(default)]
302 pub last_error: Option<String>,
303 #[serde(default)]
316 pub hold_reason: Option<String>,
317 #[serde(default)]
320 pub hold_source: Option<HoldSource>,
321 #[serde(default)]
333 pub diagnostic: Option<String>,
334 #[serde(default)]
344 pub blocked_by: Vec<String>,
345 #[serde(default)]
348 pub block_reason: Option<String>,
349 #[serde(default)]
362 pub blocked_from: Option<TaskStatus>,
363 #[serde(default)]
374 pub answers: Vec<AnsweredQuestion>,
375 #[serde(default)]
381 pub triage_applied: Vec<String>,
382 #[serde(default)]
388 pub actions_applied: Vec<String>,
389 #[serde(default)]
394 pub resume_override: Option<OperatorResume>,
395 #[serde(default)]
407 pub review_branch: Option<String>,
408 #[serde(default)]
411 pub fresh_start: bool,
412 #[serde(default)]
423 pub interrupt: bool,
424 #[serde(default)]
447 pub urgent: bool,
448 #[serde(default)]
454 pub attachments: Vec<String>,
455 #[serde(default)]
459 pub followup: Option<FollowUp>,
460 #[serde(default)]
466 pub origin_chat: Option<String>,
467 #[serde(default)]
473 pub overrides: Option<RunOverrides>,
474 #[serde(default)]
481 pub review_of: Option<String>,
482 #[serde(default)]
487 pub held_at: Option<Timestamp>,
488 #[serde(default)]
492 pub park_reason: Option<String>,
493 #[serde(default)]
496 pub chat_notice: Option<ChatNotice>,
497 pub created_at: Timestamp,
499 pub updated_at: Timestamp,
501}
502
503pub(crate) fn is_done_notice_key(key: &str) -> bool {
505 key == "done"
506 || key
507 .strip_prefix("done#")
508 .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
509}
510
511#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
513#[serde(rename_all = "lowercase")]
514pub enum ChatNoticeOutcome {
515 Posted,
517 Skipped,
519}
520
521#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
523pub struct ChatNotice {
524 pub key: String,
526 pub at: Timestamp,
528 pub outcome: ChatNoticeOutcome,
530}
531
532#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
534pub struct RunOverrides {
535 #[serde(default)]
537 pub merge: Option<String>,
538 #[serde(default)]
540 pub candidates: Option<usize>,
541 #[serde(default)]
543 pub judges: Option<usize>,
544 #[serde(default)]
546 pub reviewers: Option<usize>,
547 #[serde(default)]
549 pub review_rounds: Option<usize>,
550 #[serde(default)]
552 pub seed: Option<u64>,
553 #[serde(default)]
555 pub config: Option<PathBuf>,
556}
557
558impl RunOverrides {
559 pub fn merge_over(&mut self, other: &RunOverrides) {
561 macro_rules! take {
562 ($($f:ident),*) => {$(
563 if other.$f.is_some() {
564 self.$f.clone_from(&other.$f);
565 }
566 )*};
567 }
568 take!(
569 merge,
570 candidates,
571 judges,
572 reviewers,
573 review_rounds,
574 seed,
575 config
576 );
577 }
578
579 pub fn apply(&self, config: &mut crate::config::Config) {
581 if let Some(n) = self.candidates {
582 config.graph.implementers = n;
583 }
584 if let Some(n) = self.judges {
585 config.graph.judges = n;
586 }
587 if let Some(n) = self.reviewers {
588 config.graph.reviewers = n;
589 }
590 if let Some(n) = self.review_rounds {
591 config.graph.review_rounds = n;
592 }
593 if let Some(m) = &self.merge
594 && let Ok(mode) = crate::daemon::merge_mode(m)
595 {
596 config.merge.mode = mode;
597 }
598 if let Some(s) = self.seed {
599 config.blind.seed = Some(s);
600 }
601 }
602}
603
604#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
606pub struct FollowUp {
607 pub run: String,
609 #[serde(default)]
611 pub origin_task: Option<String>,
612 pub pr: String,
614 pub findings: Vec<String>,
616 pub generation: u32,
619}
620
621#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
624pub struct OperatorResume {
625 pub question_id: String,
627 pub at: Timestamp,
629 #[serde(default)]
632 pub conductor_rehold: Option<String>,
633 #[serde(default)]
636 pub forced: bool,
637 #[serde(default)]
644 pub pinned_run: Option<String>,
645}
646
647#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
650pub struct AnsweredQuestion {
651 pub question: String,
653 pub answer: String,
655}
656
657impl Task {
658 pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
660 let now = Timestamp::now();
661 let origin_chat = match &source {
662 Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
663 _ => None,
664 };
665 Self {
666 schema: SCHEMA,
667 id: new_id(),
668 title,
669 instruction,
670 repo,
671 source,
672 priority: 0,
673 solo: false,
674 status: TaskStatus::Queued,
675 attempts: 0,
676 runs: Vec::new(),
677 last_error: None,
678 hold_reason: None,
679 hold_source: None,
680 diagnostic: None,
681 blocked_by: Vec::new(),
682 block_reason: None,
683 blocked_from: None,
684 answers: Vec::new(),
685 triage_applied: Vec::new(),
686 actions_applied: Vec::new(),
687 resume_override: None,
688 review_branch: None,
689 fresh_start: false,
690 interrupt: false,
691 urgent: false,
692 attachments: Vec::new(),
693 followup: None,
694 origin_chat,
695 overrides: None,
696 review_of: None,
697 held_at: None,
698 park_reason: None,
699 chat_notice: None,
700 created_at: now,
701 updated_at: now,
702 }
703 }
704
705 pub fn chat_talk(&self) -> Option<&str> {
708 if let Some(id) = &self.origin_chat {
709 return Some(id);
710 }
711 match &self.source {
712 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
713 _ => None,
714 }
715 }
716
717 pub fn filed_by_chat(&self) -> Option<&str> {
721 match &self.source {
722 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
723 _ => None,
724 }
725 }
726
727 pub fn chat_notice_key(&self) -> Option<String> {
733 match self.status {
734 TaskStatus::Done => Some("done".to_owned()),
735 TaskStatus::Held => Some(format!(
736 "held#{}",
737 self.held_at.map(|t| t.to_string()).unwrap_or_default()
738 )),
739 TaskStatus::Blocked => Some(format!(
740 "blocked#{}#{}",
741 self.blocked_by.join(","),
742 self.block_reason.as_deref().unwrap_or_default()
743 )),
744 _ => None,
745 }
746 }
747
748 pub fn chat_notice_due(&self) -> Option<String> {
752 self.filed_by_chat()?;
753 if self.schema < 13 {
754 return None;
755 }
756 let key = self.chat_notice_key()?;
757 match &self.chat_notice {
758 Some(n) if n.key == key || (key == "done" && is_done_notice_key(&n.key)) => None,
759 _ => Some(key),
760 }
761 }
762
763 pub fn short(&self) -> &str {
765 short(&self.id)
766 }
767
768 pub fn mark_triage_applied(&mut self, question_id: &str) {
770 if !self.triage_applied(question_id) {
771 self.triage_applied.push(question_id.to_owned());
772 }
773 }
774
775 pub fn action_applied(&self, question_id: &str) -> bool {
778 self.actions_applied.iter().any(|id| id == question_id)
779 }
780
781 pub fn mark_action_applied(&mut self, question_id: &str) {
783 if !self.action_applied(question_id) {
784 self.actions_applied.push(question_id.to_owned());
785 }
786 }
787
788 pub fn triage_applied(&self, question_id: &str) -> bool {
790 self.triage_applied.iter().any(|id| id == question_id)
791 }
792
793 pub fn start(&mut self, run: String) {
804 self.stamp_schema();
805 self.status = TaskStatus::Running;
806 self.attempts += 1;
807 self.runs.push(run);
808 self.last_error = None;
809 self.park_reason = None;
810 self.fresh_start = false;
811 self.interrupt = false;
812 self.resume_override = None;
814 }
815
816 pub fn link_run(&mut self, run: &str) -> bool {
821 if self.runs.iter().any(|r| r == run) {
822 return false;
823 }
824 self.runs.push(run.to_owned());
825 true
826 }
827
828 pub fn succeed(&mut self) {
838 self.stamp_schema();
839 self.status = TaskStatus::Done;
840 self.resume_override = None;
841 self.last_error = None;
842 self.park_reason = None;
843 self.hold_reason = None;
844 self.hold_source = None;
845 self.diagnostic = None;
846 self.blocked_by.clear();
847 self.block_reason = None;
848 self.blocked_from = None;
849 }
850
851 pub fn already_landed(&mut self, note: impl Into<String>) {
859 self.succeed();
860 self.attempts = self.attempts.saturating_sub(1);
861 self.last_error = Some(note.into());
862 }
863
864 pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
894 if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
895 return &[];
896 }
897 &self.runs[..self.runs.len() - 1]
898 }
899
900 pub fn successor_of(&self, run: &str) -> Option<&String> {
907 let pos = self.runs.iter().position(|r| r == run)?;
908 self.runs.get(pos + 1)
909 }
910
911 pub fn earlier_attempts(&self) -> &[String] {
918 &self.runs
919 }
920
921 fn stamp_schema(&mut self) {
928 self.schema = self.schema.max(SCHEMA);
929 }
930
931 fn note_held(&mut self) {
932 self.stamp_schema();
933 if self.status != TaskStatus::Held || self.held_at.is_none() {
934 self.held_at = Some(Timestamp::now());
935 }
936 }
937
938 pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
954 let why = why.into();
955 self.diagnostic = None;
956 self.park_reason = None;
957 self.status = if self.attempts >= max_attempts {
958 self.note_held();
959 self.hold_source = Some(HoldSource::Machine);
960 self.hold_reason = Some(why.clone());
961 TaskStatus::Held
962 } else {
963 TaskStatus::Failed
964 };
965 self.last_error = Some(why);
966 }
967
968 pub fn stall(&mut self, why: impl Into<String>) {
978 self.last_error = Some(why.into());
979 self.diagnostic = None;
980 self.attempts = self.attempts.saturating_sub(1);
981 self.status = TaskStatus::Failed;
982 }
983
984 pub fn park(&mut self, why: impl Into<String>) {
990 self.diagnostic = None;
991 self.attempts = self.attempts.saturating_sub(1);
992 self.status = TaskStatus::Parked;
993 self.park_reason = Some(why.into());
994 self.schema = self.schema.max(SCHEMA);
997 }
998
999 pub fn operator_held(&self) -> bool {
1005 self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
1006 }
1007
1008 pub fn hold_manual(&mut self, reason: Option<String>) {
1018 self.park_reason = None;
1019 self.note_held();
1020 self.status = TaskStatus::Held;
1021 if reason.is_some() {
1022 self.hold_reason = reason;
1023 }
1024 self.hold_source = Some(HoldSource::Manual);
1025 self.blocked_by.clear();
1026 self.block_reason = None;
1027 self.blocked_from = None;
1028 }
1029
1030 pub fn hold_machine(&mut self, reason: Option<String>) {
1035 self.park_reason = None;
1036 self.note_held();
1037 self.status = TaskStatus::Held;
1038 if reason.is_some() {
1039 self.hold_reason = reason;
1040 }
1041 self.hold_source = Some(HoldSource::Machine);
1042 self.blocked_by.clear();
1043 self.block_reason = None;
1044 self.blocked_from = None;
1045 }
1046
1047 pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
1057 self.stamp_schema();
1058 if self.status != TaskStatus::Blocked {
1059 self.blocked_from = Some(self.status);
1060 }
1061 self.status = TaskStatus::Blocked;
1062 self.blocked_by = blocked_by;
1063 self.block_reason = reason;
1064 }
1065
1066 pub fn unblock(&mut self, resolved_id: &str) {
1089 if self.status != TaskStatus::Blocked {
1090 return;
1091 }
1092 self.blocked_by.retain(|id| id != resolved_id);
1093 self.restore_if_unblocked();
1094 }
1095
1096 pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
1107 if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
1108 return false;
1109 }
1110 self.blocked_by.retain(|id| id != deleted_id);
1111 self.restore_if_unblocked();
1112 true
1113 }
1114
1115 fn restore_if_unblocked(&mut self) {
1119 if self.blocked_by.is_empty() {
1120 self.status = match self.blocked_from {
1121 Some(TaskStatus::Running) => TaskStatus::Queued,
1122 Some(other) => other,
1123 None if self.hold_reason.is_some() || self.hold_source.is_some() => {
1124 TaskStatus::Held
1125 }
1126 None => TaskStatus::Queued,
1127 };
1128 self.block_reason = None;
1129 self.blocked_from = None;
1130 }
1131 }
1132
1133 pub fn record_answer(&mut self, question: String, answer: String) {
1138 self.answers.push(AnsweredQuestion { question, answer });
1139 }
1140
1141 pub fn request_review(&mut self, branch: String) {
1145 self.release();
1146 self.review_branch = Some(branch);
1147 }
1148
1149 pub fn requeue(&mut self) {
1152 self.release();
1153 self.review_branch = None;
1155 self.fresh_start = true;
1156 }
1157
1158 pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
1167 if branch.is_some() {
1168 self.review_branch = branch;
1169 }
1170 self.hold_machine(Some(reason));
1171 }
1172
1173 pub fn set_priority(&mut self, priority: i32) -> Result<()> {
1182 if self.status == TaskStatus::Running {
1183 bail!(
1184 "task {} is running; its priority cannot be changed until \
1185 this attempt finishes",
1186 self.short()
1187 );
1188 }
1189 self.priority = priority;
1190 Ok(())
1191 }
1192
1193 pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
1208 if interrupt && !self.status.runnable() {
1209 bail!(
1210 "task {} is {}; only a queued or failed task can be marked \
1211 to interrupt",
1212 self.short(),
1213 self.status.as_str()
1214 );
1215 }
1216 self.interrupt = interrupt;
1217 Ok(())
1218 }
1219
1220 pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
1232 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1233 bail!(
1234 "task {} is {}; only a queued or held task's instruction can \
1235 be edited",
1236 self.short(),
1237 self.status.as_str()
1238 );
1239 }
1240 self.title = title;
1241 self.instruction = instruction;
1242 Ok(())
1243 }
1244
1245 pub fn handed_off(&mut self, why: impl Into<String>) {
1265 let why = why.into();
1266 self.diagnostic = None;
1267 self.note_held();
1268 self.status = TaskStatus::Held;
1269 self.hold_source = Some(HoldSource::Machine);
1270 self.hold_reason = Some(why.clone());
1271 self.last_error = Some(why);
1272 }
1273
1274 pub fn release(&mut self) {
1278 let refused_handover = self.status == TaskStatus::Held
1279 && self.hold_source == Some(HoldSource::Machine)
1280 && self.review_branch.is_some();
1281 self.status = TaskStatus::Queued;
1282 self.held_at = None;
1283 self.attempts = 0;
1284 self.last_error = None;
1285 self.park_reason = None;
1286 self.hold_reason = None;
1289 self.hold_source = None;
1290 self.diagnostic = None;
1291 self.blocked_by.clear();
1296 self.block_reason = None;
1297 self.blocked_from = None;
1298 if !refused_handover {
1302 self.review_branch = None;
1303 }
1304 self.fresh_start = false;
1305 }
1306}
1307
1308fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
1312 use std::io::ErrorKind;
1313 let (stem, ext) = match name.rfind('.') {
1314 Some(i) if i > 0 => (&name[..i], &name[i..]),
1315 _ => (name, ""),
1316 };
1317 for n in 1u32.. {
1318 let candidate = if n == 1 {
1319 name.to_owned()
1320 } else {
1321 let suffix = format!("-{n}");
1322 let room = 64usize.saturating_sub(suffix.len() + ext.len());
1323 let stem: String = stem.chars().take(room).collect();
1324 format!("{stem}{suffix}{ext}")
1325 };
1326 if !crate::ask::valid_asset_name(&candidate) {
1327 bail!("no valid attachment name is left for `{name}`");
1328 }
1329 let path = dir.join(&candidate);
1330 match std::fs::OpenOptions::new()
1331 .write(true)
1332 .create_new(true)
1333 .open(&path)
1334 {
1335 Ok(mut out) => {
1336 let copied = std::fs::File::open(src)
1337 .and_then(|mut input| std::io::copy(&mut input, &mut out));
1338 if let Err(e) = copied {
1339 drop(out);
1340 let _ = std::fs::remove_file(&path);
1341 return Err(e).with_context(|| format!("copy {}", src.display()));
1342 }
1343 return Ok((candidate, path));
1344 }
1345 Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
1346 Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
1347 }
1348 }
1349 unreachable!("the counter never runs out")
1350}
1351
1352const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
1355
1356struct TaskLock {
1359 path: PathBuf,
1360 token: String,
1361}
1362
1363fn owner_token() -> String {
1366 static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
1367 let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1368 let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
1369 format!("{}-{:016x}", std::process::id(), r.next_u64())
1370}
1371
1372fn break_marker(path: &Path, token: &str) -> PathBuf {
1375 let mut h: u64 = 0xcbf29ce484222325;
1376 for b in token.bytes() {
1377 h ^= u64::from(b);
1378 h = h.wrapping_mul(0x100000001b3);
1379 }
1380 let mut name = path.as_os_str().to_owned();
1381 name.push(format!(".break-{h:016x}"));
1382 PathBuf::from(name)
1383}
1384
1385fn older_than_stale(path: &Path) -> bool {
1386 std::fs::metadata(path)
1387 .and_then(|m| m.modified())
1388 .ok()
1389 .and_then(|t| t.elapsed().ok())
1390 .is_some_and(|age| age > TASK_LOCK_STALE)
1391}
1392
1393const MAX_MARKER_DEPTH: u8 = 3;
1397
1398fn take_marker(path: &Path, token: &str, depth: u8) -> Option<(PathBuf, String)> {
1404 use std::io::Write;
1405 let marker = break_marker(path, token);
1406 for _ in 0..2 {
1407 match std::fs::OpenOptions::new()
1408 .write(true)
1409 .create_new(true)
1410 .open(&marker)
1411 {
1412 Ok(mut f) => {
1413 let mine = owner_token();
1414 if f.write_all(mine.as_bytes()).is_err() {
1415 drop(f);
1416 let _ = std::fs::remove_file(&marker);
1417 return None;
1418 }
1419 return Some((marker, mine));
1420 }
1421 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
1422 let judged = std::fs::read_to_string(&marker).ok();
1423 match judged.filter(|_| older_than_stale(&marker)) {
1424 Some(judged) if depth < MAX_MARKER_DEPTH => {
1425 remove_lock_if(&marker, &judged, true, depth + 1)?;
1426 }
1427 _ => return None,
1428 }
1429 }
1430 Err(_) => return None,
1431 }
1432 }
1433 None
1434}
1435
1436fn remove_lock_if(path: &Path, token: &str, require_stale: bool, depth: u8) -> Option<bool> {
1441 let (marker, mine) = take_marker(path, token, depth)?;
1442 let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
1443 && (!require_stale || older_than_stale(path));
1444 if still {
1445 let _ = std::fs::remove_file(path);
1446 }
1447 if depth >= MAX_MARKER_DEPTH {
1449 if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
1450 let _ = std::fs::remove_file(&marker);
1451 }
1452 } else {
1453 let _ = remove_lock_if(&marker, &mine, false, depth + 1);
1454 }
1455 Some(still)
1456}
1457
1458fn break_stale(path: &Path, judged: &str) -> bool {
1461 remove_lock_if(path, judged, true, 0).unwrap_or(false)
1462}
1463
1464impl Drop for TaskLock {
1465 fn drop(&mut self) {
1466 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
1467 loop {
1468 match remove_lock_if(&self.path, &self.token, false, 0) {
1469 Some(_) => return,
1470 None if std::time::Instant::now() > deadline => return,
1471 None => std::thread::sleep(std::time::Duration::from_millis(5)),
1472 }
1473 }
1474 }
1475}
1476
1477#[derive(Debug, Clone)]
1479pub struct Queue {
1480 root: PathBuf,
1481}
1482
1483impl Queue {
1484 pub fn open() -> Self {
1486 Self::at(crate::run::home().join("queue"))
1487 }
1488
1489 pub fn at(root: PathBuf) -> Self {
1492 Self { root }
1493 }
1494
1495 pub fn root(&self) -> &Path {
1497 &self.root
1498 }
1499
1500 pub fn path_of(&self, id: &str) -> PathBuf {
1502 self.root.join(format!("{id}.json"))
1503 }
1504
1505 pub fn attachments_dir(&self, id: &str) -> PathBuf {
1508 self.root.join(format!("{id}.attachments"))
1509 }
1510
1511 pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
1522 let mut wanted = Vec::new();
1523 for src in sources {
1524 let name = src
1525 .file_name()
1526 .and_then(|n| n.to_str())
1527 .with_context(|| format!("`{}` has no usable file name", src.display()))?;
1528 if !crate::ask::valid_asset_name(name) {
1529 bail!(
1530 "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
1531 with no `..`; rename the file and try again"
1532 );
1533 }
1534 if !src.is_file() {
1535 bail!("attachment `{}` is not a file", src.display());
1536 }
1537 wanted.push((src, name));
1538 }
1539 let dir = self.attachments_dir(&task.id);
1540 let existed = dir.is_dir();
1541 std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
1542 let mut created: Vec<PathBuf> = Vec::new();
1543 let mut names = Vec::new();
1544 let mut copy_all = || -> Result<()> {
1545 for (src, name) in &wanted {
1546 let (stored, path) = copy_new(&dir, src, name)?;
1547 created.push(path);
1548 names.push(stored);
1549 }
1550 Ok(())
1551 };
1552 if let Err(e) = copy_all() {
1553 for path in &created {
1554 let _ = std::fs::remove_file(path);
1555 }
1556 if !existed {
1557 let _ = std::fs::remove_dir(&dir);
1558 }
1559 return Err(e);
1560 }
1561 task.attachments.extend(names.iter().cloned());
1562 Ok(names)
1563 }
1564
1565 pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
1576 let dir = self.attachments_dir(&task.id);
1577 let existed = dir.is_dir();
1578 let before = task.attachments.len();
1579 let names = self.attach(task, sources)?;
1580 if let Err(e) = self.put(task) {
1581 for name in &names {
1582 let _ = std::fs::remove_file(dir.join(name));
1583 }
1584 if !existed {
1585 let _ = std::fs::remove_dir(&dir);
1586 }
1587 task.attachments.truncate(before);
1588 return Err(e);
1589 }
1590 Ok(names)
1591 }
1592
1593 pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
1597 let dir = self.attachments_dir(&task.id);
1598 task.attachments
1599 .iter()
1600 .map(|n| {
1601 let p = dir.join(n);
1602 std::path::absolute(&p).unwrap_or(p)
1603 })
1604 .collect()
1605 }
1606
1607 pub fn put(&self, task: &mut Task) -> Result<()> {
1616 let _lock = self.lock_task(&task.id)?;
1617 self.put_unlocked(task)
1618 }
1619
1620 pub fn create_new(&self, task: &mut Task) -> Result<bool> {
1625 let _lock = self.lock_task(&task.id)?;
1626 if self.path_of(&task.id).exists() {
1627 return Ok(false);
1628 }
1629 self.put_unlocked(task)?;
1630 Ok(true)
1631 }
1632
1633 fn put_unlocked(&self, task: &mut Task) -> Result<()> {
1635 if let Ok(stored) = read_path(&self.path_of(&task.id)) {
1636 for run in stored.runs {
1637 if !task.runs.contains(&run) {
1638 task.runs.push(run);
1639 }
1640 }
1641 if let Some(n) = stored.chat_notice
1644 && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
1645 {
1646 task.chat_notice = Some(n);
1647 }
1648 }
1649 task.updated_at = Timestamp::now();
1650 std::fs::create_dir_all(&self.root)
1651 .with_context(|| format!("create {}", self.root.display()))?;
1652 let body = serde_json::to_string_pretty(task).context("serialize task")?;
1653 let path = self.path_of(&task.id);
1654 let tmp = path.with_extension("json.tmp");
1655 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
1656 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
1657 if let (Some(notice), Some(home)) = (
1662 crate::notices::task_held(task),
1663 self.root.parent().filter(|p| !p.as_os_str().is_empty()),
1664 ) {
1665 crate::notices::raise_in(home, notice);
1666 }
1667 Ok(())
1668 }
1669
1670 pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
1675 let id = self.resolve_id(id)?;
1676 let _lock = self.lock_task(&id)?;
1677 let mut task = self.get(&id)?;
1678 if !f(&mut task) {
1679 return Ok(false);
1680 }
1681 self.put_unlocked(&mut task)?;
1682 Ok(true)
1683 }
1684
1685 pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
1690 let id = self.resolve_id(id)?;
1691 let _lock = self.lock_task(&id)?;
1694 let mut task = self.get(&id)?;
1695 if task.link_run(run) {
1696 self.put_unlocked(&mut task)?;
1697 }
1698 Ok(task)
1699 }
1700
1701 fn lock_task(&self, id: &str) -> Result<TaskLock> {
1709 std::fs::create_dir_all(&self.root)
1710 .with_context(|| format!("create {}", self.root.display()))?;
1711 let path = self.root.join(format!("{id}.write-lock"));
1712 let started = std::time::Instant::now();
1713 loop {
1714 match std::fs::OpenOptions::new()
1715 .write(true)
1716 .create_new(true)
1717 .open(&path)
1718 {
1719 Ok(mut file) => {
1720 use std::io::Write;
1721 let token = owner_token();
1722 if let Err(e) = file.write_all(token.as_bytes()) {
1723 drop(file);
1724 let _ = std::fs::remove_file(&path);
1725 return Err(e).with_context(|| format!("lock {}", path.display()));
1726 }
1727 return Ok(TaskLock { path, token });
1728 }
1729 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
1730 let judged = std::fs::read_to_string(&path).ok();
1733 let broken = judged
1734 .filter(|_| older_than_stale(&path))
1735 .is_some_and(|judged| break_stale(&path, &judged));
1736 if !broken {
1740 if started.elapsed() > TASK_LOCK_STALE {
1741 bail!("could not lock task {id}");
1742 }
1743 std::thread::sleep(std::time::Duration::from_millis(15));
1744 }
1745 }
1746 Err(e)
1750 if e.kind() == std::io::ErrorKind::PermissionDenied
1751 && started.elapsed() <= TASK_LOCK_STALE =>
1752 {
1753 std::thread::sleep(std::time::Duration::from_millis(15));
1754 }
1755 Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
1756 }
1757 }
1758 }
1759
1760 pub fn get(&self, id: &str) -> Result<Task> {
1762 let resolved = self.resolve_id(id)?;
1763 read_path(&self.path_of(&resolved))
1764 }
1765
1766 pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
1798 let _ = questions;
1799 let resolved = self.resolve_id(id)?;
1800 if in_flight {
1801 bail!("task {resolved} is being run by a live daemon right now");
1802 }
1803 self.write_tombstone(&resolved)?;
1804 if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
1805 {
1806 if self.path_of(&resolved).exists() {
1810 let _ = std::fs::remove_file(self.tombstone_path(&resolved));
1811 }
1812 return Err(e);
1813 }
1814 let (released, still_blocked) = self.release_dependents_of(&resolved);
1815 Ok(Removal {
1816 id: resolved,
1817 released,
1818 still_blocked,
1819 })
1820 }
1821
1822 fn tombstone_path(&self, id: &str) -> PathBuf {
1825 self.root.join(format!("{id}.removed"))
1826 }
1827
1828 fn write_tombstone(&self, id: &str) -> Result<()> {
1829 let path = self.tombstone_path(id);
1830 let tmp = self.root.join(format!("{id}.removed.tmp"));
1831 std::fs::write(&tmp, b"")
1832 .and_then(|()| std::fs::rename(&tmp, &path))
1833 .with_context(|| format!("write {}", path.display()))
1834 }
1835
1836 fn was_deleted(&self, id: &str) -> bool {
1840 self.tombstone_path(id).is_file() && !self.path_of(id).exists()
1841 }
1842
1843 pub fn apply_deleted_blockers(&self, task: &mut Task) -> Vec<String> {
1847 let deleted = deleted_blockers(self, &task.blocked_by);
1848 deleted
1849 .into_iter()
1850 .filter(|id| task.dependency_deleted(id))
1851 .collect()
1852 }
1853
1854 pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
1858 let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
1859 return;
1860 };
1861 let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
1862 let waiting = dependent.status == TaskStatus::Blocked;
1863 let message = match (ja, waiting) {
1864 (true, false) => format!(
1865 "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
1866 dependent.short(),
1867 short(deleted)
1868 ),
1869 (true, true) => format!(
1870 "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
1871 dependent.short(),
1872 short(deleted)
1873 ),
1874 (false, false) => format!(
1875 "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
1876 dependent.short(),
1877 short(deleted)
1878 ),
1879 (false, true) => format!(
1880 "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
1881 dependent.short(),
1882 short(deleted)
1883 ),
1884 };
1885 crate::notices::raise_in(
1886 home,
1887 crate::notices::Notice::info(
1888 &format!("unblocked:{}:{}", dependent.id, deleted),
1889 message,
1890 )
1891 .link(crate::notices::Link::Task {
1892 id: dependent.id.clone(),
1893 }),
1894 );
1895 }
1896
1897 fn release_dependents_of(&self, dependency: &str) -> (Vec<String>, Vec<String>) {
1902 let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
1903 for listed in self.list() {
1904 if listed.status != TaskStatus::Blocked
1905 || !listed.blocked_by.iter().any(|b| b == dependency)
1906 {
1907 continue;
1908 }
1909 let Ok(_claim) = self.claim(&listed.id) else {
1910 continue;
1911 };
1912 let Ok(mut task) = self.get(&listed.id) else {
1913 continue;
1914 };
1915 if !task.dependency_deleted(dependency) {
1916 continue;
1917 }
1918 if self.put(&mut task).is_err() {
1919 continue;
1920 }
1921 self.note_dependency_deleted(&task, dependency);
1922 if task.status == TaskStatus::Blocked {
1923 still_blocked.push(task.id.clone());
1924 } else {
1925 released.push(task.id.clone());
1926 }
1927 }
1928 (released, still_blocked)
1929 }
1930
1931 fn remove_record_with_attachments(
1935 &self,
1936 resolved: &str,
1937 remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
1938 ) -> Result<()> {
1939 self.sweep_removed_attachments();
1946 let attachments = self.attachments_dir(resolved);
1947 let aside = self.root.join(format!("{resolved}.attachments.removing"));
1948 let moved = match std::fs::rename(&attachments, &aside) {
1949 Ok(()) => true,
1950 Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
1951 Err(e) => {
1952 return Err(e).with_context(|| format!("remove {}", attachments.display()));
1953 }
1954 };
1955 let path = self.path_of(resolved);
1956 if let Err(e) = remove_record(&path) {
1957 if moved {
1958 let _ = std::fs::rename(&aside, &attachments);
1959 }
1960 return Err(e).with_context(|| format!("remove {}", path.display()));
1961 }
1962 if moved {
1963 if let Err(e) = std::fs::remove_dir_all(&aside) {
1964 tracing::warn!("leftover attachments {}: {e}", aside.display());
1965 }
1966 }
1967 let lock = self.lock_path(resolved);
1968 if let Err(e) = std::fs::remove_file(&lock) {
1969 if e.kind() != std::io::ErrorKind::NotFound {
1970 return Err(e).with_context(|| format!("remove {}", lock.display()));
1971 }
1972 }
1973 Ok(())
1974 }
1975
1976 fn sweep_removed_attachments(&self) {
1982 let Ok(entries) = std::fs::read_dir(&self.root) else {
1983 return;
1984 };
1985 for entry in entries.flatten() {
1986 let name = entry.file_name();
1987 let name = name.to_string_lossy();
1988 let Some(id) = name.strip_suffix(".attachments.removing") else {
1989 continue;
1990 };
1991 if !self.path_of(id).exists() {
1994 if let Err(e) = std::fs::remove_dir_all(entry.path()) {
1995 tracing::warn!("leftover attachments {}: {e}", entry.path().display());
1996 }
1997 }
1998 }
1999 self.sweep_tombstones();
2000 }
2001
2002 fn sweep_tombstones(&self) {
2011 let Ok(entries) = std::fs::read_dir(&self.root) else {
2012 return;
2013 };
2014 let tasks = self.list();
2015 for entry in entries.flatten() {
2016 let name = entry.file_name();
2017 let name = name.to_string_lossy();
2018 let Some(id) = name.strip_suffix(".removed") else {
2019 continue;
2020 };
2021 if self.path_of(id).exists()
2022 || tasks.iter().any(|t| t.blocked_by.iter().any(|b| b == id))
2023 {
2024 continue;
2025 }
2026 let old = entry
2027 .metadata()
2028 .and_then(|m| m.modified())
2029 .ok()
2030 .and_then(|m| m.elapsed().ok())
2031 .is_some_and(|age| age >= TOMBSTONE_GRACE);
2032 if old {
2033 let _ = std::fs::remove_file(entry.path());
2034 }
2035 }
2036 }
2037
2038 fn lock_path(&self, id: &str) -> PathBuf {
2041 self.root.join(format!("{id}.lock"))
2042 }
2043
2044 pub fn list(&self) -> Vec<Task> {
2057 let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
2058 .into_iter()
2059 .flatten()
2060 .flatten()
2061 .map(|e| e.path())
2062 .filter(|p| p.extension().is_some_and(|x| x == "json"))
2063 .filter_map(|p| read_path(&p).ok())
2064 .collect();
2065 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
2066 tasks
2067 }
2068
2069 pub fn superseded(&self) -> HashMap<String, String> {
2082 let mut by = HashMap::new();
2083 for task in self.list() {
2084 for earlier in &task.runs {
2085 if let Some(later) = task.successor_of(earlier) {
2086 by.insert(earlier.clone(), later.clone());
2087 }
2088 }
2089 }
2090 by
2091 }
2092
2093 pub fn superseded_by(&self, run: &str) -> Option<String> {
2101 for task in self.list() {
2102 if task.runs.iter().any(|r| r == run) {
2103 return task.successor_of(run).cloned();
2104 }
2105 }
2106 None
2107 }
2108
2109 pub fn latest_attempt(&self, run: &str) -> Option<String> {
2120 for task in self.list() {
2121 if task.runs.iter().any(|r| r == run) {
2122 return task.runs.last().filter(|last| **last != run).cloned();
2123 }
2124 }
2125 None
2126 }
2127
2128 pub fn next_runnable(&self) -> Option<Task> {
2133 let mut runnable: Vec<Task> = self
2134 .list()
2135 .into_iter()
2136 .filter(|t| t.status.runnable())
2137 .collect();
2138 runnable.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
2139 runnable.into_iter().next()
2140 }
2141
2142 pub fn claim(&self, id: &str) -> Result<Claim> {
2149 std::fs::create_dir_all(&self.root)
2150 .with_context(|| format!("create {}", self.root.display()))?;
2151 let path = self.lock_path(id);
2152 match std::fs::OpenOptions::new()
2153 .write(true)
2154 .create_new(true)
2155 .open(&path)
2156 {
2157 Ok(mut f) => {
2158 use std::io::Write as _;
2159 let _ = writeln!(f, "{}", std::process::id());
2161 Ok(Claim { path })
2162 }
2163 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2164 bail!("task {id} is already claimed ({} exists)", path.display())
2165 }
2166 Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
2167 }
2168 }
2169
2170 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
2172 if self.path_of(prefix).is_file() {
2173 return Ok(prefix.to_owned());
2174 }
2175 let hits: Vec<String> = self
2176 .list()
2177 .into_iter()
2178 .map(|t| t.id)
2179 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix))
2180 .collect();
2181 match hits.len() {
2182 1 => Ok(hits.into_iter().next().expect("exactly one hit")),
2183 0 => bail!("no task matches `{prefix}`"),
2184 _ => bail!(
2185 "`{prefix}` matches {} tasks: {}",
2186 hits.len(),
2187 hits.join(", ")
2188 ),
2189 }
2190 }
2191
2192 pub fn revision(&self) -> u64 {
2199 self.revision_excluding(&std::collections::BTreeSet::new())
2200 }
2201
2202 pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
2205 use std::hash::{Hash as _, Hasher as _};
2206
2207 let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
2208 .into_iter()
2209 .flatten()
2210 .flatten()
2211 .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
2212 .filter(|e| {
2213 let path = e.path();
2214 !path
2215 .file_stem()
2216 .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
2217 })
2218 .filter_map(|e| {
2219 let name = e.file_name().to_string_lossy().into_owned();
2220 let mtime = e
2221 .metadata()
2222 .ok()?
2223 .modified()
2224 .ok()?
2225 .duration_since(std::time::UNIX_EPOCH)
2226 .ok()?
2227 .as_millis() as u64;
2228 Some((name, mtime))
2229 })
2230 .collect();
2231
2232 if entries.is_empty() {
2233 return 0;
2234 }
2235
2236 entries.sort_unstable();
2237 let mut hasher = std::hash::DefaultHasher::new();
2238 for (name, mtime) in &entries {
2239 name.hash(&mut hasher);
2240 mtime.hash(&mut hasher);
2241 }
2242 let h = hasher.finish();
2243 if h == 0 { 1 } else { h }
2244 }
2245}
2246
2247const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
2250
2251#[derive(Debug, Clone)]
2253pub struct Removal {
2254 pub id: String,
2256 pub released: Vec<String>,
2260 pub still_blocked: Vec<String>,
2263}
2264
2265#[derive(Debug)]
2267pub struct Claim {
2268 path: PathBuf,
2269}
2270
2271impl Drop for Claim {
2272 fn drop(&mut self) {
2273 let _ = std::fs::remove_file(&self.path);
2274 }
2275}
2276
2277pub fn title_from(instruction: &str, max: usize) -> String {
2280 let Some(line) = first_line(instruction) else {
2281 return "(empty task)".to_owned();
2282 };
2283 if line.chars().count() <= max {
2284 return line.to_owned();
2285 }
2286 let head: String = line.chars().take(max.saturating_sub(1)).collect();
2287 format!("{head}…")
2288}
2289
2290pub(crate) fn first_line(instruction: &str) -> Option<&str> {
2296 let line = instruction
2297 .lines()
2298 .map(str::trim)
2299 .find(|l| !l.is_empty())?
2300 .trim_start_matches(['#', '-', '*', '>', ' '])
2301 .trim();
2302 (!line.is_empty()).then_some(line)
2303}
2304
2305fn read_path(path: &Path) -> Result<Task> {
2306 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
2307 let task: Task =
2308 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2309 if task.schema > SCHEMA {
2315 bail!(
2316 "task {} was written by a different magi (schema {}, this build \
2317 speaks {SCHEMA})",
2318 task.id,
2319 task.schema
2320 );
2321 }
2322 Ok(task)
2323}
2324
2325pub fn missing_blockers(
2339 queue: &Queue,
2340 questions: &Questions,
2341 blocked_by: &[String],
2342) -> Vec<String> {
2343 blocked_by
2344 .iter()
2345 .filter(|id| {
2346 !queue.path_of(id).is_file()
2347 && !questions.path_of(id).is_file()
2348 && !queue.was_deleted(id)
2349 })
2350 .cloned()
2351 .collect()
2352}
2353
2354pub fn deleted_blockers(queue: &Queue, blocked_by: &[String]) -> Vec<String> {
2359 blocked_by
2360 .iter()
2361 .filter(|id| queue.was_deleted(id))
2362 .cloned()
2363 .collect()
2364}
2365
2366pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
2378 missing_blocker_hold_reason_in(blocked_by, missing, "en")
2379}
2380
2381pub fn missing_blocker_hold_reason_in(
2383 blocked_by: &[String],
2384 missing: &[String],
2385 language: &str,
2386) -> String {
2387 if crate::lang::is_japanese(language) {
2388 format!(
2389 "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
2390 blocked_by.join(", "),
2391 missing.join(", "),
2392 )
2393 } else {
2394 format!(
2395 "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
2396 blocked_by.join(", "),
2397 missing.join(", "),
2398 )
2399 }
2400}
2401
2402pub fn short(id: &str) -> &str {
2404 id.split('-').next_back().unwrap_or(id)
2405}
2406
2407fn new_id() -> String {
2408 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
2409 let seed = crate::rng::entropy();
2410 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
2411}
2412
2413#[cfg(test)]
2414mod tests {
2415 #[test]
2416 fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
2417 let mut base = RunOverrides {
2418 merge: Some("pr".to_owned()),
2419 candidates: Some(3),
2420 ..RunOverrides::default()
2421 };
2422 base.merge_over(&RunOverrides {
2423 merge: Some("none".to_owned()),
2424 seed: Some(7),
2425 ..RunOverrides::default()
2426 });
2427 assert_eq!(base.merge.as_deref(), Some("none"));
2428 assert_eq!(base.candidates, Some(3));
2429 assert_eq!(base.seed, Some(7));
2430 }
2431
2432 #[test]
2433 fn an_already_landed_task_is_done_with_its_attempt_refunded() {
2434 let mut t = task("relanded");
2435 t.attempts = 1;
2436 t.status = TaskStatus::Running;
2437 t.already_landed("already in main as 0e368de");
2438 assert_eq!(t.status, TaskStatus::Done);
2439 assert_eq!(t.attempts, 0);
2440 assert!(t.hold_reason.is_none());
2441 assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
2442 }
2443
2444 #[test]
2445 fn missing_blocker_reason_follows_the_language() {
2446 let b = vec!["a".to_owned()];
2447 let en = missing_blocker_hold_reason_in(&b, &b, "en");
2448 assert_eq!(en, missing_blocker_hold_reason(&b, &b));
2449 assert!(en.starts_with("blocked on a"));
2450 assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
2451 assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
2452 }
2453
2454 use super::*;
2455
2456 #[test]
2457 fn triage_applied_survives_release_and_old_records_read_as_empty() {
2458 let mut t = Task::new(
2459 "t".to_owned(),
2460 "i".to_owned(),
2461 PathBuf::from("r"),
2462 Source::Human,
2463 );
2464 t.mark_triage_applied("q1");
2465 t.mark_triage_applied("q1");
2466 t.hold_machine(Some("x".to_owned()));
2467 t.release();
2468 assert_eq!(t.triage_applied, ["q1"]);
2469 assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
2470
2471 let mut v = serde_json::to_value(&t).unwrap();
2472 v.as_object_mut().unwrap().remove("triage_applied");
2473 let old: Task = serde_json::from_value(v).unwrap();
2474 assert!(old.triage_applied.is_empty());
2475 }
2476
2477 #[test]
2478 fn task_counts_of_empty_is_all_zero() {
2479 assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
2480 }
2481
2482 #[test]
2483 fn task_counts_of_tallies_every_status() {
2484 let mut queued = Task::new(
2485 "q".to_owned(),
2486 "i".to_owned(),
2487 PathBuf::from("."),
2488 Source::Human,
2489 );
2490 queued.status = TaskStatus::Queued;
2491 let mut running = queued.clone();
2492 running.status = TaskStatus::Running;
2493 let mut done = queued.clone();
2494 done.status = TaskStatus::Done;
2495 let mut failed = queued.clone();
2496 failed.status = TaskStatus::Failed;
2497 let mut held = queued.clone();
2498 held.status = TaskStatus::Held;
2499 let mut blocked = queued.clone();
2500 blocked.status = TaskStatus::Blocked;
2501
2502 let mut parked = queued.clone();
2503 parked.status = TaskStatus::Parked;
2504
2505 let counts = TaskCounts::of(&[
2506 queued,
2507 running,
2508 done.clone(),
2509 done,
2510 failed,
2511 held,
2512 blocked,
2513 parked,
2514 ]);
2515 assert_eq!(
2516 counts,
2517 TaskCounts {
2518 queued: 1,
2519 running: 1,
2520 done: 2,
2521 failed: 1,
2522 held: 1,
2523 blocked: 1,
2524 parked: 1,
2525 }
2526 );
2527 }
2528
2529 fn queue() -> (tempfile::TempDir, Queue) {
2532 let dir = tempfile::tempdir().unwrap();
2533 let q = Queue::at(dir.path().join("queue"));
2534 (dir, q)
2535 }
2536
2537 #[test]
2538 fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
2539 let dir = tempfile::tempdir().unwrap();
2540 let q = Queue::at(dir.path().join("queue"));
2541 let mut t = task("held");
2542 q.put(&mut t).unwrap();
2543 assert_eq!(
2544 crate::notices::Notices::at(dir.path().join("notifications"))
2545 .list()
2546 .len(),
2547 0
2548 );
2549 t.hold_machine(Some("out of attempts".to_owned()));
2550 q.put(&mut t).unwrap();
2551 let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
2552 assert_eq!(listed.len(), 1);
2553 assert!(listed[0].message.contains("out of attempts"));
2554 }
2555
2556 fn task(title: &str) -> Task {
2557 Task::new(
2558 title.to_owned(),
2559 format!("do {title}"),
2560 PathBuf::from("."),
2561 Source::Human,
2562 )
2563 }
2564
2565 #[test]
2566 fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
2567 let mut t = task("retried");
2568 assert!(
2569 t.earlier_attempts().is_empty(),
2570 "a first attempt takes nothing over"
2571 );
2572 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
2573 assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
2574 assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
2576 assert_eq!(t.successor_of("bbbb"), None);
2577 assert_eq!(t.successor_of("zzzz"), None);
2578 }
2579
2580 #[test]
2581 fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
2582 let (_dir, q) = queue();
2583 let mut t = task("retried");
2584 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
2585 q.put(&mut t).unwrap();
2586
2587 assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
2588 assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
2589 assert_eq!(
2590 q.superseded_by("cccc"),
2591 None,
2592 "the latest attempt replaces nothing"
2593 );
2594 assert_eq!(
2595 q.superseded_by("never-heard-of-it"),
2596 None,
2597 "a run belonging to no task on this queue is not superseded"
2598 );
2599
2600 let mut by = HashMap::new();
2601 by.insert("aaaa".to_owned(), "bbbb".to_owned());
2602 by.insert("bbbb".to_owned(), "cccc".to_owned());
2603 assert_eq!(
2604 q.superseded(),
2605 by,
2606 "the whole-map and single-run forms must agree"
2607 );
2608 }
2609
2610 #[test]
2611 fn superseded_attempts_is_empty_until_the_task_is_done() {
2612 let mut t = task("retried");
2613 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
2614 t.status = TaskStatus::Failed;
2615 assert_eq!(
2616 t.superseded_attempts(true),
2617 &[] as &[String],
2618 "a task still retrying has no attempt yet that a later one made moot"
2619 );
2620
2621 t.status = TaskStatus::Running;
2622 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
2623 }
2624
2625 #[test]
2626 fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
2627 let mut t = task("retried");
2628 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
2629 t.status = TaskStatus::Done;
2630 assert_eq!(
2631 t.superseded_attempts(true),
2632 &["aaaa".to_owned(), "bbbb".to_owned()],
2633 "cccc is the attempt whose success made the task done, and stays out"
2634 );
2635 }
2636
2637 #[test]
2638 fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
2639 let mut t = task("first try landed");
2640 t.runs = vec!["aaaa".to_owned()];
2641 t.status = TaskStatus::Done;
2642 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
2643 }
2644
2645 #[test]
2646 fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
2647 let mut t = task("closed by hand after a manual merge");
2654 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
2655 t.status = TaskStatus::Done;
2656 assert_eq!(
2657 t.superseded_attempts(false),
2658 &[] as &[String],
2659 "nothing here is provably why the task is done, so nothing is superseded"
2660 );
2661 }
2662
2663 #[test]
2664 fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
2665 let (_dir, q) = queue();
2666 let mut t = task("retried twice");
2667 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
2668 q.put(&mut t).unwrap();
2669
2670 assert_eq!(
2671 q.latest_attempt("aaaa"),
2672 Some("cccc".to_owned()),
2673 "an old attempt points straight at the chain's current head, not the \
2674 next attempt in the middle of it"
2675 );
2676 assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
2677 assert_eq!(
2678 q.latest_attempt("cccc"),
2679 None,
2680 "the latest attempt is not superseded by anything"
2681 );
2682 assert_eq!(
2683 q.latest_attempt("never-heard-of-it"),
2684 None,
2685 "a run belonging to no task on this queue is not superseded"
2686 );
2687 }
2688
2689 #[test]
2690 fn a_markdown_heading_is_the_title_not_decoration() {
2691 assert_eq!(
2696 title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
2697 "Rework the config loader"
2698 );
2699 assert_eq!(title_from("- fix the thing", 40), "fix the thing");
2700 assert_eq!(title_from("> quoted task", 40), "quoted task");
2701 assert_eq!(title_from(" \n\n", 40), "(empty task)");
2703 assert_eq!(title_from("###\n", 40), "(empty task)");
2704 }
2705
2706 #[test]
2707 fn a_long_title_is_elided_by_characters_not_bytes() {
2708 let long = "課題".repeat(30);
2710 let title = title_from(&long, 10);
2711 assert_eq!(title.chars().count(), 10);
2712 assert!(title.ends_with('…'));
2713 }
2714
2715 #[test]
2716 fn priority_wins_and_ties_break_oldest_first() {
2717 let (_dir, q) = queue();
2718 let mut a = task("first");
2719 let mut b = task("second");
2720 let mut c = task("urgent");
2721 a.id = "20260101-000001-aaaa".to_owned();
2723 b.id = "20260101-000002-bbbb".to_owned();
2724 c.id = "20260101-000003-cccc".to_owned();
2725 c.priority = 5;
2726 for t in [&mut a, &mut b, &mut c] {
2727 q.put(t).unwrap();
2728 }
2729
2730 assert_eq!(q.next_runnable().unwrap().id, c.id);
2732 c.hold_machine(None);
2733 q.put(&mut c).unwrap();
2734 assert_eq!(q.next_runnable().unwrap().id, a.id);
2736 assert_eq!(q.list().len(), 3, "b is still waiting its turn");
2737 }
2738
2739 #[test]
2740 fn a_blocked_task_never_starves_another_runnable_one() {
2741 let (_dir, q) = queue();
2742 let mut blocked = task("blocked");
2743 blocked.block(vec!["something".to_owned()], None);
2744 q.put(&mut blocked).unwrap();
2745
2746 let mut runnable = task("free to go");
2747 q.put(&mut runnable).unwrap();
2748
2749 let next = q.next_runnable().expect("a runnable task is still offered");
2750 assert_eq!(next.id, runnable.id);
2751 }
2752
2753 #[test]
2754 fn a_held_task_is_never_offered_to_the_loop() {
2755 let (_dir, q) = queue();
2756 let mut t = task("held");
2757 q.put(&mut t).unwrap();
2758 assert!(q.next_runnable().is_some());
2759
2760 t.hold_machine(None);
2761 q.put(&mut t).unwrap();
2762 assert!(
2763 q.next_runnable().is_none(),
2764 "a held task must wait for a human"
2765 );
2766
2767 t.status = TaskStatus::Failed;
2769 q.put(&mut t).unwrap();
2770 assert!(q.next_runnable().is_some());
2771 }
2772
2773 #[test]
2774 fn attempts_are_capped_and_then_the_task_is_held() {
2775 let mut t = task("doomed");
2776
2777 t.start("run-1".to_owned());
2778 t.fail("gate red", 2);
2779 assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
2780
2781 t.start("run-2".to_owned());
2782 t.fail("gate red", 2);
2783 assert_eq!(
2784 t.status,
2785 TaskStatus::Held,
2786 "out of attempts: stop spending money on it"
2787 );
2788 assert_eq!(t.runs, ["run-1", "run-2"]);
2789 assert_eq!(t.last_error.as_deref(), Some("gate red"));
2790 assert_eq!(
2791 t.hold_reason.as_deref(),
2792 Some("gate red"),
2793 "the hold must say why, not leave hold_reason null next to a \
2794 populated last_error"
2795 );
2796 }
2797
2798 #[test]
2799 fn handing_off_a_task_records_a_hold_reason_too() {
2800 let mut t = task("left a pull request");
2801 t.start("run-1".to_owned());
2802 t.handed_off("run ended with a pull request open [run run-1]");
2803 assert_eq!(t.status, TaskStatus::Held);
2804 assert_eq!(t.hold_source, Some(HoldSource::Machine));
2805 assert_eq!(
2806 t.hold_reason.as_deref(),
2807 Some("run ended with a pull request open [run run-1]")
2808 );
2809 assert_eq!(t.hold_reason, t.last_error);
2810 }
2811
2812 #[test]
2813 fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
2814 let mut t = task("stalled by quota");
2815
2816 t.start("run-1".to_owned());
2817 assert_eq!(t.attempts, 1);
2818 t.stall("judge-1, judge-2 out of quota");
2819 assert_eq!(
2820 t.attempts, 0,
2821 "a closed quota window must not spend the task's retry budget"
2822 );
2823 assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
2824 assert_eq!(
2825 t.last_error.as_deref(),
2826 Some("judge-1, judge-2 out of quota")
2827 );
2828
2829 for _ in 0..20 {
2832 t.start("run-n".to_owned());
2833 t.stall("still out of quota");
2834 }
2835 t.start("run-real".to_owned());
2836 t.fail("gate red", 2);
2837 assert_eq!(
2838 t.status,
2839 TaskStatus::Failed,
2840 "the first attempt that was really judged is attempt one"
2841 );
2842 }
2843
2844 #[test]
2845 fn releasing_a_held_task_gives_it_a_real_second_chance() {
2846 let mut t = task("retry me");
2847 t.start("run-1".to_owned());
2848 t.fail("gate red", 1);
2849 assert_eq!(t.status, TaskStatus::Held);
2850
2851 t.release();
2852 assert_eq!(t.status, TaskStatus::Queued);
2853 assert_eq!(t.attempts, 0);
2856 assert!(t.last_error.is_none());
2857 assert_eq!(
2858 t.runs.len(),
2859 1,
2860 "history is kept: attempts reset, evidence does not"
2861 );
2862 }
2863
2864 #[test]
2865 fn a_hold_reason_survives_and_a_release_clears_it() {
2866 let mut t = task("waiting on something else");
2867 t.hold_manual(Some(
2868 "waiting for 20260101-000000-aaaa to land first".to_owned(),
2869 ));
2870 assert_eq!(t.status, TaskStatus::Held);
2871 assert_eq!(
2872 t.hold_reason.as_deref(),
2873 Some("waiting for 20260101-000000-aaaa to land first")
2874 );
2875
2876 t.hold_manual(None);
2878 assert_eq!(
2879 t.hold_reason.as_deref(),
2880 Some("waiting for 20260101-000000-aaaa to land first"),
2881 "a bare re-hold keeps whatever a human already wrote down"
2882 );
2883
2884 let mut plain = task("no reason given");
2886 plain.hold_manual(None);
2887 assert_eq!(plain.status, TaskStatus::Held);
2888 assert!(plain.hold_reason.is_none());
2889
2890 t.release();
2891 assert_eq!(t.status, TaskStatus::Queued);
2892 assert!(
2893 t.hold_reason.is_none(),
2894 "a stale reason must not greet the next person who holds this task"
2895 );
2896 }
2897
2898 #[test]
2899 fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
2900 let mut t = task("landed by hand while held");
2905 t.hold_manual(Some("waiting on 3ed9".to_owned()));
2906 assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
2907
2908 t.succeed();
2909 assert_eq!(t.status, TaskStatus::Done);
2910 assert!(
2911 t.hold_reason.is_none(),
2912 "a done task cannot still be waiting on something"
2913 );
2914 }
2915
2916 #[test]
2917 fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
2918 let mut held = task("held straight out of blocked");
2925 held.block(
2926 vec!["20260101-000000-dead".to_owned()],
2927 Some("waiting on the migration script".to_owned()),
2928 );
2929 assert_eq!(held.status, TaskStatus::Blocked);
2930
2931 held.hold_manual(None);
2932 assert_eq!(held.status, TaskStatus::Held);
2933 assert!(
2934 held.blocked_by.is_empty(),
2935 "hold overrides the wait, same as release"
2936 );
2937 assert!(held.block_reason.is_none());
2938
2939 let mut done = task("closed straight out of blocked");
2940 done.block(
2941 vec!["20260101-000000-dead".to_owned()],
2942 Some("waiting on the migration script".to_owned()),
2943 );
2944 done.succeed();
2945 assert_eq!(done.status, TaskStatus::Done);
2946 assert!(
2947 done.blocked_by.is_empty(),
2948 "a done task cannot still be waiting on a dependency"
2949 );
2950 assert!(done.block_reason.is_none());
2951 }
2952
2953 #[test]
2954 fn a_blocked_task_is_never_offered_to_the_loop() {
2955 let mut t = task("blocked");
2956 assert!(t.status.runnable());
2957 t.block(
2958 vec!["dep-id".to_owned()],
2959 Some("waits on dep-id".to_owned()),
2960 );
2961 assert_eq!(t.status, TaskStatus::Blocked);
2962 assert!(!t.status.runnable());
2963 assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
2964 }
2965
2966 #[test]
2967 fn unblocking_the_last_dependency_returns_the_task_to_queued() {
2968 let mut t = task("blocked on two");
2969 t.block(
2970 vec!["a".to_owned(), "b".to_owned()],
2971 Some("waits on a and b".to_owned()),
2972 );
2973
2974 t.unblock("a");
2975 assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
2976 assert_eq!(t.blocked_by, ["b"]);
2977
2978 t.unblock("b");
2979 assert_eq!(t.status, TaskStatus::Queued);
2980 assert!(t.blocked_by.is_empty());
2981 assert!(t.block_reason.is_none());
2982 }
2983
2984 #[test]
2985 fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
2986 let mut t = task("never blocked");
2987 t.unblock("whatever");
2988 assert_eq!(t.status, TaskStatus::Queued);
2989 }
2990
2991 #[test]
2992 fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
2993 let mut t = task("held, then asked about");
2998 t.hold_machine(Some("out of attempts".to_owned()));
2999 assert_eq!(t.status, TaskStatus::Held);
3000
3001 t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
3002 assert_eq!(t.status, TaskStatus::Blocked);
3003
3004 t.record_answer("what now?".to_owned(), "leave it held".to_owned());
3005 t.unblock("q1");
3006 assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
3007 assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
3008 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3009 assert!(t.blocked_from.is_none(), "consumed once restored");
3010 }
3011
3012 #[test]
3013 fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
3014 let mut t = task("manually held, then asked about");
3015 t.hold_manual(Some("waiting on a dependency".to_owned()));
3016
3017 t.block(vec!["q1".to_owned()], None);
3018 t.unblock("q1");
3019
3020 assert_eq!(t.status, TaskStatus::Held);
3021 assert_eq!(t.hold_source, Some(HoldSource::Manual));
3022 }
3023
3024 #[test]
3025 fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
3026 let mut t = task("held, blocked twice");
3030 t.hold_machine(None);
3031 t.block(vec!["q1".to_owned()], Some("first".to_owned()));
3032 t.block(
3033 vec!["q1".to_owned(), "q2".to_owned()],
3034 Some("second".to_owned()),
3035 );
3036
3037 t.unblock("q1");
3038 assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
3039 t.unblock("q2");
3040 assert_eq!(t.status, TaskStatus::Held);
3041 }
3042
3043 #[test]
3044 fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
3045 let mut t = task("blocked mid-run");
3049 t.start("run-1".to_owned());
3050 assert_eq!(t.status, TaskStatus::Running);
3051
3052 t.block(vec!["q1".to_owned()], None);
3053 t.unblock("q1");
3054 assert_eq!(t.status, TaskStatus::Queued);
3055 }
3056
3057 #[test]
3058 fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
3059 let mut t = task("legacy record, held before it was blocked");
3065 t.hold_source = Some(HoldSource::Machine);
3066 t.hold_reason = Some("legacy hold reason".to_owned());
3067 t.status = TaskStatus::Blocked;
3068 t.blocked_by = vec!["q1".to_owned()];
3069 t.blocked_from = None;
3070
3071 t.unblock("q1");
3072 assert_eq!(t.status, TaskStatus::Held);
3073 }
3074
3075 #[test]
3076 fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
3077 let mut t = task("legacy record, ordinary dependency block");
3078 t.status = TaskStatus::Blocked;
3079 t.blocked_by = vec!["dep".to_owned()];
3080 t.blocked_from = None;
3081
3082 t.unblock("dep");
3083 assert_eq!(t.status, TaskStatus::Queued);
3084 }
3085
3086 #[test]
3087 fn answering_a_question_is_recorded_and_survives_a_release() {
3088 let mut t = task("asked something");
3089 t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
3090 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
3091 t.unblock("q1");
3092 assert_eq!(t.status, TaskStatus::Queued);
3093 assert_eq!(t.answers.len(), 1);
3094 assert_eq!(t.answers[0].answer, "SQLite");
3095
3096 t.release();
3100 assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
3101 }
3102
3103 #[test]
3104 fn a_refused_handover_keeps_the_review_branch_across_release() {
3105 let mut t = task("refused takeover");
3106 t.start("run-1".to_owned());
3107 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
3108 assert_eq!(t.status, TaskStatus::Held);
3109 assert_eq!(t.attempts, 1);
3110 t.release();
3111 assert_eq!(t.status, TaskStatus::Queued);
3112 assert_eq!(t.attempts, 0);
3113 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
3114
3115 t.hold_for_handover(None, "again".to_owned());
3117 t.requeue();
3118 assert!(t.review_branch.is_none());
3119
3120 let mut m = task("manual");
3122 m.review_branch = Some("magi/x/A".to_owned());
3123 m.hold_manual(None);
3124 m.release();
3125 assert!(m.review_branch.is_none());
3126 }
3127
3128 #[test]
3129 fn requesting_review_requeues_the_task_and_remembers_the_branch() {
3130 let mut t = task("blocked run with a surviving branch");
3131 t.start("run-1".to_owned());
3132 t.fail("blocked with major findings", 5);
3133 assert_eq!(t.status, TaskStatus::Failed);
3134
3135 t.request_review("magi/eba2/A".to_owned());
3136 assert_eq!(t.status, TaskStatus::Queued);
3137 assert_eq!(t.attempts, 0);
3138 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
3139
3140 t.release();
3142 assert!(t.review_branch.is_none());
3143 }
3144
3145 #[test]
3146 fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
3147 let mut t = task("retry");
3148 t.start("run-1".to_owned());
3149 t.requeue();
3150 assert!(t.fresh_start);
3151
3152 t.release();
3153 assert!(!t.fresh_start);
3154 }
3155
3156 #[test]
3157 fn priority_can_be_changed_while_queued_but_not_while_running() {
3158 let mut t = task("reprioritise me");
3159 t.set_priority(5).unwrap();
3160 assert_eq!(t.priority, 5);
3161
3162 t.start("run-1".to_owned());
3163 let err = t.set_priority(9).unwrap_err().to_string();
3164 assert!(err.contains("running"), "{err}");
3165 assert_eq!(t.priority, 5, "the rejected write must not partially apply");
3166 }
3167
3168 #[test]
3169 fn interrupt_can_be_marked_while_queued_but_not_while_running() {
3170 let mut t = task("interrupt me");
3171 assert!(!t.interrupt, "off unless asked, same as any other task");
3172
3173 t.set_interrupt(true).unwrap();
3174 assert!(t.interrupt);
3175
3176 t.start("run-1".to_owned());
3177 assert!(
3178 !t.interrupt,
3179 "the mark is one-shot: dispatching the task fulfils it, \
3180 whatever the run that follows ends up doing"
3181 );
3182 let err = t.set_interrupt(true).unwrap_err().to_string();
3183 assert!(err.contains("running"), "{err}");
3184 t.set_interrupt(false).unwrap();
3187 assert!(!t.interrupt);
3188 }
3189
3190 #[test]
3194 fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
3195 let mut t = task("interrupt me");
3196 t.set_interrupt(true).unwrap();
3197 t.start("run-1".to_owned());
3198 t.fail("mock failure", 5);
3199 assert_eq!(t.status, TaskStatus::Failed);
3200 assert!(
3201 !t.interrupt,
3202 "one attempt already spent the mark; a retry is an ordinary \
3203 requeue, not a fresh interrupt request"
3204 );
3205 }
3206
3207 #[test]
3208 fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
3209 let (_dir, q) = queue();
3210 let mut a = task("first filed");
3211 let mut b = task("second filed");
3212 a.id = "20260101-000001-aaaa".to_owned();
3213 b.id = "20260101-000002-bbbb".to_owned();
3214 q.put(&mut a).unwrap();
3215 q.put(&mut b).unwrap();
3216
3217 assert_eq!(
3218 q.next_runnable().unwrap().id,
3219 a.id,
3220 "with equal priority the older task goes first, so a burst of \
3221 new work cannot starve it"
3222 );
3223 assert_eq!(
3224 q.list()[0].id,
3225 b.id,
3226 "but the list an operator reads is newest first, the same as \
3227 before priority existed - a's turn to run does not make it the \
3228 newest task"
3229 );
3230
3231 let mut a = q.get(&a.id).unwrap();
3232 a.set_priority(10).unwrap();
3233 q.put(&mut a).unwrap();
3234
3235 assert_eq!(
3236 q.next_runnable().unwrap().id,
3237 a.id,
3238 "a raised priority must be reflected the moment it is saved"
3239 );
3240 assert_eq!(
3244 q.list()[0].id,
3245 a.id,
3246 "the raised task must sort first in the list an operator reads, \
3247 not only in next_runnable's own ordering"
3248 );
3249 }
3250
3251 #[test]
3252 fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
3253 let mut t = Task::new(
3254 "old title".to_owned(),
3255 "old instruction".to_owned(),
3256 PathBuf::from("/repo"),
3257 Source::Agent {
3258 run: "20260101-000000-beef".to_owned(),
3259 node: "implement".to_owned(),
3260 },
3261 );
3262 let id = t.id.clone();
3263 let created_at = t.created_at;
3264 t.runs.push("20260101-000000-beef".to_owned());
3265
3266 t.edit("new title".to_owned(), "new instruction".to_owned())
3267 .unwrap();
3268
3269 assert_eq!(t.title, "new title");
3270 assert_eq!(t.instruction, "new instruction");
3271 assert_eq!(t.id, id, "editing must not mint a new id");
3272 assert_eq!(t.created_at, created_at);
3273 assert_eq!(
3274 t.source,
3275 Source::Agent {
3276 run: "20260101-000000-beef".to_owned(),
3277 node: "implement".to_owned(),
3278 },
3279 "editing must not turn agent attribution into human"
3280 );
3281 assert_eq!(t.runs, ["20260101-000000-beef"]);
3282 }
3283
3284 #[test]
3285 fn editing_is_refused_once_a_task_is_running_or_finished() {
3286 let mut running = task("in flight");
3287 running.start("run-1".to_owned());
3288 let err = running
3289 .edit("x".to_owned(), "y".to_owned())
3290 .unwrap_err()
3291 .to_string();
3292 assert!(err.contains("running"), "{err}");
3293
3294 let mut done = task("finished");
3295 done.succeed();
3296 let err = done
3297 .edit("x".to_owned(), "y".to_owned())
3298 .unwrap_err()
3299 .to_string();
3300 assert!(err.contains("done"), "{err}");
3301
3302 let mut queued = task("waiting");
3304 queued.edit("x".to_owned(), "y".to_owned()).unwrap();
3305 let mut held = task("parked");
3306 held.hold_machine(None);
3307 held.edit("x".to_owned(), "y".to_owned()).unwrap();
3308 }
3309
3310 #[test]
3311 fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
3312 let (_dir, q) = queue();
3313 let path = q.path_of("20260101-000000-aaaa");
3314 std::fs::create_dir_all(q.root()).unwrap();
3315 std::fs::write(
3316 &path,
3317 serde_json::json!({
3318 "schema": SCHEMA,
3319 "id": "20260101-000000-aaaa",
3320 "title": "from before hold reasons existed",
3321 "instruction": "from before hold reasons existed",
3322 "repo": ".",
3323 "source": { "kind": "human" },
3324 "status": "held",
3325 "created_at": Timestamp::now().to_string(),
3326 "updated_at": Timestamp::now().to_string(),
3327 })
3328 .to_string(),
3329 )
3330 .unwrap();
3331
3332 let task = q.get("20260101-000000-aaaa").expect("must still read");
3333 assert!(task.hold_reason.is_none());
3334 assert!(task.operator_held());
3335 }
3336
3337 #[test]
3338 fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
3339 let (_dir, q) = queue();
3340 let path = q.path_of("20260101-000000-bbbb");
3341 std::fs::create_dir_all(q.root()).unwrap();
3342 std::fs::write(
3343 &path,
3344 serde_json::json!({
3345 "schema": 2,
3346 "id": "20260101-000000-bbbb",
3347 "title": "old manual recovery",
3348 "instruction": "old manual recovery",
3349 "repo": ".",
3350 "source": { "kind": "human" },
3351 "status": "held",
3352 "hold_reason": "active manual recovery run20260912-224242-daf5",
3353 "created_at": Timestamp::now().to_string(),
3354 "updated_at": Timestamp::now().to_string(),
3355 })
3356 .to_string(),
3357 )
3358 .unwrap();
3359
3360 let task = q.get("20260101-000000-bbbb").expect("must still read");
3361 assert_eq!(task.hold_source, None);
3362 assert!(task.operator_held());
3363 }
3364
3365 #[test]
3366 fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
3367 let (_dir, q) = queue();
3368 let path = q.path_of("20260101-000000-aaaa");
3369 std::fs::create_dir_all(q.root()).unwrap();
3370 std::fs::write(
3371 &path,
3372 serde_json::json!({
3373 "schema": SCHEMA,
3374 "id": "20260101-000000-aaaa",
3375 "title": "from before diagnostics existed",
3376 "instruction": "from before diagnostics existed",
3377 "repo": ".",
3378 "source": { "kind": "human" },
3379 "status": "held",
3380 "created_at": Timestamp::now().to_string(),
3381 "updated_at": Timestamp::now().to_string(),
3382 })
3383 .to_string(),
3384 )
3385 .unwrap();
3386
3387 let task = q.get("20260101-000000-aaaa").expect("must still read");
3388 assert!(task.diagnostic.is_none());
3389 }
3390
3391 #[test]
3392 fn a_schema_1_task_with_no_blocking_fields_still_reads() {
3393 let (_dir, q) = queue();
3397 let path = q.path_of("20260101-000000-aaaa");
3398 std::fs::create_dir_all(q.root()).unwrap();
3399 std::fs::write(
3400 &path,
3401 serde_json::json!({
3402 "schema": 1,
3403 "id": "20260101-000000-aaaa",
3404 "title": "from before blocking existed",
3405 "instruction": "from before blocking existed",
3406 "repo": ".",
3407 "source": { "kind": "human" },
3408 "status": "queued",
3409 "created_at": Timestamp::now().to_string(),
3410 "updated_at": Timestamp::now().to_string(),
3411 })
3412 .to_string(),
3413 )
3414 .unwrap();
3415
3416 let task = q.get("20260101-000000-aaaa").expect("must still read");
3417 assert!(task.blocked_by.is_empty());
3418 assert!(task.block_reason.is_none());
3419 assert!(task.answers.is_empty());
3420 assert!(task.review_branch.is_none());
3421 }
3422
3423 #[test]
3424 fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
3425 let mut held = task("diagnosed");
3430 held.start("run-1".to_owned());
3431 held.fail("gate red", 1);
3432 held.diagnostic = Some("cargo test failed: ...".to_owned());
3433 assert_eq!(held.status, TaskStatus::Held);
3434
3435 held.release();
3436 assert!(held.diagnostic.is_none());
3437
3438 held.diagnostic = Some("cargo test failed: ...".to_owned());
3439 held.succeed();
3440 assert!(held.diagnostic.is_none());
3441 }
3442
3443 #[test]
3444 fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
3445 let mut t = task("retried");
3446 t.start("run-1".to_owned());
3447 t.diagnostic = Some("stale evidence from a previous hold".to_owned());
3448 t.fail("unrelated config error", 5);
3449 assert_eq!(t.status, TaskStatus::Failed);
3450 assert!(
3451 t.diagnostic.is_none(),
3452 "fail() must not let an old diagnostic outlive the run that produced it"
3453 );
3454 }
3455
3456 #[test]
3457 fn a_claim_is_exclusive_and_releases_on_drop() {
3458 let (_dir, q) = queue();
3459 let mut t = task("contended");
3460 q.put(&mut t).unwrap();
3461
3462 let held = q.claim(&t.id).unwrap();
3463 assert!(
3464 q.claim(&t.id).is_err(),
3465 "two daemons must not drive one task into two runs"
3466 );
3467 drop(held);
3468 assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
3469 }
3470
3471 #[test]
3472 fn a_round_trip_survives_disk() {
3473 let (_dir, q) = queue();
3474 let mut t = Task::new(
3475 "titled".to_owned(),
3476 "body".to_owned(),
3477 PathBuf::from("/repo"),
3478 Source::Agent {
3479 run: "20260101-000000-beef".to_owned(),
3480 node: "implement".to_owned(),
3481 },
3482 );
3483 t.priority = 3;
3484 q.put(&mut t).unwrap();
3485
3486 let back = q.get(&t.id).unwrap();
3487 assert_eq!(back.id, t.id);
3488 assert_eq!(back.priority, 3);
3489 assert_eq!(back.source.label(), "implement@beef");
3490 assert_eq!(q.get(t.short()).unwrap().id, t.id);
3492 }
3493
3494 #[test]
3495 fn an_unreadable_task_does_not_take_the_queue_down() {
3496 let (_dir, q) = queue();
3497 let mut t = task("fine");
3498 q.put(&mut t).unwrap();
3499 std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
3500
3501 let listed = q.list();
3502 assert_eq!(listed.len(), 1, "the readable task still lists");
3503 assert_eq!(listed[0].id, t.id);
3504 }
3505
3506 #[test]
3507 fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
3508 let (_dir, q) = queue();
3509 let path = q.path_of("20260101-000000-aaaa");
3510 std::fs::create_dir_all(q.root()).unwrap();
3511 std::fs::write(
3512 &path,
3513 serde_json::json!({
3514 "schema": SCHEMA,
3515 "id": "20260101-000000-aaaa",
3516 "title": "from before solo existed",
3517 "instruction": "from before solo existed",
3518 "repo": ".",
3519 "source": { "kind": "human" },
3520 "status": "queued",
3521 "created_at": Timestamp::now().to_string(),
3522 "updated_at": Timestamp::now().to_string(),
3523 })
3524 .to_string(),
3525 )
3526 .unwrap();
3527
3528 let task = q.get("20260101-000000-aaaa").expect("must still read");
3529 assert!(!task.solo, "a queue file with no `solo` field means false");
3530 }
3531
3532 #[test]
3533 fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
3534 let (_dir, q) = queue();
3535 let path = q.path_of("20260101-000000-bbbb");
3536 std::fs::create_dir_all(q.root()).unwrap();
3537 std::fs::write(
3538 &path,
3539 serde_json::json!({
3540 "schema": SCHEMA,
3541 "id": "20260101-000000-bbbb",
3542 "title": "from before urgent existed",
3543 "instruction": "from before urgent existed",
3544 "repo": ".",
3545 "source": { "kind": "human" },
3546 "status": "queued",
3547 "created_at": Timestamp::now().to_string(),
3548 "updated_at": Timestamp::now().to_string(),
3549 })
3550 .to_string(),
3551 )
3552 .unwrap();
3553
3554 let task = q.get("20260101-000000-bbbb").expect("must still read");
3555 assert!(
3556 !task.urgent,
3557 "a queue file with no `urgent` field means false, same as `solo`"
3558 );
3559 }
3560
3561 #[test]
3562 fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
3563 let (_dir, q) = queue();
3564 let mut t = task("from the future");
3565 q.put(&mut t).unwrap();
3566 let path = q.path_of(&t.id);
3567 let body = std::fs::read_to_string(&path)
3568 .unwrap()
3569 .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
3570 std::fs::write(&path, body).unwrap();
3571
3572 let err = q.get(&t.id).unwrap_err().to_string();
3573 assert!(err.contains("schema 99"), "{err}");
3574 }
3575
3576 #[test]
3577 fn revision_moves_when_the_queue_changes() {
3578 let (_dir, q) = queue();
3579 assert_eq!(q.revision(), 0, "an empty queue has no revision");
3580 let mut t = task("first");
3581 q.put(&mut t).unwrap();
3582 assert!(q.revision() > 0, "a written task moves the revision");
3583 }
3584
3585 #[test]
3586 fn revision_moves_when_deleting_an_older_task() {
3587 let (dir, q) = queue();
3588 let questions = Questions::at(dir.path().join("questions"));
3589 let mut t1 = task("older");
3590 q.put(&mut t1).unwrap();
3591 std::thread::sleep(std::time::Duration::from_millis(10));
3593 let mut t2 = task("newer");
3594 q.put(&mut t2).unwrap();
3595
3596 let rev_before = q.revision();
3597 q.remove(&t1.id, false, &questions).unwrap();
3598 let rev_after = q.revision();
3599
3600 assert_ne!(
3601 rev_before, rev_after,
3602 "deleting an older task must change the revision so other clients see the deletion"
3603 );
3604 }
3605
3606 #[test]
3607 fn removing_a_task_takes_it_out_of_the_listing() {
3608 let (dir, q) = queue();
3609 let questions = Questions::at(dir.path().join("questions"));
3610 let mut t = task("delete me");
3611 q.put(&mut t).unwrap();
3612 let removed = q.remove(t.short(), false, &questions).unwrap();
3613 assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
3614 assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
3615 assert!(q.list().is_empty());
3616 assert!(
3617 q.remove(&t.id, false, &questions).is_err(),
3618 "removing twice is an error"
3619 );
3620 }
3621
3622 #[test]
3623 fn removing_a_task_takes_its_stale_lock_with_it() {
3624 let (dir, q) = queue();
3625 let questions = Questions::at(dir.path().join("questions"));
3626 let mut t = task("interrupted");
3627 q.put(&mut t).unwrap();
3628
3629 let claim = q.claim(&t.id).unwrap();
3632 std::mem::forget(claim);
3633 assert!(
3634 q.claim(&t.id).is_err(),
3635 "the orphaned lock is what makes the task look claimed"
3636 );
3637
3638 let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
3640 assert!(err.contains("live daemon"), "{err}");
3641 assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
3642
3643 q.remove(&t.id, false, &questions).unwrap();
3645 assert!(q.list().is_empty());
3646 let mut again = task("interrupted");
3647 again.id = t.id.clone();
3648 q.put(&mut again).unwrap();
3649 assert!(
3650 q.claim(&t.id).is_ok(),
3651 "a task that comes back must be claimable, which a left-behind lock would prevent"
3652 );
3653 }
3654
3655 fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
3656 crate::notices::Notices::at(dir.join("notifications")).list()
3657 }
3658
3659 #[test]
3660 fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
3661 let (dir, q) = queue();
3662 let questions = Questions::at(dir.path().join("questions"));
3663 let mut dep = task("dependency");
3664 q.put(&mut dep).unwrap();
3665 let mut blocked = task("waiting");
3666 blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
3667 q.put(&mut blocked).unwrap();
3668
3669 let removed = q.remove(&dep.id, false, &questions).unwrap();
3670 assert_eq!(removed.released, [blocked.id.clone()]);
3671 assert!(removed.still_blocked.is_empty());
3672
3673 let after = q.get(&blocked.id).unwrap();
3674 assert_eq!(after.status, TaskStatus::Queued);
3675 assert!(after.blocked_by.is_empty());
3676 assert!(after.block_reason.is_none());
3677 let notes = notices_of(dir.path());
3678 assert_eq!(notes.len(), 1, "{notes:?}");
3679 assert_eq!(notes[0].severity, crate::notices::Severity::Info);
3680 assert_eq!(
3681 notes[0].link,
3682 Some(crate::notices::Link::Task {
3683 id: blocked.id.clone(),
3684 })
3685 );
3686 }
3687
3688 #[test]
3689 fn removing_one_of_two_dependencies_keeps_the_other() {
3690 let (dir, q) = queue();
3691 let questions = Questions::at(dir.path().join("questions"));
3692 let mut dep = task("dependency");
3693 q.put(&mut dep).unwrap();
3694 let mut other = task("other");
3695 q.put(&mut other).unwrap();
3696 let mut blocked = task("waiting");
3697 blocked.block(
3698 vec![dep.id.clone(), other.id.clone()],
3699 Some("waits on both".to_owned()),
3700 );
3701 q.put(&mut blocked).unwrap();
3702
3703 let removed = q.remove(&dep.id, false, &questions).unwrap();
3704 assert!(removed.released.is_empty());
3705 assert_eq!(removed.still_blocked, [blocked.id.clone()]);
3706
3707 let after = q.get(&blocked.id).unwrap();
3708 assert_eq!(after.status, TaskStatus::Blocked);
3709 assert_eq!(after.blocked_by, [other.id.clone()]);
3710 assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
3713 }
3714
3715 #[test]
3716 fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
3717 let (dir, q) = queue();
3718 let questions = Questions::at(dir.path().join("questions"));
3719 let mut dep = task("dependency");
3720 q.put(&mut dep).unwrap();
3721 let mut blocked = task("waiting");
3722 blocked.block(vec![dep.id.clone()], None);
3723 q.put(&mut blocked).unwrap();
3724
3725 let claim = q.claim(&blocked.id).unwrap();
3727 let removed = q.remove(&dep.id, false, &questions).unwrap();
3728 assert!(removed.released.is_empty());
3729 drop(claim);
3730
3731 let mut task = q.get(&blocked.id).unwrap();
3732 assert_eq!(task.status, TaskStatus::Blocked);
3733 assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
3734 assert_eq!(q.apply_deleted_blockers(&mut task), [dep.id.clone()]);
3735 assert_eq!(task.status, TaskStatus::Queued);
3736 }
3737
3738 #[test]
3739 fn a_dependency_without_a_tombstone_is_still_missing() {
3740 let (dir, q) = queue();
3741 let questions = Questions::at(dir.path().join("questions"));
3742 let mut dep = task("dependency");
3743 q.put(&mut dep).unwrap();
3744 std::fs::remove_file(q.path_of(&dep.id)).unwrap();
3745 assert_eq!(
3746 missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
3747 [dep.id.clone()]
3748 );
3749 assert!(deleted_blockers(&q, std::slice::from_ref(&dep.id)).is_empty());
3750 }
3751
3752 #[test]
3753 fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
3754 let mut t = task("held");
3755 t.hold_machine(Some("because".to_owned()));
3756 t.blocked_by = vec!["gone".to_owned()];
3757 assert!(!t.dependency_deleted("gone"));
3758 assert_eq!(t.status, TaskStatus::Held);
3759 }
3760
3761 #[test]
3762 fn a_failed_record_removal_takes_the_tombstone_back() {
3763 let (_dir, q) = queue();
3764 let mut t = task("stays");
3765 q.put(&mut t).unwrap();
3766 q.write_tombstone(&t.id).unwrap();
3767 let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
3768 assert!(err.is_err());
3769 assert!(!q.was_deleted(&t.id));
3772 }
3773
3774 fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
3775 let p = dir.join(name);
3776 std::fs::write(&p, body).unwrap();
3777 p
3778 }
3779
3780 #[test]
3781 fn an_attachment_copy_survives_deleting_its_source() {
3782 let (dir, q) = queue();
3783 let src = source_file(dir.path(), "shot.png", "pixels");
3784 let mut t = task("with a picture");
3785 let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
3786 q.put(&mut t).unwrap();
3787 std::fs::remove_file(&src).unwrap();
3788 assert_eq!(names, ["shot.png"]);
3789 let loaded = q.get(&t.id).unwrap();
3790 let paths = q.attachment_paths(&loaded);
3791 assert_eq!(paths.len(), 1);
3792 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
3793 }
3794
3795 #[test]
3796 fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
3797 let (dir, q) = queue();
3798 let mut t = task("bad names");
3799 for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
3800 let src = source_file(dir.path(), name, "x");
3801 assert!(
3802 q.attach(&mut t, &[src]).is_err(),
3803 "`{name}` must be refused"
3804 );
3805 }
3806 assert!(!crate::ask::valid_asset_name("C:foo.png"));
3810 #[cfg(not(windows))]
3811 {
3812 let src = source_file(dir.path(), "C:foo.png", "x");
3813 assert!(q.attach(&mut t, &[src]).is_err());
3814 }
3815 let long = format!("{}.png", "a".repeat(70));
3816 let src = source_file(dir.path(), &long, "x");
3817 assert!(q.attach(&mut t, &[src]).is_err());
3818 assert!(t.attachments.is_empty());
3819 assert!(!q.attachments_dir(&t.id).exists());
3820 }
3821
3822 #[test]
3823 fn a_taken_attachment_name_is_numbered_not_overwritten() {
3824 let (dir, q) = queue();
3825 let a = source_file(dir.path(), "shot.png", "one");
3826 let sub = dir.path().join("other");
3827 std::fs::create_dir_all(&sub).unwrap();
3828 let b = source_file(&sub, "shot.png", "two");
3829 let mut t = task("collision");
3830 q.attach(&mut t, &[a]).unwrap();
3831 q.attach(&mut t, &[b]).unwrap();
3832 assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
3833 let paths = q.attachment_paths(&t);
3834 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
3835 assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
3836 }
3837
3838 #[test]
3839 fn a_renumbered_name_stays_inside_the_length_bound() {
3840 let (dir, q) = queue();
3841 let name = format!("{}.png", "a".repeat(60));
3842 assert_eq!(name.len(), 64);
3843 let a = source_file(dir.path(), &name, "one");
3844 let sub = dir.path().join("other");
3845 std::fs::create_dir_all(&sub).unwrap();
3846 let b = source_file(&sub, &name, "two");
3847 let mut t = task("long");
3848 q.attach(&mut t, &[a, b]).unwrap();
3849 assert_eq!(t.attachments.len(), 2);
3850 assert!(
3851 t.attachments
3852 .iter()
3853 .all(|n| crate::ask::valid_asset_name(n))
3854 );
3855 assert!(t.attachments[1].ends_with("-2.png"));
3856 }
3857
3858 #[test]
3859 fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
3860 let (dir, q) = queue();
3861 let good = source_file(dir.path(), "good.png", "ok");
3862 let mut t = task("partial");
3863 q.attach(&mut t, &[good]).unwrap();
3864 let more = source_file(dir.path(), "more.png", "ok");
3865 let missing = dir.path().join("missing.png");
3866 assert!(q.attach(&mut t, &[more, missing]).is_err());
3867 assert_eq!(t.attachments, ["good.png"]);
3868 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
3869 .unwrap()
3870 .flatten()
3871 .collect();
3872 assert_eq!(on_disk.len(), 1);
3873 }
3874
3875 fn block_put(q: &Queue, t: &Task) -> PathBuf {
3877 let tmp = q.path_of(&t.id).with_extension("json.tmp");
3878 std::fs::create_dir_all(&tmp).unwrap();
3879 tmp
3880 }
3881
3882 #[test]
3883 fn a_failed_put_leaves_no_new_attachment_directory() {
3884 let (dir, q) = queue();
3885 let mut t = task("fresh");
3886 let tmp = block_put(&q, &t);
3887 let src = source_file(dir.path(), "shot.png", "x");
3888 assert!(q.attach_and_put(&mut t, &[src]).is_err());
3889 assert!(t.attachments.is_empty());
3890 assert!(!q.attachments_dir(&t.id).exists());
3891 assert!(!q.path_of(&t.id).exists());
3892 std::fs::remove_dir(tmp).unwrap();
3893 }
3894
3895 #[test]
3896 fn a_failed_put_removes_only_the_copy_it_just_made() {
3897 let (dir, q) = queue();
3898 let mut t = task("edited");
3899 let first = source_file(dir.path(), "first.png", "1");
3900 q.attach_and_put(&mut t, &[first]).unwrap();
3901 block_put(&q, &t);
3902 let second = source_file(dir.path(), "second.png", "2");
3903 assert!(q.attach_and_put(&mut t, &[second]).is_err());
3904 assert_eq!(t.attachments, ["first.png"]);
3905 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
3906 .unwrap()
3907 .flatten()
3908 .map(|e| e.file_name().to_string_lossy().into_owned())
3909 .collect();
3910 assert_eq!(on_disk, ["first.png"]);
3911 assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
3912 }
3913
3914 #[test]
3915 fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
3916 let (dir, q) = queue();
3917 let questions = Questions::at(dir.path().join("questions"));
3918 let gone = task("gone");
3919 let mut other = task("other");
3920 let mut live = task("live");
3921 q.put(&mut other).unwrap();
3922 q.put(&mut live).unwrap();
3923 let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
3926 std::fs::create_dir_all(&orphan).unwrap();
3927 std::fs::write(orphan.join("shot.png"), "x").unwrap();
3928 let busy = q.root.join(format!("{}.attachments.removing", live.id));
3931 std::fs::create_dir_all(&busy).unwrap();
3932
3933 q.remove(&other.id, false, &questions).unwrap();
3934 assert!(!orphan.exists(), "an orphan is swept");
3935 assert!(busy.exists(), "a removal in progress is left alone");
3936 }
3937
3938 #[test]
3939 fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
3940 let (dir, q) = queue();
3941 let questions = Questions::at(dir.path().join("questions"));
3942 let mut t = task("stuck");
3943 let src = source_file(dir.path(), "shot.png", "x");
3944 q.attach_and_put(&mut t, &[src]).unwrap();
3945 let aside = q.root.join(format!("{}.attachments.removing", t.id));
3948 std::fs::create_dir_all(&aside).unwrap();
3949 std::fs::write(aside.join("old.png"), "o").unwrap();
3950 assert!(q.remove(&t.id, false, &questions).is_err());
3951 assert!(q.path_of(&t.id).exists());
3952 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
3953 }
3954
3955 #[test]
3956 fn a_failed_record_removal_puts_the_attachments_back() {
3957 let (dir, q) = queue();
3958 let mut t = task("rollback");
3959 let src = source_file(dir.path(), "shot.png", "x");
3960 q.attach_and_put(&mut t, &[src]).unwrap();
3961 let err = q
3962 .remove_record_with_attachments(&t.id, |_| {
3963 Err(std::io::Error::other("injected failure"))
3964 })
3965 .unwrap_err();
3966 assert!(format!("{err:#}").contains("injected failure"));
3967 assert!(q.path_of(&t.id).exists());
3968 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
3969 assert!(
3970 !q.root
3971 .join(format!("{}.attachments.removing", t.id))
3972 .exists()
3973 );
3974 }
3975
3976 #[test]
3977 fn editing_a_task_keeps_its_attachments() {
3978 let (dir, q) = queue();
3979 let src = source_file(dir.path(), "shot.png", "x");
3980 let mut t = task("editable");
3981 q.attach(&mut t, &[src]).unwrap();
3982 t.edit("new".to_owned(), "new text".to_owned()).unwrap();
3983 q.put(&mut t).unwrap();
3984 assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
3985 }
3986
3987 #[test]
3988 fn removing_a_task_deletes_its_attachments() {
3989 let (dir, q) = queue();
3990 let questions = Questions::at(dir.path().join("questions"));
3991 let src = source_file(dir.path(), "shot.png", "x");
3992 let mut t = task("doomed");
3993 q.attach(&mut t, &[src]).unwrap();
3994 q.put(&mut t).unwrap();
3995 assert!(q.attachments_dir(&t.id).is_dir());
3996 q.remove(&t.id, false, &questions).unwrap();
3997 assert!(!q.attachments_dir(&t.id).exists());
3998 assert!(q.list().is_empty());
3999 }
4000
4001 #[test]
4002 fn a_task_written_before_attachments_still_reads() {
4003 let (_dir, q) = queue();
4004 let mut t = task("old");
4005 q.put(&mut t).unwrap();
4006 let path = q.path_of(&t.id);
4007 let mut v: serde_json::Value =
4008 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
4009 v.as_object_mut().unwrap().remove("attachments");
4010 std::fs::write(&path, v.to_string()).unwrap();
4011 assert!(q.get(&t.id).unwrap().attachments.is_empty());
4012 }
4013
4014 #[test]
4015 fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
4016 let q = Queue::at(PathBuf::from("relative-queue"));
4017 let mut t = task("rel");
4018 t.attachments.push("shot.png".to_owned());
4019 let paths = q.attachment_paths(&t);
4020 assert!(paths[0].is_absolute(), "{}", paths[0].display());
4021 assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
4022 }
4023
4024 #[test]
4025 fn link_run_adds_a_run_once_and_touches_nothing_else() {
4026 let dir = tempfile::tempdir().unwrap();
4027 let queue = Queue::at(dir.path().join("queue"));
4028 let mut t = Task::new(
4029 "t".to_owned(),
4030 "do it".to_owned(),
4031 PathBuf::from("."),
4032 Source::Human,
4033 );
4034 queue.put(&mut t).unwrap();
4035 let before = queue.get(&t.id).unwrap();
4036
4037 let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
4038 assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
4039 assert_eq!(linked.status, before.status);
4040 assert_eq!(linked.attempts, before.attempts);
4041 assert_eq!(linked.interrupt, before.interrupt);
4042
4043 let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
4044 assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
4045 assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
4046 assert!(queue.link_run("no-such-task", "r").is_err());
4047 }
4048
4049 #[test]
4050 fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
4051 let dir = tempfile::tempdir().unwrap();
4052 let queue = Queue::at(dir.path().join("queue"));
4053 let mut t = Task::new(
4054 "t".to_owned(),
4055 "do it".to_owned(),
4056 PathBuf::from("."),
4057 Source::Human,
4058 );
4059 queue.put(&mut t).unwrap();
4060 let mut snapshot = queue.get(&t.id).unwrap();
4062 queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
4063
4064 snapshot.start("20260930-000000-aaaa".to_owned());
4065 queue.put(&mut snapshot).unwrap();
4066
4067 let stored = queue.get(&t.id).unwrap();
4068 assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
4069 assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
4070 assert_eq!(stored.attempts, 1);
4071 }
4072
4073 #[test]
4074 fn concurrent_links_and_daemon_saves_lose_nothing() {
4075 let dir = tempfile::tempdir().unwrap();
4076 let queue = Queue::at(dir.path().join("queue"));
4077 let mut t = Task::new(
4078 "t".to_owned(),
4079 "do it".to_owned(),
4080 PathBuf::from("."),
4081 Source::Human,
4082 );
4083 queue.put(&mut t).unwrap();
4084 let id = t.id.clone();
4085
4086 let linkers: Vec<_> = (0..4)
4087 .map(|n| {
4088 let (queue, id) = (queue.clone(), id.clone());
4089 std::thread::spawn(move || {
4090 for k in 0..10 {
4091 queue
4092 .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
4093 .unwrap();
4094 }
4095 })
4096 })
4097 .collect();
4098 let mut mine = queue.get(&id).unwrap();
4101 for k in 0..10 {
4102 mine.start(format!("20260930-000009-d{k:03}"));
4103 queue.put(&mut mine).unwrap();
4104 }
4105 for l in linkers {
4106 l.join().unwrap();
4107 }
4108
4109 let stored = queue.get(&id).unwrap();
4110 assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
4111 assert_eq!(
4112 stored.attempts, 10,
4113 "linking never rewinds the daemon's work"
4114 );
4115 }
4116
4117 fn age_lock(path: &Path) {
4118 let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
4119 f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
4120 .unwrap();
4121 }
4122
4123 #[test]
4124 fn concurrent_stale_takeover_yields_one_holder() {
4125 use std::sync::atomic::{AtomicUsize, Ordering};
4126 use std::sync::{Arc, Barrier};
4127 for _ in 0..5 {
4128 let dir = tempfile::tempdir().unwrap();
4129 let q = Queue::at(dir.path().to_path_buf());
4130 let lock = dir.path().join("t.write-lock");
4131 std::fs::write(&lock, "dead-0000").unwrap();
4132 age_lock(&lock);
4133 let n = 6;
4134 let barrier = Arc::new(Barrier::new(n));
4135 let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
4136 let handles: Vec<_> = (0..n)
4137 .map(|_| {
4138 let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
4139 std::thread::spawn(move || {
4140 b.wait();
4141 let g = q.lock_task("t").unwrap();
4142 let held = now.fetch_add(1, Ordering::SeqCst) + 1;
4143 max.fetch_max(held, Ordering::SeqCst);
4144 std::thread::sleep(std::time::Duration::from_millis(20));
4145 now.fetch_sub(1, Ordering::SeqCst);
4146 drop(g);
4147 })
4148 })
4149 .collect();
4150 for h in handles {
4151 h.join().unwrap();
4152 }
4153 assert_eq!(max.load(Ordering::SeqCst), 1);
4154 assert!(!lock.exists());
4155 }
4156 }
4157
4158 #[test]
4159 fn dropping_a_stolen_lock_leaves_the_new_holders_lock() {
4160 let dir = tempfile::tempdir().unwrap();
4161 let q = Queue::at(dir.path().to_path_buf());
4162 let lock = dir.path().join("t.write-lock");
4163 let a = q.lock_task("t").unwrap();
4164 age_lock(&lock);
4165 let b = q.lock_task("t").unwrap();
4166 assert_ne!(a.token, b.token);
4167 drop(a);
4168 assert_eq!(std::fs::read_to_string(&lock).unwrap(), b.token);
4169 drop(b);
4170 assert!(!lock.exists());
4171 }
4172
4173 #[test]
4174 fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
4175 let dir = tempfile::tempdir().unwrap();
4176 let lock = dir.path().join("t.write-lock");
4177 std::fs::write(&lock, "new-token").unwrap();
4178 assert!(!break_stale(&lock, "old-token"));
4179 assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
4180 assert!(!break_stale(&lock, "new-token"));
4182 assert!(lock.exists());
4183 age_lock(&lock);
4184 assert!(break_stale(&lock, "new-token"));
4185 assert!(!lock.exists());
4186 }
4187
4188 #[test]
4189 fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
4190 let dir = tempfile::tempdir().unwrap();
4191 let lock = dir.path().join("t.write-lock");
4192 std::fs::write(&lock, "dead-1").unwrap();
4193 age_lock(&lock);
4194 let marker = break_marker(&lock, "dead-1");
4195 std::fs::write(&marker, "crashed-remover").unwrap();
4196 assert!(!break_stale(&lock, "dead-1"));
4198 assert!(lock.exists());
4199 age_lock(&marker);
4201 assert!(break_stale(&lock, "dead-1"));
4202 assert!(!lock.exists());
4203 assert!(!marker.exists());
4204 }
4205
4206 fn chat_task() -> Task {
4207 Task::new(
4208 "t".to_owned(),
4209 "do t".to_owned(),
4210 PathBuf::from("."),
4211 Source::Agent {
4212 run: "20260902-000000-beef".to_owned(),
4213 node: CHAT_NODE.to_owned(),
4214 },
4215 )
4216 }
4217
4218 #[test]
4219 fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
4220 let mut t = chat_task();
4221 assert_eq!(t.chat_notice_due(), None, "queued");
4222 t.start("r1".to_owned());
4223 assert_eq!(t.chat_notice_due(), None, "running");
4224 t.fail("boom", 3);
4225 assert_eq!(t.status, TaskStatus::Failed);
4226 assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
4227 t.start("r2".to_owned());
4228 t.park("upgrade");
4229 assert_eq!(t.chat_notice_due(), None, "parked");
4230 t.fail("boom", 1);
4231 assert_eq!(t.status, TaskStatus::Held);
4232 let key = t
4233 .chat_notice_due()
4234 .expect("held on the last attempt is final");
4235 t.chat_notice = Some(ChatNotice {
4236 key,
4237 at: Timestamp::now(),
4238 outcome: ChatNoticeOutcome::Posted,
4239 });
4240 assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
4241 t.release();
4242 assert_eq!(t.chat_notice_due(), None, "requeued after a release");
4243 t.start("r3".to_owned());
4244 t.succeed();
4245 assert!(t.chat_notice_due().is_some(), "done is a new final state");
4246 }
4247
4248 #[test]
4249 fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
4250 for key in ["done", "done#0", "done#1", "done#12"] {
4251 let mut t = chat_task();
4252 t.start("r1".to_owned());
4253 t.succeed();
4254 t.chat_notice = Some(ChatNotice {
4255 key: key.to_owned(),
4256 at: Timestamp::now(),
4257 outcome: ChatNoticeOutcome::Posted,
4258 });
4259 t.requeue();
4260 t.start("r2".to_owned());
4261 t.succeed();
4262 assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
4263 assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
4264 }
4265 }
4266
4267 #[test]
4268 fn a_new_hold_after_a_release_is_a_new_cause() {
4269 let mut t = chat_task();
4270 t.hold_manual(Some("wait".to_owned()));
4271 let first = t.chat_notice_due().expect("held");
4272 t.chat_notice = Some(ChatNotice {
4273 key: first.clone(),
4274 at: Timestamp::now(),
4275 outcome: ChatNoticeOutcome::Posted,
4276 });
4277 t.hold_manual(Some("still waiting".to_owned()));
4278 assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
4279 t.release();
4280 std::thread::sleep(std::time::Duration::from_millis(5));
4281 t.hold_manual(None);
4282 let second = t.chat_notice_due().expect("a fresh hold");
4283 assert_ne!(first, second);
4284 }
4285
4286 #[test]
4287 fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
4288 let mut human = task("h");
4289 human.succeed();
4290 assert_eq!(human.chat_notice_due(), None);
4291 let mut inherited = task("i");
4293 inherited.origin_chat = Some("20260902-000000-beef".to_owned());
4294 inherited.succeed();
4295 assert_eq!(inherited.chat_notice_due(), None);
4296 let mut old = chat_task();
4297 old.schema = 12;
4298 old.status = TaskStatus::Done;
4299 assert_eq!(old.chat_notice_due(), None, "history is not replayed");
4300 let mut live = chat_task();
4301 live.schema = 12;
4302 live.start("r1".to_owned());
4303 live.succeed();
4304 assert!(live.chat_notice_due().is_some(), "started by this build");
4305 let mut held = chat_task();
4306 held.schema = 12;
4307 held.hold_machine(Some("config".to_owned()));
4308 assert!(held.chat_notice_due().is_some(), "held before any start");
4309 }
4310
4311 #[test]
4312 fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
4313 let tmp = tempfile::tempdir().expect("tempdir");
4314 let q = Queue::at(tmp.path().join("queue"));
4315 let mut t = chat_task();
4316 t.succeed();
4317 q.put(&mut t).expect("put");
4318 let mut stale = q.get(&t.id).expect("get");
4319 let key = t.chat_notice_due().expect("due");
4320 q.modify(&t.id, |x| {
4321 x.chat_notice = Some(ChatNotice {
4322 key: key.clone(),
4323 at: Timestamp::now(),
4324 outcome: ChatNoticeOutcome::Posted,
4325 });
4326 true
4327 })
4328 .expect("modify");
4329 q.put(&mut stale).expect("stale put");
4330 assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
4331 }
4332}