1use std::collections::HashMap;
34use std::num::NonZeroUsize;
35use std::path::{Path, PathBuf};
36
37use anyhow::{Context, Result, bail};
38use jiff::Timestamp;
39use serde::{Deserialize, Serialize};
40
41use crate::ask::Questions;
42use crate::id_reserve::{IdSource, path_exists};
43
44pub const SCHEMA: u32 = 20;
135
136pub(crate) fn dispatch_order(a: &Task, b: &Task) -> std::cmp::Ordering {
142 match (a.upgrade_resume_key(), b.upgrade_resume_key()) {
143 (Some(a), Some(b)) => a.cmp(&b),
144 (Some(_), None) => std::cmp::Ordering::Less,
145 (None, Some(_)) => std::cmp::Ordering::Greater,
146 (None, None) => b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)),
147 }
148}
149
150#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
152#[serde(rename_all = "lowercase")]
153pub enum HoldSource {
154 Manual,
156 Machine,
158}
159
160impl HoldSource {
161 pub fn label(self) -> &'static str {
163 match self {
164 Self::Manual => "manual",
165 Self::Machine => "machine",
166 }
167 }
168}
169
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
173#[serde(tag = "kind", rename_all = "lowercase")]
174pub enum Source {
175 Human,
177 Agent {
180 run: String,
182 node: String,
184 },
185 Issue {
187 number: u64,
189 repo: String,
191 },
192}
193
194pub const CHAT_NODE: &str = "chat";
197
198impl Source {
199 pub fn label(&self) -> String {
201 match self {
202 Self::Human => "human".to_owned(),
203 Self::Agent { run, node } => format!("{node}@{}", short(run)),
204 Self::Issue { number, .. } => format!("issue #{number}"),
205 }
206 }
207}
208
209#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
211#[serde(rename_all = "lowercase")]
212pub enum TaskStatus {
213 Queued,
215 Running,
217 Done,
219 Failed,
221 Held,
223 Blocked,
227 Parked,
232}
233
234impl TaskStatus {
235 pub fn runnable(self) -> bool {
237 matches!(self, Self::Queued | Self::Failed | Self::Parked)
238 }
239
240 pub fn as_str(self) -> &'static str {
242 match self {
243 Self::Queued => "queued",
244 Self::Running => "running",
245 Self::Done => "done",
246 Self::Failed => "failed",
247 Self::Held => "held",
248 Self::Blocked => "blocked",
249 Self::Parked => "parked",
250 }
251 }
252}
253
254#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
257pub struct TaskCounts {
258 pub queued: usize,
260 pub running: usize,
262 pub done: usize,
264 pub failed: usize,
266 pub held: usize,
268 pub blocked: usize,
270 pub parked: usize,
272}
273
274impl TaskCounts {
275 pub fn of(tasks: &[Task]) -> Self {
279 Self::of_statuses(tasks.iter().map(|task| task.status))
280 }
281
282 pub(crate) fn of_statuses(statuses: impl IntoIterator<Item = TaskStatus>) -> Self {
284 let mut counts = Self::default();
285 for status in statuses {
286 match status {
287 TaskStatus::Queued => counts.queued += 1,
288 TaskStatus::Running => counts.running += 1,
289 TaskStatus::Done => counts.done += 1,
290 TaskStatus::Failed => counts.failed += 1,
291 TaskStatus::Held => counts.held += 1,
292 TaskStatus::Blocked => counts.blocked += 1,
293 TaskStatus::Parked => counts.parked += 1,
294 }
295 }
296 counts
297 }
298}
299
300#[derive(Debug, Clone, Serialize, Deserialize)]
302#[serde(deny_unknown_fields)]
303pub struct Task {
304 pub schema: u32,
306 pub id: String,
308 pub title: String,
310 pub instruction: String,
312 #[serde(default, skip_serializing_if = "crate::refs::Sources::is_empty")]
314 pub sources: crate::refs::Sources,
315 pub repo: PathBuf,
317 pub source: Source,
319 #[serde(default)]
321 pub chat: Option<String>,
322 #[serde(default)]
324 pub priority: i32,
325 #[serde(default)]
336 pub solo: bool,
337 pub status: TaskStatus,
339 #[serde(default)]
341 pub attempts: usize,
342 #[serde(default)]
344 pub attempt_limit: Option<NonZeroUsize>,
345 #[serde(default)]
348 pub runs: Vec<String>,
349 #[serde(default)]
351 pub startup_failures: Vec<crate::run::StartupFailure>,
352 #[serde(default)]
354 pub last_error: Option<String>,
355 #[serde(default)]
368 pub hold_reason: Option<String>,
369 #[serde(default)]
372 pub hold_source: Option<HoldSource>,
373 #[serde(default)]
385 pub diagnostic: Option<String>,
386 #[serde(default)]
396 pub blocked_by: Vec<String>,
397 #[serde(default)]
400 pub block_reason: Option<String>,
401 #[serde(default)]
414 pub blocked_from: Option<TaskStatus>,
415 #[serde(default)]
417 pub blocked_at: Option<Timestamp>,
418 #[serde(default)]
429 pub answers: Vec<AnsweredQuestion>,
430 #[serde(default)]
436 pub triage_applied: Vec<String>,
437 #[serde(default)]
443 pub actions_applied: Vec<String>,
444 #[serde(default)]
449 pub resume_override: Option<OperatorResume>,
450 #[serde(default)]
453 pub resumption: Option<Resumption>,
454 #[serde(default)]
456 pub resumption_history: Vec<Resumption>,
457 #[serde(default)]
459 pub operator_decided_at: Option<Timestamp>,
460 #[serde(default)]
472 pub review_branch: Option<String>,
473 #[serde(default)]
475 pub review_instruction: Option<String>,
476 #[serde(default)]
478 pub pr_recovery: Option<crate::pr_recovery::Selection>,
479 #[serde(default)]
482 pub fresh_start: bool,
483 #[serde(default)]
485 pub replacement: Option<crate::merge_decision::Replacement>,
486 #[serde(default)]
497 pub interrupt: bool,
498 #[serde(default)]
521 pub urgent: bool,
522 #[serde(skip)]
527 pub(crate) unfiled: bool,
528 #[serde(default)]
534 pub attachments: Vec<String>,
535 #[serde(default)]
539 pub followup: Option<FollowUp>,
540 #[serde(default)]
546 pub origin_chat: Option<String>,
547 #[serde(default)]
554 pub overrides: Option<RunOverrides>,
555 #[serde(default)]
562 pub review_of: Option<String>,
563 #[serde(default)]
568 pub held_at: Option<Timestamp>,
569 #[serde(default)]
573 pub park_reason: Option<String>,
574 #[serde(default)]
577 pub chat_notice: Option<ChatNotice>,
578 #[serde(default)]
580 pub chat_notices: Vec<String>,
581 #[serde(default)]
584 pub hold_notice_cause: Option<NoticeCause>,
585 #[serde(default)]
590 pub passes: Vec<TaskPass>,
591 #[serde(default)]
595 pub dispatched_at: Option<Timestamp>,
596 #[serde(default)]
598 pub recovery_budget: Option<RecoveryBudget>,
599 #[serde(default)]
602 pub recovery_admission: Option<RecoveryAdmission>,
603 #[serde(default)]
606 pub recovery_generations: Vec<RecoveryBudget>,
607 pub created_at: Timestamp,
609 pub updated_at: Timestamp,
611}
612
613#[derive(Debug, Clone, Serialize, Deserialize)]
615pub struct RecoveryAdmission {
616 pub base: String,
618 pub inputs: String,
620}
621
622#[derive(Debug, Clone, Serialize, Deserialize)]
625pub struct RecoveryBudget {
626 pub instruction: String,
628 pub base: String,
630 pub inputs: String,
632 pub stage: String,
634 pub failures: usize,
636 #[serde(default)]
639 pub progress: Option<String>,
640}
641
642fn recovery_progress(state: &crate::run::RunState) -> Option<String> {
643 use crate::proc::Quiet as _;
644 use std::hash::{Hash, Hasher};
645 let mut patches = Vec::new();
646 for candidate in state.candidates.iter().filter(|c| c.commits > 0) {
647 let output = std::process::Command::new("git")
648 .current_dir(&state.repo)
649 .args(["diff", "--binary", &state.base_commit, &candidate.branch])
650 .quiet()
651 .output()
652 .ok()?;
653 if !output.status.success() {
654 return None;
655 }
656 patches.push(output.stdout);
657 }
658 patches.sort();
659 patches.dedup();
660 let mut hash = std::collections::hash_map::DefaultHasher::new();
661 patches.hash(&mut hash);
662 Some(format!("{:x}", hash.finish()))
663}
664
665pub(crate) fn recovery_inputs(sources: &crate::refs::Sources) -> String {
669 serde_json::to_string(
670 &sources
671 .start_from
672 .iter()
673 .map(|i| ("start", &i.token, &i.sha))
674 .chain(
675 sources
676 .pick_commits
677 .iter()
678 .map(|i| ("pick", &i.token, &i.sha)),
679 )
680 .collect::<Vec<_>>(),
681 )
682 .unwrap_or_default()
683}
684
685#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
687pub struct NoticeCause {
688 pub stage: String,
690 #[serde(default, skip_serializing_if = "Option::is_none")]
693 pub target: Option<String>,
694 pub run: Option<String>,
696 pub pass: Option<usize>,
698}
699
700pub(crate) fn is_done_notice_key(key: &str) -> bool {
702 key == "done"
703 || key
704 .strip_prefix("done#")
705 .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
706}
707
708#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
710#[serde(rename_all = "snake_case")]
711pub enum PassCost {
712 Spent,
714 Refunded,
716}
717
718#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
720pub struct TaskPass {
721 pub run: String,
723 pub pass_index: usize,
725 pub cost: PassCost,
727 pub attempts_after: usize,
729 #[serde(default)]
731 pub started_at: Option<Timestamp>,
732 pub settled_at: Timestamp,
734 #[serde(default)]
736 pub parked: Option<crate::run::ParkCause>,
737}
738
739#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
741#[serde(rename_all = "lowercase")]
742pub enum ChatNoticeOutcome {
743 Posted,
745 Skipped,
747}
748
749#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
751pub struct ChatNotice {
752 pub key: String,
754 pub at: Timestamp,
756 pub outcome: ChatNoticeOutcome,
758}
759
760#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
762pub struct RunOverrides {
763 #[serde(default)]
766 pub inherit_daemon_options: bool,
767 #[serde(default)]
769 pub merge: Option<String>,
770 #[serde(default)]
772 pub candidates: Option<usize>,
773 #[serde(default)]
775 pub judges: Option<usize>,
776 #[serde(default)]
778 pub reviewers: Option<usize>,
779 #[serde(default)]
781 pub review_rounds: Option<usize>,
782 #[serde(default)]
784 pub seed: Option<u64>,
785 #[serde(default)]
787 pub config: Option<PathBuf>,
788}
789
790impl RunOverrides {
791 pub fn merge_over(&mut self, other: &RunOverrides) {
793 macro_rules! take {
794 ($($f:ident),*) => {$(
795 if other.$f.is_some() {
796 self.$f.clone_from(&other.$f);
797 }
798 )*};
799 }
800 take!(
801 merge,
802 candidates,
803 judges,
804 reviewers,
805 review_rounds,
806 seed,
807 config
808 );
809 }
810
811 pub fn apply(&self, config: &mut crate::config::Config) {
813 if let Some(n) = self.candidates {
814 config.graph.implementers = n;
815 }
816 if let Some(n) = self.judges {
817 config.graph.judges = n;
818 }
819 if let Some(n) = self.reviewers {
820 config.graph.reviewers = n;
821 }
822 if let Some(n) = self.review_rounds {
823 config.graph.review_rounds = n;
824 }
825 if let Some(m) = &self.merge
826 && let Ok(mode) = crate::daemon::merge_mode(m)
827 {
828 config.merge.mode = mode;
829 }
830 if let Some(s) = self.seed {
831 config.blind.seed = Some(s);
832 }
833 }
834}
835
836#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
838pub struct FollowUp {
839 pub run: String,
841 #[serde(default)]
843 pub origin_task: Option<String>,
844 pub pr: String,
846 pub findings: Vec<String>,
848 pub generation: u32,
851}
852
853#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
855pub struct Resumption {
856 pub generation: u64,
858 pub source: String,
860 pub question_id: Option<String>,
862 pub at: Timestamp,
864 pub instruction: String,
866 pub answers_through: usize,
868 pub runs_through: usize,
871 pub cleared_reason: Option<String>,
873 #[serde(default)]
875 pub blocking_evidence: Vec<String>,
876}
877
878#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
881pub struct OperatorResume {
882 pub question_id: String,
884 pub at: Timestamp,
886 #[serde(default)]
889 pub conductor_rehold: Option<String>,
890 #[serde(default)]
893 pub forced: bool,
894 #[serde(default)]
901 pub pinned_run: Option<String>,
902}
903
904#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
907pub struct AnsweredQuestion {
908 pub question: String,
910 pub answer: String,
912}
913
914impl Task {
915 pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
917 let now = Timestamp::now();
918 let origin_chat = match &source {
919 Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
920 _ => None,
921 };
922 Self {
923 schema: SCHEMA,
924 id: new_id(),
925 title,
926 instruction,
927 sources: crate::refs::Sources::default(),
928 repo,
929 source,
930 chat: None,
931 priority: 0,
932 solo: false,
933 status: TaskStatus::Queued,
934 attempts: 0,
935 attempt_limit: None,
936 runs: Vec::new(),
937 startup_failures: Vec::new(),
938 last_error: None,
939 hold_reason: None,
940 hold_source: None,
941 diagnostic: None,
942 blocked_by: Vec::new(),
943 block_reason: None,
944 blocked_from: None,
945 blocked_at: None,
946 answers: Vec::new(),
947 triage_applied: Vec::new(),
948 actions_applied: Vec::new(),
949 resume_override: None,
950 resumption: None,
951 resumption_history: Vec::new(),
952 operator_decided_at: None,
953 review_branch: None,
954 review_instruction: None,
955 pr_recovery: None,
956 fresh_start: false,
957 replacement: None,
958 interrupt: false,
959 urgent: false,
960 unfiled: true,
961 attachments: Vec::new(),
962 followup: None,
963 origin_chat,
964 overrides: None,
965 review_of: None,
966 held_at: None,
967 park_reason: None,
968 chat_notice: None,
969 chat_notices: Vec::new(),
970 hold_notice_cause: None,
971 passes: Vec::new(),
972 dispatched_at: None,
973 recovery_budget: None,
974 recovery_admission: None,
975 recovery_generations: Vec::new(),
976 created_at: now,
977 updated_at: now,
978 }
979 }
980
981 pub fn chat_talk(&self) -> Option<&str> {
984 if let Some(id) = &self.origin_chat {
985 return Some(id);
986 }
987 match &self.source {
988 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
989 _ => None,
990 }
991 }
992
993 pub fn filed_by_chat(&self) -> Option<&str> {
997 match &self.source {
998 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
999 _ => None,
1000 }
1001 }
1002
1003 pub fn chat_notice_key(&self) -> Option<String> {
1006 match self.status {
1007 TaskStatus::Done => Some("done".to_owned()),
1008 TaskStatus::Held => {
1009 let cause = self
1010 .hold_notice_cause
1011 .clone()
1012 .unwrap_or_else(|| self.unresolved_hold_cause());
1013 Some(format!(
1014 "cause:{}",
1015 serde_json::to_string(&cause).expect("cause")
1016 ))
1017 }
1018 TaskStatus::Blocked => {
1019 let ids: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
1020 let evidence = self
1023 .hold_notice_cause
1024 .as_ref()
1025 .filter(|c| c.stage.starts_with("conductor-evidence:"));
1026 Some(format!(
1027 "dependencies:{}",
1028 serde_json::to_string(&(ids, evidence)).expect("ids")
1029 ))
1030 }
1031 _ => None,
1032 }
1033 }
1034
1035 fn unresolved_hold_cause(&self) -> NoticeCause {
1036 NoticeCause {
1037 stage: "hold".into(),
1038 target: None,
1039 run: self.runs.last().cloned(),
1040 pass: None,
1041 }
1042 }
1043
1044 pub(crate) fn legacy_chat_notice_key(&self) -> Option<String> {
1046 let key = match self.status {
1047 TaskStatus::Held if self.hold_notice_cause.is_none() => Some(format!(
1048 "held#{}",
1049 self.held_at.map(|t| t.to_string()).unwrap_or_default()
1050 )),
1051 TaskStatus::Blocked => Some(format!(
1052 "blocked#{}#{}",
1053 self.blocked_by.join(","),
1054 self.block_reason.as_deref().unwrap_or_default()
1055 )),
1056 _ => None,
1057 };
1058 key.filter(|key| self.legacy_notice_matches(key))
1059 }
1060
1061 pub(crate) fn legacy_notice_matches(&self, key: &str) -> bool {
1062 match self.status {
1063 TaskStatus::Held if self.hold_notice_cause.is_none() => key
1064 .strip_prefix("held#")
1065 .is_some_and(|time| time.is_empty() || time.parse::<Timestamp>().is_ok()),
1066 TaskStatus::Blocked
1067 if self
1068 .hold_notice_cause
1069 .as_ref()
1070 .is_none_or(|c| !c.stage.starts_with("conductor-evidence:")) =>
1071 {
1072 key.strip_prefix("blocked#").is_some_and(|tail| {
1073 let Some((ids, _)) = tail.split_once('#') else {
1074 return false;
1075 };
1076 let old: std::collections::BTreeSet<_> =
1077 ids.split(',').filter(|id| !id.is_empty()).collect();
1078 let current: std::collections::BTreeSet<_> =
1079 self.blocked_by.iter().map(String::as_str).collect();
1080 old == current
1081 })
1082 }
1083 _ => false,
1084 }
1085 }
1086
1087 fn import_chat_notice(&mut self) {
1088 if let Some(n) = &self.chat_notice {
1089 let key = if is_done_notice_key(&n.key) {
1090 "done".to_owned()
1091 } else if self.legacy_notice_matches(&n.key) {
1092 self.chat_notice_key().unwrap_or_else(|| n.key.clone())
1093 } else {
1094 n.key.clone()
1095 };
1096 if !self.chat_notices.contains(&key) {
1097 self.chat_notices.push(key);
1098 }
1099 }
1100 }
1101
1102 pub(crate) fn notice_cause(&mut self, stage: &str, run: Option<&str>, pass: Option<usize>) {
1105 self.import_chat_notice();
1106 self.hold_notice_cause = Some(NoticeCause {
1107 stage: stage.to_owned(),
1108 target: None,
1109 run: run.map(str::to_owned),
1110 pass,
1111 });
1112 }
1113
1114 pub(crate) fn startup_notice_cause(&mut self, stage: &str) {
1117 let recovery_only = matches!(
1118 stage,
1119 "startup-source" | "startup-candidates" | "startup-public-head"
1120 );
1121 match self.hold_notice_cause.as_ref() {
1122 Some(cause) if recovery_only && !cause.stage.starts_with("startup-") => return,
1123 None if recovery_only
1124 && self
1125 .chat_notice
1126 .as_ref()
1127 .is_some_and(|n| !is_done_notice_key(&n.key)) =>
1128 {
1129 return;
1130 }
1131 _ => {}
1132 }
1133 self.notice_cause(stage, None, None);
1134 }
1135
1136 pub(crate) fn fail_startup(
1139 &mut self,
1140 stage: &str,
1141 why: impl Into<String>,
1142 max_attempts: usize,
1143 ) {
1144 let unresolved = self.hold_notice_cause.clone();
1145 self.fail(why, max_attempts);
1146 self.hold_notice_cause = unresolved;
1147 self.startup_notice_cause(stage);
1148 }
1149
1150 pub(crate) fn fail_startup_target(
1152 &mut self,
1153 stage: &str,
1154 target: String,
1155 why: impl Into<String>,
1156 max_attempts: usize,
1157 ) {
1158 self.fail_startup(stage, why, max_attempts);
1159 if let Some(cause) = self.hold_notice_cause.as_mut() {
1160 cause.target = Some(target);
1161 }
1162 }
1163
1164 pub fn chat_notice_due(&self) -> Option<String> {
1166 self.filed_by_chat()?;
1167 if self.schema < 13 {
1168 return None;
1169 }
1170 let key = self.chat_notice_key()?;
1171 if self.chat_notices.contains(&key) {
1172 return None;
1173 }
1174 match &self.chat_notice {
1175 Some(n)
1176 if n.key == key
1177 || (key == "done" && is_done_notice_key(&n.key))
1178 || self.legacy_notice_matches(&n.key) =>
1179 {
1180 None
1181 }
1182 _ => Some(key),
1183 }
1184 }
1185
1186 pub fn short(&self) -> &str {
1188 short(&self.id)
1189 }
1190
1191 pub fn mark_triage_applied(&mut self, question_id: &str) {
1193 if !self.triage_applied(question_id) {
1194 self.triage_applied.push(question_id.to_owned());
1195 }
1196 }
1197
1198 pub fn action_applied(&self, question_id: &str) -> bool {
1201 self.actions_applied.iter().any(|id| id == question_id)
1202 }
1203
1204 pub fn mark_action_applied(&mut self, question_id: &str) {
1206 if !self.action_applied(question_id) {
1207 self.actions_applied.push(question_id.to_owned());
1208 }
1209 }
1210
1211 pub fn triage_applied(&self, question_id: &str) -> bool {
1213 self.triage_applied.iter().any(|id| id == question_id)
1214 }
1215
1216 pub fn start(&mut self, run: String) {
1229 self.stamp_schema();
1230 self.status = TaskStatus::Running;
1231 self.attempts += 1;
1232 self.runs.retain(|id| id != &run);
1233 self.runs.push(run);
1234 self.last_error = None;
1235 self.park_reason = None;
1236 self.fresh_start = false;
1237 self.interrupt = false;
1238 self.resume_override = None;
1240 }
1241
1242 pub fn record_startup_failure(&mut self, failure: crate::run::StartupFailure) {
1244 if !self.startup_failures.iter().any(|f| f.run == failure.run) {
1245 self.startup_failures.push(failure);
1246 }
1247 }
1248
1249 pub fn link_run(&mut self, run: &str) -> bool {
1254 if self.runs.iter().any(|r| r == run) {
1255 return false;
1256 }
1257 self.runs.push(run.to_owned());
1258 true
1259 }
1260
1261 pub fn succeed(&mut self) {
1271 self.stamp_schema();
1272 self.status = TaskStatus::Done;
1273 self.resume_override = None;
1274 self.last_error = None;
1275 self.park_reason = None;
1276 self.hold_reason = None;
1277 self.hold_source = None;
1278 self.diagnostic = None;
1279 self.blocked_by.clear();
1280 self.block_reason = None;
1281 self.blocked_from = None;
1282 }
1283
1284 pub fn already_landed(&mut self, note: impl Into<String>) {
1292 self.succeed();
1293 self.attempts = self.attempts.saturating_sub(1);
1294 self.last_error = Some(note.into());
1295 }
1296
1297 pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
1327 if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
1328 return &[];
1329 }
1330 &self.runs[..self.runs.len() - 1]
1331 }
1332
1333 pub fn successor_of(&self, run: &str) -> Option<&String> {
1341 let pos = self.runs.iter().rposition(|r| r == run)?;
1342 self.runs.get(pos + 1)
1343 }
1344
1345 pub fn earlier_attempts(&self) -> &[String] {
1352 &self.runs
1353 }
1354
1355 fn stamp_schema(&mut self) {
1362 self.schema = self.schema.max(SCHEMA);
1363 }
1364
1365 fn note_held(&mut self) {
1366 self.stamp_schema();
1367 let continuing = self.status == TaskStatus::Held
1368 || (self.status == TaskStatus::Blocked
1369 && (self.blocked_from == Some(TaskStatus::Held)
1370 || (self.blocked_from.is_none()
1371 && (self.hold_reason.is_some() || self.hold_source.is_some()))));
1372 if !continuing || self.held_at.is_none() {
1373 self.held_at = Some(Timestamp::now());
1374 }
1375 }
1376
1377 pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
1393 let why = why.into();
1394 if self.status == TaskStatus::Running {
1395 let run = self.runs.last().cloned();
1396 self.notice_cause("run-failed", run.as_deref(), Some(self.attempts));
1397 }
1398 self.diagnostic = None;
1399 self.park_reason = None;
1400 self.status = if self.attempts >= self.effective_attempt_limit(max_attempts) {
1401 self.note_held();
1402 self.hold_source = Some(HoldSource::Machine);
1403 self.hold_reason = Some(why.clone());
1404 TaskStatus::Held
1405 } else {
1406 TaskStatus::Failed
1407 };
1408 self.last_error = Some(why);
1409 }
1410
1411 pub fn stall(&mut self, why: impl Into<String>) {
1421 self.last_error = Some(why.into());
1422 self.diagnostic = None;
1423 self.attempts = self.attempts.saturating_sub(1);
1424 self.status = TaskStatus::Failed;
1425 }
1426
1427 pub fn park(&mut self, why: impl Into<String>) {
1433 self.diagnostic = None;
1434 self.attempts = self.attempts.saturating_sub(1);
1435 self.status = TaskStatus::Parked;
1436 self.park_reason = Some(why.into());
1437 self.schema = self.schema.max(SCHEMA);
1440 }
1441
1442 pub fn record_pass(
1444 &mut self,
1445 run: &str,
1446 pass_index: usize,
1447 attempts_before: usize,
1448 parked: Option<crate::run::ParkCause>,
1449 ) {
1450 self.record_pass_at(run, pass_index, None, attempts_before, parked);
1451 }
1452
1453 pub fn record_pass_at(
1456 &mut self,
1457 run: &str,
1458 pass_index: usize,
1459 started_at: Option<Timestamp>,
1460 attempts_before: usize,
1461 parked: Option<crate::run::ParkCause>,
1462 ) {
1463 if self
1464 .passes
1465 .iter()
1466 .any(|p| p.run == run && p.pass_index == pass_index)
1467 {
1468 return;
1469 }
1470 let cost = if self.attempts < attempts_before {
1471 PassCost::Refunded
1472 } else {
1473 PassCost::Spent
1474 };
1475 self.passes.push(TaskPass {
1476 run: run.to_owned(),
1477 pass_index,
1478 started_at: self.dispatched_at.or(started_at),
1479 cost,
1480 attempts_after: self.attempts,
1481 settled_at: Timestamp::now(),
1482 parked,
1483 });
1484 self.schema = self.schema.max(SCHEMA);
1485 }
1486
1487 pub(crate) fn upgrade_resume_key(&self) -> Option<(Timestamp, &str)> {
1490 if self.status != TaskStatus::Parked || self.fresh_start {
1491 return None;
1492 }
1493 let pass = self.passes.last()?;
1494 if self.runs.last() != Some(&pass.run)
1495 || !matches!(pass.parked, Some(crate::run::ParkCause::Upgrade { .. }))
1496 {
1497 return None;
1498 }
1499 Some((pass.started_at.unwrap_or(pass.settled_at), &self.id))
1502 }
1503
1504 pub fn operator_held(&self) -> bool {
1510 self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
1511 }
1512
1513 pub fn hold_manual(&mut self, reason: Option<String>) {
1523 self.operator_decided_at = Some(Timestamp::now());
1525 self.hold_unrequested_retry(reason);
1526 }
1527
1528 pub(crate) fn hold_unrequested_retry(&mut self, reason: Option<String>) {
1531 self.park_reason = None;
1532 self.note_held();
1533 self.status = TaskStatus::Held;
1534 if reason.is_some() {
1535 self.hold_reason = reason;
1536 }
1537 self.hold_source = Some(HoldSource::Manual);
1538 self.blocked_by.clear();
1539 self.block_reason = None;
1540 self.blocked_from = None;
1541 }
1542
1543 pub fn hold_machine(&mut self, reason: Option<String>) {
1548 self.park_reason = None;
1549 self.note_held();
1550 self.status = TaskStatus::Held;
1551 if reason.is_some() {
1552 self.hold_reason = reason;
1553 }
1554 self.hold_source = Some(HoldSource::Machine);
1555 self.blocked_by.clear();
1556 self.block_reason = None;
1557 self.blocked_from = None;
1558 }
1559
1560 pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
1570 self.stamp_schema();
1571 let before: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
1572 let after: std::collections::BTreeSet<_> = blocked_by.iter().collect();
1573 if self.status != TaskStatus::Blocked || before != after {
1574 self.blocked_at = Some(Timestamp::now());
1575 }
1576 if self.status != TaskStatus::Blocked {
1577 self.blocked_from = Some(self.status);
1578 }
1579 self.status = TaskStatus::Blocked;
1580 self.blocked_by = blocked_by;
1581 self.block_reason = reason;
1582 }
1583
1584 pub fn unblock(&mut self, resolved_id: &str) {
1607 if self.status != TaskStatus::Blocked {
1608 return;
1609 }
1610 self.blocked_by.retain(|id| id != resolved_id);
1611 self.restore_if_unblocked();
1612 }
1613
1614 pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
1625 if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
1626 return false;
1627 }
1628 self.blocked_by.retain(|id| id != deleted_id);
1629 self.restore_if_unblocked();
1630 true
1631 }
1632
1633 fn restore_if_unblocked(&mut self) {
1637 if self.blocked_by.is_empty() {
1638 self.status = match self.blocked_from {
1639 Some(TaskStatus::Running) => TaskStatus::Queued,
1640 Some(other) => other,
1641 None if self.hold_reason.is_some() || self.hold_source.is_some() => {
1642 TaskStatus::Held
1643 }
1644 None => TaskStatus::Queued,
1645 };
1646 self.block_reason = None;
1647 self.blocked_from = None;
1648 }
1649 }
1650
1651 pub fn record_answer(&mut self, question: String, answer: String) {
1656 self.answers.push(AnsweredQuestion { question, answer });
1657 }
1658
1659 pub fn request_review(&mut self, branch: String) {
1663 self.release();
1664 self.review_branch = Some(branch);
1665 self.review_instruction = Some(self.instruction.clone());
1666 }
1667
1668 pub fn requeue(&mut self) {
1671 self.release();
1672 self.review_branch = None;
1674 self.review_instruction = None;
1675 self.fresh_start = true;
1676 }
1677
1678 pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
1687 if branch.is_some() {
1688 self.review_branch = branch;
1689 self.review_instruction = Some(self.instruction.clone());
1690 }
1691 self.hold_machine(Some(reason));
1692 }
1693
1694 pub fn set_urgent(&mut self, urgent: bool) -> Result<()> {
1696 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1697 bail!(
1698 "task {} is {}; urgent can only change while queued or held; \
1699 wait for the active attempt to finish and requeue or hold the task first",
1700 self.short(),
1701 self.status.as_str()
1702 );
1703 }
1704 self.urgent = urgent;
1705 Ok(())
1706 }
1707
1708 pub fn set_priority(&mut self, priority: i32) -> Result<()> {
1717 if self.status == TaskStatus::Running {
1718 bail!(
1719 "task {} is running; its priority cannot be changed until \
1720 this attempt finishes",
1721 self.short()
1722 );
1723 }
1724 self.priority = priority;
1725 Ok(())
1726 }
1727
1728 pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
1743 if interrupt && !self.status.runnable() {
1744 bail!(
1745 "task {} is {}; only a queued or failed task can be marked \
1746 to interrupt",
1747 self.short(),
1748 self.status.as_str()
1749 );
1750 }
1751 self.interrupt = interrupt;
1752 Ok(())
1753 }
1754
1755 pub fn effective_attempt_limit(&self, daemon_limit: usize) -> usize {
1757 self.attempt_limit.map_or(daemon_limit, NonZeroUsize::get)
1758 }
1759
1760 pub fn change_attempt_limit(&mut self, limit: Option<NonZeroUsize>) -> Result<()> {
1763 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1764 bail!("only queued or held tasks can change the attempt limit");
1765 }
1766 self.stamp_schema();
1767 self.attempt_limit = limit;
1768 Ok(())
1769 }
1770
1771 pub fn implementers(&self) -> Option<usize> {
1773 if self.solo {
1774 Some(1)
1775 } else {
1776 self.overrides.as_ref().and_then(|o| o.candidates)
1777 }
1778 }
1779
1780 pub fn change_implementers(&mut self, count: u8) -> Result<()> {
1783 self.check_implementers_editable()?;
1784 let explicit = self.overrides.as_ref().and_then(|o| o.config.as_deref());
1785 let (mut config, _) = crate::config::Config::discover(&self.repo, explicit)?;
1786 if let Some(o) = &self.overrides {
1787 o.apply(&mut config);
1788 }
1789 self.set_implementers(count, &config)
1790 }
1791
1792 fn check_implementers_editable(&self) -> Result<()> {
1793 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1794 bail!(
1795 "task {} is {}; only queued or held tasks can change implementers",
1796 self.short(),
1797 self.status.as_str()
1798 );
1799 }
1800 Ok(())
1801 }
1802
1803 pub fn set_implementers(&mut self, count: u8, config: &crate::config::Config) -> Result<()> {
1805 self.check_implementers_editable()?;
1806 crate::talk::check_implementers(count, config).map_err(anyhow::Error::msg)?;
1807 self.stamp_schema();
1808 self.solo = count == 1;
1809 let mut overrides = self.overrides.clone().unwrap_or_else(|| RunOverrides {
1810 inherit_daemon_options: true,
1811 ..Default::default()
1812 });
1813 overrides.candidates = (count > 1).then_some(usize::from(count));
1814 let mut settings = overrides.clone();
1815 settings.inherit_daemon_options = false;
1816 self.overrides = if overrides.inherit_daemon_options && settings == RunOverrides::default()
1818 {
1819 None
1820 } else {
1821 Some(overrides)
1822 };
1823 Ok(())
1824 }
1825
1826 pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
1838 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1839 bail!(
1840 "task {} is {}; only a queued or held task's instruction can \
1841 be edited",
1842 self.short(),
1843 self.status.as_str()
1844 );
1845 }
1846 if self.review_branch.is_some() && self.review_instruction.is_none() {
1849 self.review_instruction = Some(self.instruction.clone());
1850 }
1851 if self.instruction != instruction {
1852 self.resume_override = None;
1855 }
1856 self.title = title;
1857 self.instruction = instruction;
1858 Ok(())
1859 }
1860
1861 pub fn handed_off(&mut self, why: impl Into<String>) {
1881 let why = why.into();
1882 self.diagnostic = None;
1883 self.note_held();
1884 self.status = TaskStatus::Held;
1885 self.hold_source = Some(HoldSource::Machine);
1886 self.hold_reason = Some(why.clone());
1887 self.last_error = Some(why);
1888 }
1889
1890 pub fn release_operator(&mut self, source: &str, question_id: Option<&str>) {
1892 self.recovery_budget = None;
1893 self.recovery_generations.clear();
1894 self.recovery_admission = None;
1895 if self
1896 .resumption
1897 .as_ref()
1898 .is_some_and(|r| r.instruction != self.instruction)
1899 {
1900 self.resume_override = None;
1903 }
1904 let approval = Resumption {
1905 generation: self.resumption.as_ref().map_or(1, |r| r.generation + 1),
1906 source: source.to_owned(),
1907 question_id: question_id.map(str::to_owned),
1908 at: Timestamp::now(),
1909 instruction: self.instruction.clone(),
1910 answers_through: self.answers.len(),
1911 runs_through: self.runs.len(),
1912 blocking_evidence: Vec::new(),
1913 cleared_reason: self
1914 .hold_reason
1915 .clone()
1916 .or_else(|| self.block_reason.clone()),
1917 };
1918 self.release();
1919 self.operator_decided_at = Some(approval.at);
1920 if let Some(previous) = self.resumption.replace(approval) {
1921 self.resumption_history.push(previous);
1922 }
1923 }
1924
1925 pub(crate) fn recovery_failed(&mut self, state: &crate::run::RunState) {
1927 let (base, inputs) = self.recovery_admission.as_ref().map_or_else(
1928 || (state.base_commit.clone(), recovery_inputs(&state.sources)),
1929 |a| (a.base.clone(), a.inputs.clone()),
1930 );
1931 let progress = recovery_progress(state);
1932 let stage = if state.prep_failure.is_some() {
1933 "preparation"
1934 } else {
1935 "execution"
1936 };
1937 let matches = |b: &RecoveryBudget| {
1938 b.instruction == self.instruction
1939 && b.base == base
1940 && b.inputs == inputs
1941 && (progress.is_none() || b.progress == progress)
1942 };
1943 if !self.recovery_budget.as_ref().is_some_and(matches) {
1944 let previous = self
1945 .recovery_generations
1946 .iter()
1947 .position(matches)
1948 .map(|index| self.recovery_generations.remove(index));
1949 if let Some(old) = self.recovery_budget.take() {
1950 self.recovery_generations.push(old);
1951 }
1952 self.recovery_budget = previous;
1953 }
1954 if let Some(budget) = &mut self.recovery_budget {
1955 budget.failures = budget.failures.saturating_add(1);
1956 budget.stage = stage.to_owned();
1957 } else {
1958 self.recovery_budget = Some(RecoveryBudget {
1959 instruction: self.instruction.clone(),
1960 base,
1961 inputs,
1962 stage: stage.to_owned(),
1963 failures: 1,
1964 progress,
1965 });
1966 }
1967 }
1968
1969 pub(crate) fn recovery_exhausted(
1970 &self,
1971 base: &str,
1972 inputs: &str,
1973 limit: usize,
1974 ) -> Option<String> {
1975 let same_input = |b: &&RecoveryBudget| {
1976 b.instruction == self.instruction && b.base == base && b.inputs == inputs
1977 };
1978 let budget = self
1979 .recovery_budget
1980 .as_ref()
1981 .filter(same_input)
1982 .or_else(|| self.recovery_generations.iter().rev().find(same_input))?;
1983 (budget.instruction == self.instruction && budget.base == base && budget.inputs == inputs
1984 && budget.failures >= limit.max(1)).then(|| format!(
1985 "[recovery-budget] {} failed passes against unchanged inputs ({}). Inspect the retained recovery sources and correct the cause, change the inputs, or explicitly release the task to approve another generation. Automatic requeue does not reset this budget.",
1986 budget.failures, budget.stage))
1987 }
1988
1989 pub fn release(&mut self) {
1993 self.import_chat_notice();
1996 if self.hold_notice_cause.is_none()
1997 && matches!(self.status, TaskStatus::Held | TaskStatus::Blocked)
1998 {
1999 self.hold_notice_cause = Some(self.unresolved_hold_cause());
2000 }
2001 let refused_handover = self.status == TaskStatus::Held
2002 && self.hold_source == Some(HoldSource::Machine)
2003 && self.review_branch.is_some()
2004 && self
2005 .review_instruction
2006 .as_ref()
2007 .is_none_or(|i| *i == self.instruction);
2008 self.status = TaskStatus::Queued;
2009 self.held_at = None;
2010 self.attempts = 0;
2011 self.last_error = None;
2012 self.park_reason = None;
2013 self.hold_reason = None;
2016 self.hold_source = None;
2017 self.diagnostic = None;
2018 self.blocked_by.clear();
2023 self.block_reason = None;
2024 self.blocked_from = None;
2025 if !refused_handover {
2029 self.review_branch = None;
2030 self.review_instruction = None;
2031 }
2032 self.fresh_start = false;
2033 }
2034}
2035
2036fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
2040 use std::io::ErrorKind;
2041 let (stem, ext) = match name.rfind('.') {
2042 Some(i) if i > 0 => (&name[..i], &name[i..]),
2043 _ => (name, ""),
2044 };
2045 for n in 1u32.. {
2046 let candidate = if n == 1 {
2047 name.to_owned()
2048 } else {
2049 let suffix = format!("-{n}");
2050 let room = 64usize.saturating_sub(suffix.len() + ext.len());
2051 let stem: String = stem.chars().take(room).collect();
2052 format!("{stem}{suffix}{ext}")
2053 };
2054 if !crate::ask::valid_asset_name(&candidate) {
2055 bail!("no valid attachment name is left for `{name}`");
2056 }
2057 let path = dir.join(&candidate);
2058 match std::fs::OpenOptions::new()
2059 .write(true)
2060 .create_new(true)
2061 .open(&path)
2062 {
2063 Ok(mut out) => {
2064 let copied = std::fs::File::open(src)
2065 .and_then(|mut input| std::io::copy(&mut input, &mut out));
2066 if let Err(e) = copied {
2067 drop(out);
2068 let _ = std::fs::remove_file(&path);
2069 return Err(e).with_context(|| format!("copy {}", src.display()));
2070 }
2071 return Ok((candidate, path));
2072 }
2073 Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
2074 Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
2075 }
2076 }
2077 unreachable!("the counter never runs out")
2078}
2079
2080const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
2083
2084pub(crate) struct TaskLock {
2087 path: PathBuf,
2088 token: String,
2089}
2090
2091fn owner_token() -> String {
2094 static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
2095 let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2096 let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
2097 format!("{}-{:016x}", std::process::id(), r.next_u64())
2098}
2099
2100fn break_marker(path: &Path, token: &str) -> PathBuf {
2103 let mut h: u64 = 0xcbf29ce484222325;
2104 for b in token.bytes() {
2105 h ^= u64::from(b);
2106 h = h.wrapping_mul(0x100000001b3);
2107 }
2108 let mut name = path.as_os_str().to_owned();
2109 name.push(format!(".break-{h:016x}"));
2110 PathBuf::from(name)
2111}
2112
2113fn older_than_stale(path: &Path) -> bool {
2114 std::fs::metadata(path)
2115 .and_then(|m| m.modified())
2116 .ok()
2117 .and_then(|t| t.elapsed().ok())
2118 .is_some_and(|age| age > TASK_LOCK_STALE)
2119}
2120
2121fn writer_may_be_gone(token: &str) -> bool {
2122 let Some(pid) = token
2123 .split_once('-')
2124 .and_then(|(pid, _)| pid.parse::<u32>().ok())
2125 else {
2126 return true;
2129 };
2130 if pid == std::process::id() {
2131 return false;
2132 }
2133 crate::proc::pid_status(pid) == Some(false)
2135}
2136
2137const MAX_MARKER_DEPTH: u8 = 3;
2141
2142fn take_marker(
2148 path: &Path,
2149 token: &str,
2150 depth: u8,
2151 retry: crate::lock_file::Retry,
2152) -> Option<(PathBuf, String)> {
2153 use std::io::Write;
2154 let marker = break_marker(path, token);
2155 for _ in 0..2 {
2156 match retry.create(&marker) {
2157 Ok(mut f) => {
2158 let mine = owner_token();
2159 if f.write_all(mine.as_bytes()).is_err() {
2160 drop(f);
2161 let _ = std::fs::remove_file(&marker);
2162 return None;
2163 }
2164 return Some((marker, mine));
2165 }
2166 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2167 let judged = std::fs::read_to_string(&marker).ok();
2168 match judged.filter(|_| older_than_stale(&marker)) {
2169 Some(judged) if depth < MAX_MARKER_DEPTH => {
2170 remove_lock_if(&marker, &judged, true, depth + 1, retry)?;
2171 }
2172 _ => return None,
2173 }
2174 }
2175 Err(_) => return None,
2176 }
2177 }
2178 None
2179}
2180
2181fn remove_lock_if(
2186 path: &Path,
2187 token: &str,
2188 require_stale: bool,
2189 depth: u8,
2190 retry: crate::lock_file::Retry,
2191) -> Option<bool> {
2192 let (marker, mine) = take_marker(path, token, depth, retry)?;
2193 let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
2194 && (!require_stale || older_than_stale(path) && writer_may_be_gone(token));
2195 if still {
2196 let _ = std::fs::remove_file(path);
2197 }
2198 if depth >= MAX_MARKER_DEPTH {
2200 if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
2201 let _ = std::fs::remove_file(&marker);
2202 }
2203 } else {
2204 let _ = remove_lock_if(&marker, &mine, false, depth + 1, retry);
2205 }
2206 Some(still)
2207}
2208
2209#[cfg(test)]
2212fn break_stale(path: &Path, judged: &str) -> bool {
2213 break_stale_with(path, judged, crate::lock_file::Retry::new(None))
2214}
2215
2216fn break_stale_with(path: &Path, judged: &str, retry: crate::lock_file::Retry) -> bool {
2217 remove_lock_if(path, judged, true, 0, retry).unwrap_or(false)
2218}
2219
2220impl Drop for TaskLock {
2221 fn drop(&mut self) {
2222 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
2223 let retry = crate::lock_file::Retry::new(Some(deadline));
2224 loop {
2225 match remove_lock_if(&self.path, &self.token, false, 0, retry) {
2226 Some(_) => return,
2227 None if std::time::Instant::now() > deadline => return,
2228 None => std::thread::sleep(std::time::Duration::from_millis(5)),
2229 }
2230 }
2231 }
2232}
2233
2234#[derive(Debug, Clone)]
2236pub struct Queue {
2237 root: PathBuf,
2238 ids: IdSource,
2239}
2240
2241fn validate_attachments(sources: &[PathBuf]) -> Result<Vec<(&PathBuf, &str)>> {
2242 let mut wanted = Vec::new();
2243 for src in sources {
2244 let name = src
2245 .file_name()
2246 .and_then(|n| n.to_str())
2247 .with_context(|| format!("`{}` has no usable file name", src.display()))?;
2248 if !crate::ask::valid_asset_name(name) {
2249 bail!(
2250 "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
2251 with no `..`; rename the file and try again"
2252 );
2253 }
2254 if !src.is_file() {
2255 bail!("attachment `{}` is not a file", src.display());
2256 }
2257 wanted.push((src, name));
2258 }
2259 Ok(wanted)
2260}
2261
2262impl Queue {
2263 pub fn open() -> Self {
2265 Self::at(crate::run::home().join("queue"))
2266 }
2267
2268 pub fn at(root: PathBuf) -> Self {
2271 Self {
2272 root,
2273 ids: IdSource::new(new_id),
2274 }
2275 }
2276
2277 #[doc(hidden)]
2280 pub fn with_id_source(mut self, f: impl Fn() -> String + Send + Sync + 'static) -> Self {
2281 self.ids = IdSource::custom(f);
2282 self
2283 }
2284
2285 fn file_with(
2296 &self,
2297 task: &mut Task,
2298 hook: impl FnMut(&mut Task) -> Result<bool>,
2299 undo: impl FnMut(&mut Task),
2300 ) -> Result<()> {
2301 self.reserve_id(task, hook, undo, true)
2302 }
2303
2304 fn reserve_id(
2312 &self,
2313 task: &mut Task,
2314 mut hook: impl FnMut(&mut Task) -> Result<bool>,
2315 mut undo: impl FnMut(&mut Task),
2316 store: bool,
2317 ) -> Result<()> {
2318 let redraw_first = self.ids.is_custom() && task.attachments.is_empty();
2319 let initial = task.id.clone();
2320 self.ids
2321 .reserve((!redraw_first).then_some(initial.as_str()), "task", |id| {
2322 task.id = id.to_owned();
2323 let _lock = self.lock_task(&task.id)?;
2324 if path_exists(&self.path_of(&task.id))? || !hook(task)? {
2325 return Ok(false);
2326 }
2327 if store {
2328 if let Err(e) = self.write_record(task) {
2329 undo(task);
2330 return Err(e);
2331 }
2332 task.unfiled = false;
2333 }
2334 Ok(true)
2335 })?;
2336 Ok(())
2337 }
2338
2339 fn id_in_use(&self, id: &str, check_attachments: bool) -> Result<bool> {
2344 if path_exists(&self.lock_path(id))? {
2345 return Ok(true);
2346 }
2347 Ok(check_attachments && path_exists(&self.attachments_dir(id))?)
2348 }
2349
2350 fn plain_filing_hook(&self, task: &Task) -> Result<bool> {
2354 Ok(!self.id_in_use(&task.id, task.attachments.is_empty())?)
2355 }
2356
2357 pub fn file_claimed(&self, task: &mut Task) -> Result<Claim> {
2360 let mut claim = None;
2361 self.file_with(
2362 task,
2363 |t| {
2364 if self.id_in_use(&t.id, true)? {
2365 return Ok(false);
2366 }
2367 claim = Some(self.claim(&t.id)?);
2368 Ok(true)
2369 },
2370 |_| {},
2371 )?;
2372 claim.context("claim missing after filing")
2373 }
2374
2375 pub fn root(&self) -> &Path {
2377 &self.root
2378 }
2379
2380 pub fn path_of(&self, id: &str) -> PathBuf {
2382 self.root.join(format!("{id}.json"))
2383 }
2384
2385 pub fn attachments_dir(&self, id: &str) -> PathBuf {
2388 self.root.join(format!("{id}.attachments"))
2389 }
2390
2391 pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2402 let wanted = validate_attachments(sources)?;
2403 if task.unfiled && task.attachments.is_empty() {
2404 let names = std::cell::RefCell::new(Vec::new());
2407 self.reserve_id(
2408 task,
2409 |t| {
2410 if self.id_in_use(&t.id, true)? {
2411 return Ok(false);
2412 }
2413 *names.borrow_mut() = self.copy_attachments(t, &wanted)?;
2414 Ok(true)
2415 },
2416 |_| {},
2417 false,
2418 )?;
2419 return Ok(names.into_inner());
2420 }
2421 self.copy_attachments(task, &wanted)
2422 }
2423
2424 fn copy_attachments(
2426 &self,
2427 task: &mut Task,
2428 wanted: &[(&PathBuf, &str)],
2429 ) -> Result<Vec<String>> {
2430 let dir = self.attachments_dir(&task.id);
2431 let existed = dir.is_dir();
2432 std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
2433 let mut created: Vec<PathBuf> = Vec::new();
2434 let mut names = Vec::new();
2435 let mut copy_all = || -> Result<()> {
2436 for (src, name) in wanted {
2437 let (stored, path) = copy_new(&dir, src, name)?;
2438 created.push(path);
2439 names.push(stored);
2440 }
2441 Ok(())
2442 };
2443 if let Err(e) = copy_all() {
2444 for path in &created {
2445 let _ = std::fs::remove_file(path);
2446 }
2447 if !existed {
2448 let _ = std::fs::remove_dir(&dir);
2449 }
2450 return Err(e);
2451 }
2452 task.attachments.extend(names.iter().cloned());
2453 Ok(names)
2454 }
2455
2456 pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2467 if task.unfiled && task.attachments.is_empty() {
2468 return self.attach_and_file(task, sources);
2469 }
2470 let dir = self.attachments_dir(&task.id);
2471 let existed = dir.is_dir();
2472 let before = task.attachments.len();
2473 let names = self.attach(task, sources)?;
2474 if let Err(e) = self.put(task) {
2475 for name in &names {
2476 let _ = std::fs::remove_file(dir.join(name));
2477 }
2478 if !existed {
2479 let _ = std::fs::remove_dir(&dir);
2480 }
2481 task.attachments.truncate(before);
2482 return Err(e);
2483 }
2484 Ok(names)
2485 }
2486
2487 fn attach_and_file(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2491 let wanted = validate_attachments(sources)?;
2492 let before = task.attachments.len();
2493 let names = std::cell::RefCell::new(Vec::new());
2494 let result = self.file_with(
2495 task,
2496 |t| {
2497 if self.id_in_use(&t.id, true)? {
2498 return Ok(false);
2499 }
2500 *names.borrow_mut() = self.copy_attachments(t, &wanted)?;
2501 Ok(true)
2502 },
2503 |t| {
2504 let dir = self.attachments_dir(&t.id);
2505 for name in names.borrow().iter() {
2506 let _ = std::fs::remove_file(dir.join(name));
2507 }
2508 let _ = std::fs::remove_dir(&dir);
2510 t.attachments.truncate(before);
2511 },
2512 );
2513 result.map(|()| names.into_inner())
2514 }
2515
2516 pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
2520 let dir = self.attachments_dir(&task.id);
2521 task.attachments
2522 .iter()
2523 .map(|n| {
2524 let p = dir.join(n);
2525 std::path::absolute(&p).unwrap_or(p)
2526 })
2527 .collect()
2528 }
2529
2530 pub fn put(&self, task: &mut Task) -> Result<()> {
2539 if task.unfiled {
2540 return self.file_with(task, |t| self.plain_filing_hook(t), |_| {});
2542 }
2543 let _lock = self.lock_task(&task.id)?;
2544 self.put_unlocked(task)
2545 }
2546
2547 pub fn create_new(&self, task: &mut Task) -> Result<bool> {
2552 let _lock = self.lock_task(&task.id)?;
2553 if path_exists(&self.path_of(&task.id))? {
2554 return Ok(false);
2555 }
2556 self.write_record(task)?;
2557 task.unfiled = false;
2558 Ok(true)
2559 }
2560
2561 fn put_unlocked(&self, task: &mut Task) -> Result<()> {
2563 self.merge_stored(task);
2564 self.write_record(task)
2565 }
2566
2567 fn merge_stored(&self, task: &mut Task) {
2569 if let Ok(mut stored) = read_path(&self.path_of(&task.id)) {
2570 stored.import_chat_notice();
2571 if task.chat.is_none() {
2574 task.chat = stored.chat.clone();
2575 }
2576 for key in stored.chat_notices {
2577 if !task.chat_notices.contains(&key) {
2578 task.chat_notices.push(key);
2579 }
2580 }
2581 for failure in stored.startup_failures {
2582 task.record_startup_failure(failure);
2583 }
2584 for run in stored.runs {
2585 if !task.runs.contains(&run) {
2586 task.runs.push(run);
2587 }
2588 }
2589 for p in stored.passes {
2590 if !task
2591 .passes
2592 .iter()
2593 .any(|q| q.run == p.run && q.pass_index == p.pass_index)
2594 {
2595 task.passes.push(p);
2596 }
2597 }
2598 if let Some(n) = stored.chat_notice
2601 && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
2602 {
2603 task.chat_notice = Some(n);
2604 }
2605 }
2606 }
2607
2608 pub(crate) fn write_startup_history(&self, task: &Task) -> Result<()> {
2611 let path = self.path_of(&task.id);
2612 let tmp = path.with_extension("json.tmp");
2613 std::fs::write(&tmp, serde_json::to_vec_pretty(task)?)?;
2614 std::fs::rename(tmp, path)?;
2615 Ok(())
2616 }
2617
2618 fn write_record(&self, task: &mut Task) -> Result<()> {
2620 task.import_chat_notice();
2621 task.updated_at = Timestamp::now();
2622 std::fs::create_dir_all(&self.root)
2623 .with_context(|| format!("create {}", self.root.display()))?;
2624 let body = serde_json::to_string_pretty(task).context("serialize task")?;
2625 let path = self.path_of(&task.id);
2626 let tmp = path.with_extension("json.tmp");
2627 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
2628 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
2629 if let (Some(notice), Some(home)) = (
2634 crate::notices::task_held(task),
2635 self.root.parent().filter(|p| !p.as_os_str().is_empty()),
2636 ) {
2637 crate::notices::raise_in(home, notice);
2638 }
2639 Ok(())
2640 }
2641
2642 pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
2647 let id = self.resolve_id(id)?;
2648 let _lock = self.lock_task(&id)?;
2649 let mut task = self.get(&id)?;
2650 if !f(&mut task) {
2651 return Ok(false);
2652 }
2653 self.put_unlocked(&mut task)?;
2654 Ok(true)
2655 }
2656
2657 pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
2662 let id = self.resolve_id(id)?;
2663 let _lock = self.lock_task(&id)?;
2666 let mut task = self.get(&id)?;
2667 if task.link_run(run) {
2668 self.put_unlocked(&mut task)?;
2669 }
2670 Ok(task)
2671 }
2672
2673 pub(crate) fn lock_task(&self, id: &str) -> Result<TaskLock> {
2681 std::fs::create_dir_all(&self.root)
2682 .with_context(|| format!("create {}", self.root.display()))?;
2683 let path = self.root.join(format!("{id}.write-lock"));
2684 let started = std::time::Instant::now();
2685 let retry = crate::lock_file::Retry::task_write(started + TASK_LOCK_STALE);
2686 let marker_retry = crate::lock_file::Retry::new(Some(started + TASK_LOCK_STALE));
2687 loop {
2688 match retry.create(&path) {
2689 Ok(mut file) => {
2690 use std::io::Write;
2691 let token = owner_token();
2692 if let Err(e) = file.write_all(token.as_bytes()) {
2693 drop(file);
2694 let _ = std::fs::remove_file(&path);
2695 return Err(e).with_context(|| format!("lock {}", path.display()));
2696 }
2697 return Ok(TaskLock { path, token });
2698 }
2699 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2700 let judged = std::fs::read_to_string(&path).ok();
2703 let broken = judged
2704 .filter(|_| older_than_stale(&path))
2705 .is_some_and(|judged| break_stale_with(&path, &judged, marker_retry));
2706 if !broken {
2710 if started.elapsed() > TASK_LOCK_STALE {
2711 bail!("could not lock task {id}");
2712 }
2713 std::thread::sleep(std::time::Duration::from_millis(15));
2714 }
2715 }
2716 Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
2717 }
2718 }
2719 }
2720
2721 pub fn get(&self, id: &str) -> Result<Task> {
2723 let resolved = self.resolve_id(id)?;
2724 read_path(&self.path_of(&resolved))
2725 }
2726
2727 pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
2759 let resolved = self.resolve_id(id)?;
2760 if in_flight {
2761 bail!("task {resolved} is being run by a live daemon right now");
2762 }
2763 self.write_tombstone(&resolved)?;
2764 if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
2765 {
2766 if self.path_of(&resolved).exists() {
2770 let _ = std::fs::remove_file(self.tombstone_path(&resolved));
2771 }
2772 return Err(e);
2773 }
2774 let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
2775 Ok(Removal {
2776 id: resolved,
2777 released,
2778 still_blocked,
2779 })
2780 }
2781
2782 fn tombstone_path(&self, id: &str) -> PathBuf {
2785 self.root.join(format!("{id}.removed"))
2786 }
2787
2788 fn write_tombstone(&self, id: &str) -> Result<()> {
2789 let path = self.tombstone_path(id);
2790 let tmp = self.root.join(format!("{id}.removed.tmp"));
2791 std::fs::write(&tmp, b"")
2792 .and_then(|()| std::fs::rename(&tmp, &path))
2793 .with_context(|| format!("write {}", path.display()))
2794 }
2795
2796 fn was_deleted(&self, id: &str) -> bool {
2800 self.tombstone_path(id).is_file() && !self.path_of(id).exists()
2801 }
2802
2803 pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
2807 let deleted = deleted_blockers(self, questions, &task.blocked_by);
2808 deleted
2809 .into_iter()
2810 .filter(|id| task.dependency_deleted(id))
2811 .collect()
2812 }
2813
2814 pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
2818 let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
2819 return;
2820 };
2821 let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
2822 let waiting = dependent.status == TaskStatus::Blocked;
2823 let message = match (ja, waiting) {
2824 (true, false) => format!(
2825 "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
2826 dependent.short(),
2827 short(deleted)
2828 ),
2829 (true, true) => format!(
2830 "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
2831 dependent.short(),
2832 short(deleted)
2833 ),
2834 (false, false) => format!(
2835 "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
2836 dependent.short(),
2837 short(deleted)
2838 ),
2839 (false, true) => format!(
2840 "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
2841 dependent.short(),
2842 short(deleted)
2843 ),
2844 };
2845 crate::notices::raise_in(
2846 home,
2847 crate::notices::Notice::info(
2848 &format!("unblocked:{}:{}", dependent.id, deleted),
2849 message,
2850 )
2851 .link(crate::notices::Link::Task {
2852 id: dependent.id.clone(),
2853 }),
2854 );
2855 }
2856
2857 fn release_dependents_of(
2862 &self,
2863 dependency: &str,
2864 questions: &Questions,
2865 ) -> (Vec<String>, Vec<String>) {
2866 let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
2867 for listed in self.list() {
2868 if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
2869 matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
2870 }) {
2871 continue;
2872 }
2873 let Ok(_claim) = self.claim(&listed.id) else {
2874 continue;
2875 };
2876 let Ok(mut task) = self.get(&listed.id) else {
2877 continue;
2878 };
2879 let references: Vec<String> = task.blocked_by.iter().filter(|id| {
2880 matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
2881 }).cloned().collect();
2882 if references.is_empty() {
2883 continue;
2884 }
2885 for reference in references {
2886 task.dependency_deleted(&reference);
2887 }
2888 if self.put(&mut task).is_err() {
2889 continue;
2890 }
2891 self.note_dependency_deleted(&task, dependency);
2892 if task.status == TaskStatus::Blocked {
2893 still_blocked.push(task.id.clone());
2894 } else {
2895 released.push(task.id.clone());
2896 }
2897 }
2898 (released, still_blocked)
2899 }
2900
2901 fn remove_record_with_attachments(
2905 &self,
2906 resolved: &str,
2907 remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
2908 ) -> Result<()> {
2909 self.sweep_removed_attachments();
2916 let attachments = self.attachments_dir(resolved);
2917 let aside = self.root.join(format!("{resolved}.attachments.removing"));
2918 let moved = match std::fs::rename(&attachments, &aside) {
2919 Ok(()) => true,
2920 Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
2921 Err(e) => {
2922 return Err(e).with_context(|| format!("remove {}", attachments.display()));
2923 }
2924 };
2925 let path = self.path_of(resolved);
2926 if let Err(e) = remove_record(&path) {
2927 if moved {
2928 let _ = std::fs::rename(&aside, &attachments);
2929 }
2930 return Err(e).with_context(|| format!("remove {}", path.display()));
2931 }
2932 if moved {
2933 if let Err(e) = std::fs::remove_dir_all(&aside) {
2934 tracing::warn!("leftover attachments {}: {e}", aside.display());
2935 }
2936 }
2937 let lock = self.lock_path(resolved);
2938 if let Err(e) = std::fs::remove_file(&lock) {
2939 if e.kind() != std::io::ErrorKind::NotFound {
2940 return Err(e).with_context(|| format!("remove {}", lock.display()));
2941 }
2942 }
2943 Ok(())
2944 }
2945
2946 fn sweep_removed_attachments(&self) {
2952 let Ok(entries) = std::fs::read_dir(&self.root) else {
2953 return;
2954 };
2955 for entry in entries.flatten() {
2956 let name = entry.file_name();
2957 let name = name.to_string_lossy();
2958 let Some(id) = name.strip_suffix(".attachments.removing") else {
2959 continue;
2960 };
2961 if !self.path_of(id).exists() {
2964 if let Err(e) = std::fs::remove_dir_all(entry.path()) {
2965 tracing::warn!("leftover attachments {}: {e}", entry.path().display());
2966 }
2967 }
2968 }
2969 self.sweep_tombstones();
2970 }
2971
2972 fn sweep_tombstones(&self) {
2981 let Ok(entries) = std::fs::read_dir(&self.root) else {
2982 return;
2983 };
2984 let tasks = self.list();
2985 for entry in entries.flatten() {
2986 let name = entry.file_name();
2987 let name = name.to_string_lossy();
2988 let Some(id) = name.strip_suffix(".removed") else {
2989 continue;
2990 };
2991 if self.path_of(id).exists()
2992 || tasks.iter().any(|t| {
2993 t.blocked_by
2994 .iter()
2995 .any(|b| id.starts_with(b) || id.ends_with(b))
2996 })
2997 {
2998 continue;
2999 }
3000 let old = entry
3001 .metadata()
3002 .and_then(|m| m.modified())
3003 .ok()
3004 .and_then(|m| m.elapsed().ok())
3005 .is_some_and(|age| age >= TOMBSTONE_GRACE);
3006 if old {
3007 let _ = std::fs::remove_file(entry.path());
3008 }
3009 }
3010 }
3011
3012 fn lock_path(&self, id: &str) -> PathBuf {
3015 self.root.join(format!("{id}.lock"))
3016 }
3017
3018 pub fn list(&self) -> Vec<Task> {
3031 let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
3032 .into_iter()
3033 .flatten()
3034 .flatten()
3035 .map(|e| e.path())
3036 .filter(|p| p.extension().is_some_and(|x| x == "json"))
3037 .filter_map(|p| read_path(&p).ok())
3038 .collect();
3039 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
3040 tasks
3041 }
3042
3043 pub fn unreadable(&self) -> Vec<PathBuf> {
3047 std::fs::read_dir(&self.root)
3048 .into_iter()
3049 .flatten()
3050 .flatten()
3051 .map(|e| e.path())
3052 .filter(|p| p.extension().is_some_and(|x| x == "json"))
3053 .filter(|p| read_path(p).is_err())
3054 .collect()
3055 }
3056
3057 pub fn superseded(&self) -> HashMap<String, String> {
3070 let mut by = HashMap::new();
3071 for task in self.list() {
3072 for earlier in &task.runs {
3073 if let Some(later) = task.successor_of(earlier) {
3074 by.insert(earlier.clone(), later.clone());
3075 }
3076 }
3077 }
3078 by
3079 }
3080
3081 pub fn superseded_by(&self, run: &str) -> Option<String> {
3089 for task in self.list() {
3090 if task.runs.iter().any(|r| r == run) {
3091 return task.successor_of(run).cloned();
3092 }
3093 }
3094 None
3095 }
3096
3097 pub fn latest_attempt(&self, run: &str) -> Option<String> {
3108 for task in self.list() {
3109 if task.runs.iter().any(|r| r == run) {
3110 return task.runs.last().filter(|last| **last != run).cloned();
3111 }
3112 }
3113 None
3114 }
3115
3116 pub fn next_runnable(&self) -> Option<Task> {
3121 let mut runnable: Vec<Task> = self
3122 .list()
3123 .into_iter()
3124 .filter(|t| t.status.runnable())
3125 .collect();
3126 runnable.sort_unstable_by(dispatch_order);
3127 runnable.into_iter().next()
3128 }
3129
3130 pub fn claim(&self, id: &str) -> Result<Claim> {
3137 std::fs::create_dir_all(&self.root)
3138 .with_context(|| format!("create {}", self.root.display()))?;
3139 let path = self.lock_path(id);
3140 let retry = crate::lock_file::Retry::new(None);
3141 match retry.create(&path) {
3142 Ok(mut f) => {
3143 use std::io::Write as _;
3144 let _ = writeln!(f, "{}", std::process::id());
3146 Ok(Claim { path })
3147 }
3148 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
3149 bail!("task {id} is already claimed ({} exists)", path.display())
3150 }
3151 Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
3152 }
3153 }
3154
3155 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
3157 if !prefix.is_empty()
3158 && prefix != "."
3159 && prefix != ".."
3160 && !prefix.contains(['/', '\\'])
3161 && self.path_of(prefix).is_file()
3162 {
3163 return Ok(prefix.to_owned());
3164 }
3165 let ids = record_ids(&self.root, "json")?;
3166 match resolve_reference(prefix, ids.iter().map(String::as_str)) {
3167 Reference::Found(id) => Ok(id),
3168 Reference::Missing => bail!("no task matches `{prefix}`"),
3169 Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
3170 }
3171 }
3172
3173 pub fn revision(&self) -> u64 {
3180 self.revision_excluding(&std::collections::BTreeSet::new())
3181 }
3182
3183 pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
3186 use std::hash::{Hash as _, Hasher as _};
3187
3188 let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
3189 .into_iter()
3190 .flatten()
3191 .flatten()
3192 .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
3193 .filter(|e| {
3194 let path = e.path();
3195 !path
3196 .file_stem()
3197 .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
3198 })
3199 .filter_map(|e| {
3200 let name = e.file_name().to_string_lossy().into_owned();
3201 let mtime = e
3202 .metadata()
3203 .ok()?
3204 .modified()
3205 .ok()?
3206 .duration_since(std::time::UNIX_EPOCH)
3207 .ok()?
3208 .as_millis() as u64;
3209 Some((name, mtime))
3210 })
3211 .collect();
3212
3213 if entries.is_empty() {
3214 return 0;
3215 }
3216
3217 entries.sort_unstable();
3218 let mut hasher = std::hash::DefaultHasher::new();
3219 for (name, mtime) in &entries {
3220 name.hash(&mut hasher);
3221 mtime.hash(&mut hasher);
3222 }
3223 let h = hasher.finish();
3224 if h == 0 { 1 } else { h }
3225 }
3226}
3227
3228const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
3231
3232#[derive(Debug, Clone)]
3234pub struct Removal {
3235 pub id: String,
3237 pub released: Vec<String>,
3241 pub still_blocked: Vec<String>,
3244}
3245
3246#[derive(Debug)]
3248pub struct Claim {
3249 path: PathBuf,
3250}
3251
3252impl Drop for Claim {
3253 fn drop(&mut self) {
3254 let _ = std::fs::remove_file(&self.path);
3255 }
3256}
3257
3258pub fn title_from(instruction: &str, max: usize) -> String {
3261 let Some(line) = first_line(instruction) else {
3262 return "(empty task)".to_owned();
3263 };
3264 if line.chars().count() <= max {
3265 return line.to_owned();
3266 }
3267 let head: String = line.chars().take(max.saturating_sub(1)).collect();
3268 format!("{head}…")
3269}
3270
3271pub(crate) fn first_line(instruction: &str) -> Option<&str> {
3277 let line = instruction
3278 .lines()
3279 .map(str::trim)
3280 .find(|l| !l.is_empty())?
3281 .trim_start_matches(['#', '-', '*', '>', ' '])
3282 .trim();
3283 (!line.is_empty()).then_some(line)
3284}
3285
3286fn read_path(path: &Path) -> Result<Task> {
3287 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
3288 let task: Task =
3289 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
3290 if task.schema > SCHEMA {
3296 bail!(
3297 "task {} was written by a different magi (schema {}, this build \
3298 speaks {SCHEMA})",
3299 task.id,
3300 task.schema
3301 );
3302 }
3303 Ok(task)
3304}
3305
3306#[derive(Debug, PartialEq, Eq)]
3309pub(crate) enum Reference {
3310 Found(String),
3311 Missing,
3312 Ambiguous,
3313}
3314
3315pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
3316 let mut hits = Vec::new();
3317 for candidate in ids {
3318 if candidate == id {
3319 return Reference::Found(candidate.to_owned());
3320 }
3321 if candidate.starts_with(id) || candidate.ends_with(id) {
3322 hits.push(candidate);
3323 }
3324 }
3325 match hits.as_slice() {
3326 [] => Reference::Missing,
3327 [one] => Reference::Found((*one).to_owned()),
3328 _ => Reference::Ambiguous,
3329 }
3330}
3331
3332pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
3335 let entries = match std::fs::read_dir(root) {
3336 Ok(entries) => entries,
3337 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
3338 Err(e) => return Err(e.into()),
3339 };
3340 let mut ids = Vec::new();
3341 for entry in entries {
3342 let path = entry?.path();
3343 if path.extension().is_some_and(|e| e == extension)
3344 && let Some(id) = path.file_stem().and_then(|s| s.to_str())
3345 {
3346 ids.push(id.to_owned());
3347 }
3348 }
3349 Ok(ids)
3350}
3351
3352#[derive(Debug, PartialEq, Eq)]
3355pub enum Dependency {
3356 Task(String),
3358 Question(String),
3360 Deleted(String),
3362 Missing,
3364 Ambiguous,
3366 Unknown,
3368}
3369
3370pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
3372 let (Ok(tasks), Ok(qs), Ok(removed)) = (
3373 record_ids(queue.root(), "json"),
3374 record_ids(questions.root(), "json"),
3375 record_ids(queue.root(), "removed"),
3376 ) else {
3377 return Dependency::Unknown;
3378 };
3379 let mut ids = tasks
3382 .iter()
3383 .chain(qs.iter())
3384 .chain(removed.iter())
3385 .cloned()
3386 .collect::<Vec<_>>();
3387 ids.sort();
3388 ids.dedup();
3389 match resolve_reference(id, ids.iter().map(String::as_str)) {
3390 Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
3391 Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
3392 Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
3393 Reference::Found(_) => Dependency::Unknown,
3394 Reference::Missing => Dependency::Missing,
3395 Reference::Ambiguous => Dependency::Ambiguous,
3396 }
3397}
3398
3399pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
3401 for id in ids {
3402 if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
3403 dependency(queue, questions, id)
3404 {
3405 *id = full;
3406 }
3407 }
3408}
3409
3410pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
3413 let inventory = crate::blockers::Inventory::load(queue, questions);
3414 let mut text = String::from(
3415 "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
3416 );
3417 match record_ids(queue.root(), "json") {
3418 Ok(ids) => {
3419 for id in ids {
3420 match queue.get(&id) {
3421 Ok(task) => text.push_str(&format!(
3422 "{}: {}; blocked_by={:?}; chain={:?}\n",
3423 task.id,
3424 task.status.as_str(),
3425 task.blocked_by,
3426 inventory.waits_on(&task)
3427 )),
3428 Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
3429 }
3430 }
3431 }
3432 Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
3433 }
3434 text
3435}
3436
3437pub fn missing_blockers(
3439 queue: &Queue,
3440 questions: &Questions,
3441 blocked_by: &[String],
3442) -> Vec<String> {
3443 blocked_by
3444 .iter()
3445 .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
3446 .cloned()
3447 .collect()
3448}
3449
3450pub fn deleted_blockers(
3453 queue: &Queue,
3454 questions: &Questions,
3455 blocked_by: &[String],
3456) -> Vec<String> {
3457 blocked_by
3458 .iter()
3459 .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
3460 .cloned()
3461 .collect()
3462}
3463
3464pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
3476 missing_blocker_hold_reason_in(blocked_by, missing, "en")
3477}
3478
3479pub fn missing_blocker_hold_reason_in(
3481 blocked_by: &[String],
3482 missing: &[String],
3483 language: &str,
3484) -> String {
3485 if crate::lang::is_japanese(language) {
3486 format!(
3487 "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
3488 blocked_by.join(", "),
3489 missing.join(", "),
3490 )
3491 } else {
3492 format!(
3493 "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
3494 blocked_by.join(", "),
3495 missing.join(", "),
3496 )
3497 }
3498}
3499
3500pub fn short(id: &str) -> &str {
3502 id.split('-').next_back().unwrap_or(id)
3503}
3504
3505fn new_id() -> String {
3506 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
3507 let seed = crate::rng::entropy();
3508 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
3509}
3510
3511#[cfg(test)]
3512mod tests {
3513 #[test]
3514 fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
3515 let t = Task::new(
3516 "retry".into(),
3517 "do it".into(),
3518 PathBuf::from("."),
3519 Source::Human,
3520 );
3521 let mut value = serde_json::to_value(&t).unwrap();
3522 for field in ["resumption", "resumption_history", "operator_decided_at"] {
3523 value.as_object_mut().unwrap().remove(field);
3524 }
3525 value["schema"] = serde_json::json!(15);
3526 let mut legacy: Task = serde_json::from_value(value).unwrap();
3527 legacy.hold_machine(Some("automatic retry".into()));
3528 legacy.release();
3529 assert!(legacy.resumption.is_none());
3530 assert!(legacy.resumption_history.is_empty());
3531 assert!(legacy.operator_decided_at.is_none());
3532 legacy.release_operator("cli", None);
3533 legacy.start("run".into());
3534 let reloaded: Task =
3535 serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
3536 assert!(reloaded.resume_override.is_none());
3537 assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
3538 }
3539
3540 #[test]
3541 fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
3542 let mut base = RunOverrides {
3543 merge: Some("pr".to_owned()),
3544 candidates: Some(3),
3545 ..RunOverrides::default()
3546 };
3547 base.merge_over(&RunOverrides {
3548 merge: Some("none".to_owned()),
3549 seed: Some(7),
3550 ..RunOverrides::default()
3551 });
3552 assert_eq!(base.merge.as_deref(), Some("none"));
3553 assert_eq!(base.candidates, Some(3));
3554 assert_eq!(base.seed, Some(7));
3555 }
3556
3557 #[test]
3558 fn an_already_landed_task_is_done_with_its_attempt_refunded() {
3559 let mut t = task("relanded");
3560 t.attempts = 1;
3561 t.status = TaskStatus::Running;
3562 t.already_landed("already in main as 0e368de");
3563 assert_eq!(t.status, TaskStatus::Done);
3564 assert_eq!(t.attempts, 0);
3565 assert!(t.hold_reason.is_none());
3566 assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
3567 }
3568
3569 #[test]
3570 fn missing_blocker_reason_follows_the_language() {
3571 let b = vec!["a".to_owned()];
3572 let en = missing_blocker_hold_reason_in(&b, &b, "en");
3573 assert_eq!(en, missing_blocker_hold_reason(&b, &b));
3574 assert!(en.starts_with("blocked on a"));
3575 assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
3576 assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
3577 }
3578
3579 use super::*;
3580
3581 #[test]
3582 fn triage_applied_survives_release_and_old_records_read_as_empty() {
3583 let mut t = Task::new(
3584 "t".to_owned(),
3585 "i".to_owned(),
3586 PathBuf::from("r"),
3587 Source::Human,
3588 );
3589 t.mark_triage_applied("q1");
3590 t.mark_triage_applied("q1");
3591 t.hold_machine(Some("x".to_owned()));
3592 t.release();
3593 assert_eq!(t.triage_applied, ["q1"]);
3594 assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
3595
3596 let mut v = serde_json::to_value(&t).unwrap();
3597 v.as_object_mut().unwrap().remove("triage_applied");
3598 let old: Task = serde_json::from_value(v).unwrap();
3599 assert!(old.triage_applied.is_empty());
3600 }
3601
3602 #[test]
3603 fn task_counts_of_empty_is_all_zero() {
3604 assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
3605 }
3606
3607 #[test]
3608 fn task_counts_of_tallies_every_status() {
3609 let mut queued = Task::new(
3610 "q".to_owned(),
3611 "i".to_owned(),
3612 PathBuf::from("."),
3613 Source::Human,
3614 );
3615 queued.status = TaskStatus::Queued;
3616 let mut running = queued.clone();
3617 running.status = TaskStatus::Running;
3618 let mut done = queued.clone();
3619 done.status = TaskStatus::Done;
3620 let mut failed = queued.clone();
3621 failed.status = TaskStatus::Failed;
3622 let mut held = queued.clone();
3623 held.status = TaskStatus::Held;
3624 let mut blocked = queued.clone();
3625 blocked.status = TaskStatus::Blocked;
3626
3627 let mut parked = queued.clone();
3628 parked.status = TaskStatus::Parked;
3629
3630 let counts = TaskCounts::of(&[
3631 queued,
3632 running,
3633 done.clone(),
3634 done,
3635 failed,
3636 held,
3637 blocked,
3638 parked,
3639 ]);
3640 assert_eq!(
3641 counts,
3642 TaskCounts {
3643 queued: 1,
3644 running: 1,
3645 done: 2,
3646 failed: 1,
3647 held: 1,
3648 blocked: 1,
3649 parked: 1,
3650 }
3651 );
3652 }
3653
3654 #[test]
3655 fn task_implementers_persist_normalized_without_losing_history() {
3656 let (_dir, q) = queue();
3657 let mut t = task("keep me");
3658 t.hold_manual(Some("waiting".into()));
3659 t.runs.push("earlier-run".into());
3660 t.attachments.push("reference.txt".into());
3661 let cfg: crate::config::Config =
3662 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
3663 .unwrap();
3664 q.put(&mut t).unwrap();
3665 let original = serde_json::to_value(&t).unwrap();
3666 for n in [2, 1, 2] {
3667 t.set_implementers(n, &cfg).unwrap();
3668 q.put(&mut t).unwrap();
3669 t = q.get(&t.id).unwrap();
3670 assert_eq!(t.implementers(), Some(usize::from(n)));
3671 assert_eq!(t.solo, n == 1);
3672 if n == 1 {
3673 assert!(t.overrides.is_none());
3674 }
3675 let mut actual = serde_json::to_value(&t).unwrap();
3676 for field in ["solo", "overrides", "updated_at"] {
3677 actual[field] = original[field].clone();
3678 }
3679 assert_eq!(actual, original);
3680 }
3681 t.overrides.as_mut().unwrap().judges = Some(2);
3682 t.set_implementers(1, &cfg).unwrap();
3683 assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
3684 assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
3685 }
3686
3687 #[test]
3688 fn task_implementers_old_records_keep_inheritance_until_changed() {
3689 let mut value = serde_json::to_value(task("old task")).unwrap();
3690 value.as_object_mut().unwrap().remove("solo");
3691 value.as_object_mut().unwrap().remove("overrides");
3692 let mut old: Task = serde_json::from_value(value).unwrap();
3693 assert_eq!(old.implementers(), None);
3694 old.set_implementers(1, &crate::config::Config::default())
3695 .unwrap();
3696 assert!(old.solo);
3697 assert!(old.overrides.is_none());
3698 }
3699
3700 #[test]
3701 fn task_implementers_reject_invalid_counts_and_states() {
3702 let cfg = crate::config::Config::default();
3703 let mut t = task("unchanged");
3704 for n in [0, 4, 255] {
3705 assert!(t.set_implementers(n, &cfg).is_err());
3706 }
3707 for status in [
3708 TaskStatus::Running,
3709 TaskStatus::Done,
3710 TaskStatus::Failed,
3711 TaskStatus::Blocked,
3712 TaskStatus::Parked,
3713 ] {
3714 t.status = status;
3715 let before = serde_json::to_value(&t).unwrap();
3716 assert!(t.set_implementers(1, &cfg).is_err());
3717 assert_eq!(serde_json::to_value(&t).unwrap(), before);
3718 }
3719 }
3720
3721 fn queue() -> (tempfile::TempDir, Queue) {
3724 let dir = tempfile::tempdir().unwrap();
3725 let q = Queue::at(dir.path().join("queue"));
3726 (dir, q)
3727 }
3728
3729 #[test]
3730 fn a_stale_write_does_not_drop_the_recorded_task_chat() {
3731 let (_dir, q) = queue();
3732 let mut stale = task("chat");
3733 q.put(&mut stale).unwrap();
3734 let mut fresh = q.get(&stale.id).unwrap();
3735 fresh.chat = Some("talk-1".to_owned());
3736 q.put(&mut fresh).unwrap();
3737 q.put(&mut stale).unwrap();
3738 assert_eq!(q.get(&stale.id).unwrap().chat.as_deref(), Some("talk-1"));
3739 }
3740
3741 #[test]
3742 fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
3743 let dir = tempfile::tempdir().unwrap();
3744 let q = Queue::at(dir.path().join("queue"));
3745 let mut t = task("held");
3746 q.put(&mut t).unwrap();
3747 assert_eq!(
3748 crate::notices::Notices::at(dir.path().join("notifications"))
3749 .list()
3750 .len(),
3751 0
3752 );
3753 t.hold_machine(Some("out of attempts".to_owned()));
3754 q.put(&mut t).unwrap();
3755 let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
3756 assert_eq!(listed.len(), 1);
3757 assert!(listed[0].message.contains("out of attempts"));
3758 }
3759
3760 fn task(title: &str) -> Task {
3761 Task::new(
3762 title.to_owned(),
3763 format!("do {title}"),
3764 PathBuf::from("."),
3765 Source::Human,
3766 )
3767 }
3768
3769 #[test]
3770 fn attempt_limit_inherits_old_records_and_controls_failure() {
3771 let mut t = task("budget");
3772 let mut old = serde_json::to_value(&t).unwrap();
3773 old.as_object_mut().unwrap().remove("attempt_limit");
3774 old["schema"] = serde_json::json!(15);
3775 let legacy: Task = serde_json::from_value(old).unwrap();
3776 assert_eq!(legacy.attempt_limit, None);
3777 assert_eq!(legacy.effective_attempt_limit(2), 2);
3778 t.attempts = 2;
3779 t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
3780 t.fail("retry", 1);
3781 assert_eq!(t.status, TaskStatus::Failed);
3782 t.attempts = 3;
3783 t.fail("stop", 10);
3784 assert_eq!(t.status, TaskStatus::Held);
3785 t.change_attempt_limit(None).unwrap();
3786 assert_eq!(t.status, TaskStatus::Held);
3787 assert_eq!(t.attempts, 3);
3788 t.fail("inherit", 4);
3789 assert_eq!(t.status, TaskStatus::Failed);
3790 for status in [
3791 TaskStatus::Running,
3792 TaskStatus::Done,
3793 TaskStatus::Failed,
3794 TaskStatus::Blocked,
3795 TaskStatus::Parked,
3796 ] {
3797 t.status = status;
3798 assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
3799 assert_eq!(t.attempt_limit, None);
3800 }
3801 }
3802
3803 #[test]
3804 fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
3805 let mut t = task("retried");
3806 assert!(
3807 t.earlier_attempts().is_empty(),
3808 "a first attempt takes nothing over"
3809 );
3810 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3811 assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
3812 assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
3814 assert_eq!(t.successor_of("bbbb"), None);
3815 assert_eq!(t.successor_of("zzzz"), None);
3816 }
3817
3818 #[test]
3819 fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
3820 let (_dir, q) = queue();
3821 let mut t = task("resumed");
3822 for (runs, successor) in [
3823 (vec!["aaaa", "aaaa"], None),
3824 (vec!["aaaa", "bbbb", "aaaa"], None),
3825 (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
3826 (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
3827 ] {
3828 t.runs = runs.into_iter().map(str::to_owned).collect();
3829 q.put(&mut t).unwrap();
3830
3831 assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
3832 assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
3833 assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
3834 }
3835 }
3836
3837 #[test]
3838 fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
3839 let (_dir, q) = queue();
3840 let mut t = task("retried");
3841 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3842 q.put(&mut t).unwrap();
3843
3844 assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
3845 assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
3846 assert_eq!(
3847 q.superseded_by("cccc"),
3848 None,
3849 "the latest attempt replaces nothing"
3850 );
3851 assert_eq!(
3852 q.superseded_by("never-heard-of-it"),
3853 None,
3854 "a run belonging to no task on this queue is not superseded"
3855 );
3856
3857 let mut by = HashMap::new();
3858 by.insert("aaaa".to_owned(), "bbbb".to_owned());
3859 by.insert("bbbb".to_owned(), "cccc".to_owned());
3860 assert_eq!(
3861 q.superseded(),
3862 by,
3863 "the whole-map and single-run forms must agree"
3864 );
3865 }
3866
3867 #[test]
3868 fn superseded_attempts_is_empty_until_the_task_is_done() {
3869 let mut t = task("retried");
3870 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3871 t.status = TaskStatus::Failed;
3872 assert_eq!(
3873 t.superseded_attempts(true),
3874 &[] as &[String],
3875 "a task still retrying has no attempt yet that a later one made moot"
3876 );
3877
3878 t.status = TaskStatus::Running;
3879 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3880 }
3881
3882 #[test]
3883 fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
3884 let mut t = task("retried");
3885 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3886 t.status = TaskStatus::Done;
3887 assert_eq!(
3888 t.superseded_attempts(true),
3889 &["aaaa".to_owned(), "bbbb".to_owned()],
3890 "cccc is the attempt whose success made the task done, and stays out"
3891 );
3892 }
3893
3894 #[test]
3895 fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
3896 let mut t = task("first try landed");
3897 t.runs = vec!["aaaa".to_owned()];
3898 t.status = TaskStatus::Done;
3899 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3900 }
3901
3902 #[test]
3903 fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
3904 let mut t = task("closed by hand after a manual merge");
3911 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3912 t.status = TaskStatus::Done;
3913 assert_eq!(
3914 t.superseded_attempts(false),
3915 &[] as &[String],
3916 "nothing here is provably why the task is done, so nothing is superseded"
3917 );
3918 }
3919
3920 #[test]
3921 fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
3922 let (_dir, q) = queue();
3923 let mut t = task("retried twice");
3924 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3925 q.put(&mut t).unwrap();
3926
3927 assert_eq!(
3928 q.latest_attempt("aaaa"),
3929 Some("cccc".to_owned()),
3930 "an old attempt points straight at the chain's current head, not the \
3931 next attempt in the middle of it"
3932 );
3933 assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
3934 assert_eq!(
3935 q.latest_attempt("cccc"),
3936 None,
3937 "the latest attempt is not superseded by anything"
3938 );
3939 assert_eq!(
3940 q.latest_attempt("never-heard-of-it"),
3941 None,
3942 "a run belonging to no task on this queue is not superseded"
3943 );
3944 }
3945
3946 #[test]
3947 fn a_markdown_heading_is_the_title_not_decoration() {
3948 assert_eq!(
3953 title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
3954 "Rework the config loader"
3955 );
3956 assert_eq!(title_from("- fix the thing", 40), "fix the thing");
3957 assert_eq!(title_from("> quoted task", 40), "quoted task");
3958 assert_eq!(title_from(" \n\n", 40), "(empty task)");
3960 assert_eq!(title_from("###\n", 40), "(empty task)");
3961 }
3962
3963 #[test]
3964 fn a_long_title_is_elided_by_characters_not_bytes() {
3965 let long = "課題".repeat(30);
3967 let title = title_from(&long, 10);
3968 assert_eq!(title.chars().count(), 10);
3969 assert!(title.ends_with('…'));
3970 }
3971
3972 #[test]
3973 fn priority_wins_and_ties_break_oldest_first() {
3974 let (_dir, q) = queue();
3975 let mut a = task("first");
3976 let mut b = task("second");
3977 let mut c = task("urgent");
3978 a.id = "20260101-000001-aaaa".to_owned();
3980 b.id = "20260101-000002-bbbb".to_owned();
3981 c.id = "20260101-000003-cccc".to_owned();
3982 c.priority = 5;
3983 for t in [&mut a, &mut b, &mut c] {
3984 q.put(t).unwrap();
3985 }
3986
3987 assert_eq!(q.next_runnable().unwrap().id, c.id);
3989 c.hold_machine(None);
3990 q.put(&mut c).unwrap();
3991 assert_eq!(q.next_runnable().unwrap().id, a.id);
3993 assert_eq!(q.list().len(), 3, "b is still waiting its turn");
3994 }
3995
3996 #[test]
3997 fn a_blocked_task_never_starves_another_runnable_one() {
3998 let (_dir, q) = queue();
3999 let mut blocked = task("blocked");
4000 blocked.block(vec!["something".to_owned()], None);
4001 q.put(&mut blocked).unwrap();
4002
4003 let mut runnable = task("free to go");
4004 q.put(&mut runnable).unwrap();
4005
4006 let next = q.next_runnable().expect("a runnable task is still offered");
4007 assert_eq!(next.id, runnable.id);
4008 }
4009
4010 #[test]
4011 fn a_held_task_is_never_offered_to_the_loop() {
4012 let (_dir, q) = queue();
4013 let mut t = task("held");
4014 q.put(&mut t).unwrap();
4015 assert!(q.next_runnable().is_some());
4016
4017 t.hold_machine(None);
4018 q.put(&mut t).unwrap();
4019 assert!(
4020 q.next_runnable().is_none(),
4021 "a held task must wait for a human"
4022 );
4023
4024 t.status = TaskStatus::Failed;
4026 q.put(&mut t).unwrap();
4027 assert!(q.next_runnable().is_some());
4028 }
4029
4030 #[test]
4031 fn attempts_are_capped_and_then_the_task_is_held() {
4032 let mut t = task("doomed");
4033
4034 t.start("run-1".to_owned());
4035 t.fail("gate red", 2);
4036 assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
4037
4038 t.start("run-2".to_owned());
4039 t.fail("gate red", 2);
4040 assert_eq!(
4041 t.status,
4042 TaskStatus::Held,
4043 "out of attempts: stop spending money on it"
4044 );
4045 assert_eq!(t.runs, ["run-1", "run-2"]);
4046 assert_eq!(t.last_error.as_deref(), Some("gate red"));
4047 assert_eq!(
4048 t.hold_reason.as_deref(),
4049 Some("gate red"),
4050 "the hold must say why, not leave hold_reason null next to a \
4051 populated last_error"
4052 );
4053 }
4054
4055 #[test]
4056 fn handing_off_a_task_records_a_hold_reason_too() {
4057 let mut t = task("left a pull request");
4058 t.start("run-1".to_owned());
4059 t.handed_off("run ended with a pull request open [run run-1]");
4060 assert_eq!(t.status, TaskStatus::Held);
4061 assert_eq!(t.hold_source, Some(HoldSource::Machine));
4062 assert_eq!(
4063 t.hold_reason.as_deref(),
4064 Some("run ended with a pull request open [run run-1]")
4065 );
4066 assert_eq!(t.hold_reason, t.last_error);
4067 }
4068
4069 #[test]
4070 fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
4071 let mut t = task("stalled by quota");
4072
4073 t.start("run-1".to_owned());
4074 assert_eq!(t.attempts, 1);
4075 t.stall("judge-1, judge-2 out of quota");
4076 assert_eq!(
4077 t.attempts, 0,
4078 "a closed quota window must not spend the task's retry budget"
4079 );
4080 assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
4081 assert_eq!(
4082 t.last_error.as_deref(),
4083 Some("judge-1, judge-2 out of quota")
4084 );
4085
4086 for _ in 0..20 {
4089 t.start("run-n".to_owned());
4090 t.stall("still out of quota");
4091 }
4092 t.start("run-real".to_owned());
4093 t.fail("gate red", 2);
4094 assert_eq!(
4095 t.status,
4096 TaskStatus::Failed,
4097 "the first attempt that was really judged is attempt one"
4098 );
4099 }
4100
4101 #[test]
4102 fn releasing_a_held_task_gives_it_a_real_second_chance() {
4103 let mut t = task("retry me");
4104 t.start("run-1".to_owned());
4105 t.fail("gate red", 1);
4106 assert_eq!(t.status, TaskStatus::Held);
4107
4108 t.release();
4109 assert_eq!(t.status, TaskStatus::Queued);
4110 assert_eq!(t.attempts, 0);
4113 assert!(t.last_error.is_none());
4114 assert_eq!(
4115 t.runs.len(),
4116 1,
4117 "history is kept: attempts reset, evidence does not"
4118 );
4119 }
4120
4121 #[test]
4122 fn a_hold_reason_survives_and_a_release_clears_it() {
4123 let mut t = task("waiting on something else");
4124 t.hold_manual(Some(
4125 "waiting for 20260101-000000-aaaa to land first".to_owned(),
4126 ));
4127 assert_eq!(t.status, TaskStatus::Held);
4128 assert_eq!(
4129 t.hold_reason.as_deref(),
4130 Some("waiting for 20260101-000000-aaaa to land first")
4131 );
4132
4133 t.hold_manual(None);
4135 assert_eq!(
4136 t.hold_reason.as_deref(),
4137 Some("waiting for 20260101-000000-aaaa to land first"),
4138 "a bare re-hold keeps whatever a human already wrote down"
4139 );
4140
4141 let mut plain = task("no reason given");
4143 plain.hold_manual(None);
4144 assert_eq!(plain.status, TaskStatus::Held);
4145 assert!(plain.hold_reason.is_none());
4146
4147 t.release();
4148 assert_eq!(t.status, TaskStatus::Queued);
4149 assert!(
4150 t.hold_reason.is_none(),
4151 "a stale reason must not greet the next person who holds this task"
4152 );
4153 }
4154
4155 #[test]
4156 fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
4157 let mut t = task("landed by hand while held");
4162 t.hold_manual(Some("waiting on 3ed9".to_owned()));
4163 assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
4164
4165 t.succeed();
4166 assert_eq!(t.status, TaskStatus::Done);
4167 assert!(
4168 t.hold_reason.is_none(),
4169 "a done task cannot still be waiting on something"
4170 );
4171 }
4172
4173 #[test]
4174 fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
4175 let mut held = task("held straight out of blocked");
4182 held.block(
4183 vec!["20260101-000000-dead".to_owned()],
4184 Some("waiting on the migration script".to_owned()),
4185 );
4186 assert_eq!(held.status, TaskStatus::Blocked);
4187
4188 held.hold_manual(None);
4189 assert_eq!(held.status, TaskStatus::Held);
4190 assert!(
4191 held.blocked_by.is_empty(),
4192 "hold overrides the wait, same as release"
4193 );
4194 assert!(held.block_reason.is_none());
4195
4196 let mut done = task("closed straight out of blocked");
4197 done.block(
4198 vec!["20260101-000000-dead".to_owned()],
4199 Some("waiting on the migration script".to_owned()),
4200 );
4201 done.succeed();
4202 assert_eq!(done.status, TaskStatus::Done);
4203 assert!(
4204 done.blocked_by.is_empty(),
4205 "a done task cannot still be waiting on a dependency"
4206 );
4207 assert!(done.block_reason.is_none());
4208 }
4209
4210 #[test]
4211 fn a_blocked_task_is_never_offered_to_the_loop() {
4212 let mut t = task("blocked");
4213 assert!(t.status.runnable());
4214 t.block(
4215 vec!["dep-id".to_owned()],
4216 Some("waits on dep-id".to_owned()),
4217 );
4218 assert_eq!(t.status, TaskStatus::Blocked);
4219 assert!(!t.status.runnable());
4220 assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
4221 }
4222
4223 #[test]
4224 fn unblocking_the_last_dependency_returns_the_task_to_queued() {
4225 let mut t = task("blocked on two");
4226 t.block(
4227 vec!["a".to_owned(), "b".to_owned()],
4228 Some("waits on a and b".to_owned()),
4229 );
4230
4231 t.unblock("a");
4232 assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
4233 assert_eq!(t.blocked_by, ["b"]);
4234
4235 t.unblock("b");
4236 assert_eq!(t.status, TaskStatus::Queued);
4237 assert!(t.blocked_by.is_empty());
4238 assert!(t.block_reason.is_none());
4239 }
4240
4241 #[test]
4242 fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
4243 let mut t = task("never blocked");
4244 t.unblock("whatever");
4245 assert_eq!(t.status, TaskStatus::Queued);
4246 }
4247
4248 #[test]
4249 fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
4250 let mut t = task("held, then asked about");
4255 t.hold_machine(Some("out of attempts".to_owned()));
4256 assert_eq!(t.status, TaskStatus::Held);
4257
4258 t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
4259 assert_eq!(t.status, TaskStatus::Blocked);
4260
4261 t.record_answer("what now?".to_owned(), "leave it held".to_owned());
4262 t.unblock("q1");
4263 assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
4264 assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
4265 assert_eq!(t.hold_source, Some(HoldSource::Machine));
4266 assert!(t.blocked_from.is_none(), "consumed once restored");
4267 }
4268
4269 #[test]
4270 fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
4271 let mut t = task("manually held, then asked about");
4272 t.hold_manual(Some("waiting on a dependency".to_owned()));
4273
4274 t.block(vec!["q1".to_owned()], None);
4275 t.unblock("q1");
4276
4277 assert_eq!(t.status, TaskStatus::Held);
4278 assert_eq!(t.hold_source, Some(HoldSource::Manual));
4279 }
4280
4281 #[test]
4282 fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
4283 let mut t = task("held, blocked twice");
4287 t.hold_machine(None);
4288 t.block(vec!["q1".to_owned()], Some("first".to_owned()));
4289 t.block(
4290 vec!["q1".to_owned(), "q2".to_owned()],
4291 Some("second".to_owned()),
4292 );
4293
4294 t.unblock("q1");
4295 assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
4296 t.unblock("q2");
4297 assert_eq!(t.status, TaskStatus::Held);
4298 }
4299
4300 #[test]
4301 fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
4302 let mut t = task("blocked mid-run");
4306 t.start("run-1".to_owned());
4307 assert_eq!(t.status, TaskStatus::Running);
4308
4309 t.block(vec!["q1".to_owned()], None);
4310 t.unblock("q1");
4311 assert_eq!(t.status, TaskStatus::Queued);
4312 }
4313
4314 #[test]
4315 fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
4316 let mut t = task("legacy record, held before it was blocked");
4322 t.hold_source = Some(HoldSource::Machine);
4323 t.hold_reason = Some("legacy hold reason".to_owned());
4324 t.status = TaskStatus::Blocked;
4325 t.blocked_by = vec!["q1".to_owned()];
4326 t.blocked_from = None;
4327
4328 t.unblock("q1");
4329 assert_eq!(t.status, TaskStatus::Held);
4330 }
4331
4332 #[test]
4333 fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
4334 let mut t = task("legacy record, ordinary dependency block");
4335 t.status = TaskStatus::Blocked;
4336 t.blocked_by = vec!["dep".to_owned()];
4337 t.blocked_from = None;
4338
4339 t.unblock("dep");
4340 assert_eq!(t.status, TaskStatus::Queued);
4341 }
4342
4343 #[test]
4344 fn answering_a_question_is_recorded_and_survives_a_release() {
4345 let mut t = task("asked something");
4346 t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
4347 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
4348 t.unblock("q1");
4349 assert_eq!(t.status, TaskStatus::Queued);
4350 assert_eq!(t.answers.len(), 1);
4351 assert_eq!(t.answers[0].answer, "SQLite");
4352
4353 t.release();
4357 assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
4358 }
4359
4360 #[test]
4361 fn resumption_review_branch_requires_unchanged_authored_criteria() {
4362 for source in ["cli", "web"] {
4363 let mut t = task("review old candidate");
4364 t.request_review("candidate-A".into());
4365 t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
4366 let mut unchanged = t.clone();
4367 unchanged.release_operator(source, None);
4368 assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
4369 t.edit(
4370 "review approved head".into(),
4371 "Reuse measured PR head B; fix Clippy; review".into(),
4372 )
4373 .unwrap();
4374 t.release_operator(source, None);
4375 assert!(t.review_branch.is_none());
4376 assert!(t.review_instruction.is_none());
4377 let mut legacy = unchanged;
4379 legacy.review_instruction = None;
4380 legacy.hold_for_handover(None, "still checked out".into());
4381 legacy
4382 .edit("new plan".into(), "Reuse measured PR head B".into())
4383 .unwrap();
4384 legacy.release_operator(source, None);
4385 assert!(legacy.review_branch.is_none());
4386 }
4387 }
4388
4389 #[test]
4390 fn a_refused_handover_keeps_the_review_branch_across_release() {
4391 let mut t = task("refused takeover");
4392 t.start("run-1".to_owned());
4393 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
4394 assert_eq!(t.status, TaskStatus::Held);
4395 assert_eq!(t.attempts, 1);
4396 t.release();
4397 assert_eq!(t.status, TaskStatus::Queued);
4398 assert_eq!(t.attempts, 0);
4399 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4400
4401 t.hold_for_handover(None, "again".to_owned());
4403 t.requeue();
4404 assert!(t.review_branch.is_none());
4405
4406 let mut m = task("manual");
4408 m.review_branch = Some("magi/x/A".to_owned());
4409 m.hold_manual(None);
4410 m.release();
4411 assert!(m.review_branch.is_none());
4412 }
4413
4414 #[test]
4415 fn requesting_review_requeues_the_task_and_remembers_the_branch() {
4416 let mut t = task("blocked run with a surviving branch");
4417 t.start("run-1".to_owned());
4418 t.fail("blocked with major findings", 5);
4419 assert_eq!(t.status, TaskStatus::Failed);
4420
4421 t.request_review("magi/eba2/A".to_owned());
4422 assert_eq!(t.status, TaskStatus::Queued);
4423 assert_eq!(t.attempts, 0);
4424 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4425
4426 t.release();
4428 assert!(t.review_branch.is_none());
4429 }
4430
4431 #[test]
4432 fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
4433 let mut t = task("retry");
4434 t.start("run-1".to_owned());
4435 t.requeue();
4436 assert!(t.fresh_start);
4437
4438 t.release();
4439 assert!(!t.fresh_start);
4440 }
4441
4442 #[test]
4443 fn urgent_only_changes_queued_or_held_and_preserves_settings() {
4444 for status in [
4445 TaskStatus::Queued,
4446 TaskStatus::Held,
4447 TaskStatus::Running,
4448 TaskStatus::Done,
4449 TaskStatus::Failed,
4450 TaskStatus::Blocked,
4451 TaskStatus::Parked,
4452 ] {
4453 let mut t = task("urgent settings");
4454 t.status = status;
4455 t.attempts = 2;
4456 t.hold_reason = Some("waiting".into());
4457 t.priority = 7;
4458 t.interrupt = true;
4459 let before = serde_json::to_value(&t).unwrap();
4460 for urgent in [true, true, false, false] {
4461 let result = t.set_urgent(urgent);
4462 let mut expected = before.clone();
4463 if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
4464 result.unwrap();
4465 expected["urgent"] = urgent.into();
4466 } else {
4467 let error = result.unwrap_err().to_string();
4468 assert!(error.contains(status.as_str()), "{error}");
4469 assert!(error.contains("queued or held"), "{error}");
4470 }
4471 assert_eq!(serde_json::to_value(&t).unwrap(), expected);
4472 }
4473 }
4474 }
4475
4476 #[test]
4477 fn priority_can_be_changed_while_queued_but_not_while_running() {
4478 let mut t = task("reprioritise me");
4479 t.set_priority(5).unwrap();
4480 assert_eq!(t.priority, 5);
4481
4482 t.start("run-1".to_owned());
4483 let err = t.set_priority(9).unwrap_err().to_string();
4484 assert!(err.contains("running"), "{err}");
4485 assert_eq!(t.priority, 5, "the rejected write must not partially apply");
4486 }
4487
4488 #[test]
4489 fn interrupt_can_be_marked_while_queued_but_not_while_running() {
4490 let mut t = task("interrupt me");
4491 assert!(!t.interrupt, "off unless asked, same as any other task");
4492
4493 t.set_interrupt(true).unwrap();
4494 assert!(t.interrupt);
4495
4496 t.start("run-1".to_owned());
4497 assert!(
4498 !t.interrupt,
4499 "the mark is one-shot: dispatching the task fulfils it, \
4500 whatever the run that follows ends up doing"
4501 );
4502 let err = t.set_interrupt(true).unwrap_err().to_string();
4503 assert!(err.contains("running"), "{err}");
4504 t.set_interrupt(false).unwrap();
4507 assert!(!t.interrupt);
4508 }
4509
4510 #[test]
4514 fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
4515 let mut t = task("interrupt me");
4516 t.set_interrupt(true).unwrap();
4517 t.start("run-1".to_owned());
4518 t.fail("mock failure", 5);
4519 assert_eq!(t.status, TaskStatus::Failed);
4520 assert!(
4521 !t.interrupt,
4522 "one attempt already spent the mark; a retry is an ordinary \
4523 requeue, not a fresh interrupt request"
4524 );
4525 }
4526
4527 #[test]
4528 fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
4529 let (_dir, q) = queue();
4530 let mut a = task("first filed");
4531 let mut b = task("second filed");
4532 a.id = "20260101-000001-aaaa".to_owned();
4533 b.id = "20260101-000002-bbbb".to_owned();
4534 q.put(&mut a).unwrap();
4535 q.put(&mut b).unwrap();
4536
4537 assert_eq!(
4538 q.next_runnable().unwrap().id,
4539 a.id,
4540 "with equal priority the older task goes first, so a burst of \
4541 new work cannot starve it"
4542 );
4543 assert_eq!(
4544 q.list()[0].id,
4545 b.id,
4546 "but the list an operator reads is newest first, the same as \
4547 before priority existed - a's turn to run does not make it the \
4548 newest task"
4549 );
4550
4551 let mut a = q.get(&a.id).unwrap();
4552 a.set_priority(10).unwrap();
4553 q.put(&mut a).unwrap();
4554
4555 assert_eq!(
4556 q.next_runnable().unwrap().id,
4557 a.id,
4558 "a raised priority must be reflected the moment it is saved"
4559 );
4560 assert_eq!(
4564 q.list()[0].id,
4565 a.id,
4566 "the raised task must sort first in the list an operator reads, \
4567 not only in next_runnable's own ordering"
4568 );
4569 }
4570
4571 #[test]
4572 fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
4573 let mut t = Task::new(
4574 "old title".to_owned(),
4575 "old instruction".to_owned(),
4576 PathBuf::from("/repo"),
4577 Source::Agent {
4578 run: "20260101-000000-beef".to_owned(),
4579 node: "implement".to_owned(),
4580 },
4581 );
4582 let id = t.id.clone();
4583 let created_at = t.created_at;
4584 t.runs.push("20260101-000000-beef".to_owned());
4585
4586 t.edit("new title".to_owned(), "new instruction".to_owned())
4587 .unwrap();
4588
4589 assert_eq!(t.title, "new title");
4590 assert_eq!(t.instruction, "new instruction");
4591 assert_eq!(t.id, id, "editing must not mint a new id");
4592 assert_eq!(t.created_at, created_at);
4593 assert_eq!(
4594 t.source,
4595 Source::Agent {
4596 run: "20260101-000000-beef".to_owned(),
4597 node: "implement".to_owned(),
4598 },
4599 "editing must not turn agent attribution into human"
4600 );
4601 assert_eq!(t.runs, ["20260101-000000-beef"]);
4602 }
4603
4604 #[test]
4605 fn editing_is_refused_once_a_task_is_running_or_finished() {
4606 let mut running = task("in flight");
4607 running.start("run-1".to_owned());
4608 let err = running
4609 .edit("x".to_owned(), "y".to_owned())
4610 .unwrap_err()
4611 .to_string();
4612 assert!(err.contains("running"), "{err}");
4613
4614 let mut done = task("finished");
4615 done.succeed();
4616 let err = done
4617 .edit("x".to_owned(), "y".to_owned())
4618 .unwrap_err()
4619 .to_string();
4620 assert!(err.contains("done"), "{err}");
4621
4622 let mut queued = task("waiting");
4624 queued.edit("x".to_owned(), "y".to_owned()).unwrap();
4625 let mut held = task("parked");
4626 held.hold_machine(None);
4627 held.edit("x".to_owned(), "y".to_owned()).unwrap();
4628 }
4629
4630 #[test]
4631 fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
4632 let (_dir, q) = queue();
4633 let path = q.path_of("20260101-000000-aaaa");
4634 std::fs::create_dir_all(q.root()).unwrap();
4635 std::fs::write(
4636 &path,
4637 serde_json::json!({
4638 "schema": SCHEMA,
4639 "id": "20260101-000000-aaaa",
4640 "title": "from before hold reasons existed",
4641 "instruction": "from before hold reasons existed",
4642 "repo": ".",
4643 "source": { "kind": "human" },
4644 "status": "held",
4645 "created_at": Timestamp::now().to_string(),
4646 "updated_at": Timestamp::now().to_string(),
4647 })
4648 .to_string(),
4649 )
4650 .unwrap();
4651
4652 let task = q.get("20260101-000000-aaaa").expect("must still read");
4653 assert!(task.hold_reason.is_none());
4654 assert!(task.operator_held());
4655 }
4656
4657 #[test]
4658 fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
4659 let (_dir, q) = queue();
4660 let path = q.path_of("20260101-000000-bbbb");
4661 std::fs::create_dir_all(q.root()).unwrap();
4662 std::fs::write(
4663 &path,
4664 serde_json::json!({
4665 "schema": 2,
4666 "id": "20260101-000000-bbbb",
4667 "title": "old manual recovery",
4668 "instruction": "old manual recovery",
4669 "repo": ".",
4670 "source": { "kind": "human" },
4671 "status": "held",
4672 "hold_reason": "active manual recovery run20260912-224242-daf5",
4673 "created_at": Timestamp::now().to_string(),
4674 "updated_at": Timestamp::now().to_string(),
4675 })
4676 .to_string(),
4677 )
4678 .unwrap();
4679
4680 let task = q.get("20260101-000000-bbbb").expect("must still read");
4681 assert_eq!(task.hold_source, None);
4682 assert!(task.operator_held());
4683 }
4684
4685 #[test]
4686 fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
4687 let (_dir, q) = queue();
4688 let path = q.path_of("20260101-000000-aaaa");
4689 std::fs::create_dir_all(q.root()).unwrap();
4690 std::fs::write(
4691 &path,
4692 serde_json::json!({
4693 "schema": SCHEMA,
4694 "id": "20260101-000000-aaaa",
4695 "title": "from before diagnostics existed",
4696 "instruction": "from before diagnostics existed",
4697 "repo": ".",
4698 "source": { "kind": "human" },
4699 "status": "held",
4700 "created_at": Timestamp::now().to_string(),
4701 "updated_at": Timestamp::now().to_string(),
4702 })
4703 .to_string(),
4704 )
4705 .unwrap();
4706
4707 let task = q.get("20260101-000000-aaaa").expect("must still read");
4708 assert!(task.diagnostic.is_none());
4709 }
4710
4711 #[test]
4712 fn a_schema_1_task_with_no_blocking_fields_still_reads() {
4713 let (_dir, q) = queue();
4717 let path = q.path_of("20260101-000000-aaaa");
4718 std::fs::create_dir_all(q.root()).unwrap();
4719 std::fs::write(
4720 &path,
4721 serde_json::json!({
4722 "schema": 1,
4723 "id": "20260101-000000-aaaa",
4724 "title": "from before blocking existed",
4725 "instruction": "from before blocking existed",
4726 "repo": ".",
4727 "source": { "kind": "human" },
4728 "status": "queued",
4729 "created_at": Timestamp::now().to_string(),
4730 "updated_at": Timestamp::now().to_string(),
4731 })
4732 .to_string(),
4733 )
4734 .unwrap();
4735
4736 let task = q.get("20260101-000000-aaaa").expect("must still read");
4737 assert!(task.blocked_by.is_empty());
4738 assert!(task.block_reason.is_none());
4739 assert!(task.answers.is_empty());
4740 assert!(task.review_branch.is_none());
4741 }
4742
4743 #[test]
4744 fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
4745 let mut held = task("diagnosed");
4750 held.start("run-1".to_owned());
4751 held.fail("gate red", 1);
4752 held.diagnostic = Some("cargo test failed: ...".to_owned());
4753 assert_eq!(held.status, TaskStatus::Held);
4754
4755 held.release();
4756 assert!(held.diagnostic.is_none());
4757
4758 held.diagnostic = Some("cargo test failed: ...".to_owned());
4759 held.succeed();
4760 assert!(held.diagnostic.is_none());
4761 }
4762
4763 #[test]
4764 fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
4765 let mut t = task("retried");
4766 t.start("run-1".to_owned());
4767 t.diagnostic = Some("stale evidence from a previous hold".to_owned());
4768 t.fail("unrelated config error", 5);
4769 assert_eq!(t.status, TaskStatus::Failed);
4770 assert!(
4771 t.diagnostic.is_none(),
4772 "fail() must not let an old diagnostic outlive the run that produced it"
4773 );
4774 }
4775
4776 #[test]
4777 fn a_claim_is_exclusive_and_releases_on_drop() {
4778 let (_dir, q) = queue();
4779 let mut t = task("contended");
4780 q.put(&mut t).unwrap();
4781
4782 let held = q.claim(&t.id).unwrap();
4783 assert!(
4784 q.claim(&t.id).is_err(),
4785 "two daemons must not drive one task into two runs"
4786 );
4787 drop(held);
4788 assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
4789 }
4790
4791 #[test]
4792 fn a_round_trip_survives_disk() {
4793 let (_dir, q) = queue();
4794 let mut t = Task::new(
4795 "titled".to_owned(),
4796 "body".to_owned(),
4797 PathBuf::from("/repo"),
4798 Source::Agent {
4799 run: "20260101-000000-beef".to_owned(),
4800 node: "implement".to_owned(),
4801 },
4802 );
4803 t.priority = 3;
4804 q.put(&mut t).unwrap();
4805
4806 let back = q.get(&t.id).unwrap();
4807 assert_eq!(back.id, t.id);
4808 assert_eq!(back.priority, 3);
4809 assert_eq!(back.source.label(), "implement@beef");
4810 assert_eq!(q.get(t.short()).unwrap().id, t.id);
4812 }
4813
4814 #[test]
4815 fn an_unreadable_task_does_not_take_the_queue_down() {
4816 let (_dir, q) = queue();
4817 let mut t = task("fine");
4818 q.put(&mut t).unwrap();
4819 std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
4820
4821 let listed = q.list();
4822 assert_eq!(listed.len(), 1, "the readable task still lists");
4823 assert_eq!(listed[0].id, t.id);
4824 }
4825
4826 #[test]
4827 fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
4828 let (_dir, q) = queue();
4829 let path = q.path_of("20260101-000000-aaaa");
4830 std::fs::create_dir_all(q.root()).unwrap();
4831 std::fs::write(
4832 &path,
4833 serde_json::json!({
4834 "schema": SCHEMA,
4835 "id": "20260101-000000-aaaa",
4836 "title": "from before solo existed",
4837 "instruction": "from before solo existed",
4838 "repo": ".",
4839 "source": { "kind": "human" },
4840 "status": "queued",
4841 "created_at": Timestamp::now().to_string(),
4842 "updated_at": Timestamp::now().to_string(),
4843 })
4844 .to_string(),
4845 )
4846 .unwrap();
4847
4848 let task = q.get("20260101-000000-aaaa").expect("must still read");
4849 assert!(!task.solo, "a queue file with no `solo` field means false");
4850 }
4851
4852 #[test]
4853 fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
4854 let (_dir, q) = queue();
4855 let path = q.path_of("20260101-000000-bbbb");
4856 std::fs::create_dir_all(q.root()).unwrap();
4857 std::fs::write(
4858 &path,
4859 serde_json::json!({
4860 "schema": SCHEMA,
4861 "id": "20260101-000000-bbbb",
4862 "title": "from before urgent existed",
4863 "instruction": "from before urgent existed",
4864 "repo": ".",
4865 "source": { "kind": "human" },
4866 "status": "queued",
4867 "created_at": Timestamp::now().to_string(),
4868 "updated_at": Timestamp::now().to_string(),
4869 })
4870 .to_string(),
4871 )
4872 .unwrap();
4873
4874 let task = q.get("20260101-000000-bbbb").expect("must still read");
4875 assert!(
4876 !task.urgent,
4877 "a queue file with no `urgent` field means false, same as `solo`"
4878 );
4879 }
4880
4881 #[test]
4882 fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
4883 let (_dir, q) = queue();
4884 let mut t = task("from the future");
4885 q.put(&mut t).unwrap();
4886 let path = q.path_of(&t.id);
4887 let body = std::fs::read_to_string(&path)
4888 .unwrap()
4889 .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
4890 std::fs::write(&path, body).unwrap();
4891
4892 let err = q.get(&t.id).unwrap_err().to_string();
4893 assert!(err.contains("schema 99"), "{err}");
4894 }
4895
4896 #[test]
4897 fn revision_moves_when_the_queue_changes() {
4898 let (_dir, q) = queue();
4899 assert_eq!(q.revision(), 0, "an empty queue has no revision");
4900 let mut t = task("first");
4901 q.put(&mut t).unwrap();
4902 assert!(q.revision() > 0, "a written task moves the revision");
4903 }
4904
4905 #[test]
4906 fn revision_moves_when_deleting_an_older_task() {
4907 let (dir, q) = queue();
4908 let questions = Questions::at(dir.path().join("questions"));
4909 let mut t1 = task("older");
4910 q.put(&mut t1).unwrap();
4911 std::thread::sleep(std::time::Duration::from_millis(10));
4913 let mut t2 = task("newer");
4914 q.put(&mut t2).unwrap();
4915
4916 let rev_before = q.revision();
4917 q.remove(&t1.id, false, &questions).unwrap();
4918 let rev_after = q.revision();
4919
4920 assert_ne!(
4921 rev_before, rev_after,
4922 "deleting an older task must change the revision so other clients see the deletion"
4923 );
4924 }
4925
4926 #[test]
4927 fn removing_a_task_takes_it_out_of_the_listing() {
4928 let (dir, q) = queue();
4929 let questions = Questions::at(dir.path().join("questions"));
4930 let mut t = task("delete me");
4931 q.put(&mut t).unwrap();
4932 let removed = q.remove(t.short(), false, &questions).unwrap();
4933 assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
4934 assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
4935 assert!(q.list().is_empty());
4936 assert!(
4937 q.remove(&t.id, false, &questions).is_err(),
4938 "removing twice is an error"
4939 );
4940 }
4941
4942 #[test]
4943 fn removing_a_task_takes_its_stale_lock_with_it() {
4944 let (dir, q) = queue();
4945 let questions = Questions::at(dir.path().join("questions"));
4946 let mut t = task("interrupted");
4947 q.put(&mut t).unwrap();
4948
4949 let claim = q.claim(&t.id).unwrap();
4952 std::mem::forget(claim);
4953 assert!(
4954 q.claim(&t.id).is_err(),
4955 "the orphaned lock is what makes the task look claimed"
4956 );
4957
4958 let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
4960 assert!(err.contains("live daemon"), "{err}");
4961 assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
4962
4963 q.remove(&t.id, false, &questions).unwrap();
4965 assert!(q.list().is_empty());
4966 let mut again = task("interrupted");
4967 again.id = t.id.clone();
4968 q.put(&mut again).unwrap();
4969 assert!(
4970 q.claim(&t.id).is_ok(),
4971 "a task that comes back must be claimable, which a left-behind lock would prevent"
4972 );
4973 }
4974
4975 fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
4976 crate::notices::Notices::at(dir.join("notifications")).list()
4977 }
4978
4979 #[test]
4980 fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
4981 let (dir, q) = queue();
4982 let questions = Questions::at(dir.path().join("questions"));
4983 let mut dep = task("dependency");
4984 q.put(&mut dep).unwrap();
4985 let mut blocked = task("waiting");
4986 blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
4987 q.put(&mut blocked).unwrap();
4988
4989 let removed = q.remove(&dep.id, false, &questions).unwrap();
4990 assert_eq!(removed.released, [blocked.id.clone()]);
4991 assert!(removed.still_blocked.is_empty());
4992
4993 let after = q.get(&blocked.id).unwrap();
4994 assert_eq!(after.status, TaskStatus::Queued);
4995 assert!(after.blocked_by.is_empty());
4996 assert!(after.block_reason.is_none());
4997 let notes = notices_of(dir.path());
4998 assert_eq!(notes.len(), 1, "{notes:?}");
4999 assert_eq!(notes[0].severity, crate::notices::Severity::Info);
5000 assert_eq!(
5001 notes[0].link,
5002 Some(crate::notices::Link::Task {
5003 id: blocked.id.clone(),
5004 })
5005 );
5006 }
5007
5008 #[test]
5009 fn removing_one_of_two_dependencies_keeps_the_other() {
5010 let (dir, q) = queue();
5011 let questions = Questions::at(dir.path().join("questions"));
5012 let mut dep = task("dependency");
5013 q.put(&mut dep).unwrap();
5014 let mut other = task("other");
5015 q.put(&mut other).unwrap();
5016 let mut blocked = task("waiting");
5017 blocked.block(
5018 vec![dep.id.clone(), other.id.clone()],
5019 Some("waits on both".to_owned()),
5020 );
5021 q.put(&mut blocked).unwrap();
5022
5023 let removed = q.remove(&dep.id, false, &questions).unwrap();
5024 assert!(removed.released.is_empty());
5025 assert_eq!(removed.still_blocked, [blocked.id.clone()]);
5026
5027 let after = q.get(&blocked.id).unwrap();
5028 assert_eq!(after.status, TaskStatus::Blocked);
5029 assert_eq!(after.blocked_by, [other.id.clone()]);
5030 assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
5033 }
5034
5035 #[test]
5036 fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
5037 let (dir, q) = queue();
5038 let questions = Questions::at(dir.path().join("questions"));
5039 let mut dep = task("dependency");
5040 q.put(&mut dep).unwrap();
5041 let mut blocked = task("waiting");
5042 blocked.block(vec![dep.id.clone()], None);
5043 q.put(&mut blocked).unwrap();
5044
5045 let claim = q.claim(&blocked.id).unwrap();
5047 let removed = q.remove(&dep.id, false, &questions).unwrap();
5048 assert!(removed.released.is_empty());
5049 drop(claim);
5050
5051 let mut task = q.get(&blocked.id).unwrap();
5052 assert_eq!(task.status, TaskStatus::Blocked);
5053 assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
5054 assert_eq!(
5055 q.apply_deleted_blockers(&mut task, &questions),
5056 [dep.id.clone()]
5057 );
5058 assert_eq!(task.status, TaskStatus::Queued);
5059 }
5060
5061 #[test]
5062 fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
5063 let (dir, q) = queue();
5064 let questions = Questions::at(dir.path().join("questions"));
5065 let mut dep = task("dependency");
5066 dep.id = "20261013-193611-ca5b".into();
5067 q.put(&mut dep).unwrap();
5068 let mut waiting = task("waiting");
5069 waiting.block(vec!["ca5b".into()], None);
5070 q.put(&mut waiting).unwrap();
5071 q.remove(&dep.id, false, &questions).unwrap();
5072 assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
5073 assert_eq!(
5074 dependency(&q, &questions, "ca5b"),
5075 Dependency::Deleted(dep.id)
5076 );
5077 }
5078
5079 #[test]
5080 fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
5081 let (dir, queue) = queue();
5082 let questions = Questions::at(dir.path().join("questions"));
5083 let mut dep = task("dependency");
5084 dep.id = "20261013-193611-ca5b".into();
5085 queue.put(&mut dep).unwrap();
5086 std::fs::create_dir_all(questions.root()).unwrap();
5087 let question_id = "20261014-193611-ca5b";
5088 std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
5089 assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
5090 assert_eq!(
5091 dependency(&queue, &questions, "ca5b"),
5092 Dependency::Ambiguous
5093 );
5094 assert_eq!(
5095 dependency(&queue, &questions, &dep.id),
5096 Dependency::Task(dep.id)
5097 );
5098 assert_eq!(
5099 dependency(&queue, &questions, question_id),
5100 Dependency::Question(question_id.into())
5101 );
5102 assert_eq!(
5103 dependency(&queue, &questions, "missing"),
5104 Dependency::Missing
5105 );
5106 }
5107
5108 #[test]
5109 fn dependency_resolution_directory_errors_are_unknown_not_missing() {
5110 let dir = tempfile::tempdir().unwrap();
5111 let q = Queue::at(dir.path().join("not-a-directory"));
5112 std::fs::write(q.root(), "file").unwrap();
5113 let questions = Questions::at(dir.path().join("questions"));
5114 assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
5115 assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
5116 }
5117
5118 #[test]
5119 fn a_dependency_without_a_tombstone_is_still_missing() {
5120 let (dir, q) = queue();
5121 let questions = Questions::at(dir.path().join("questions"));
5122 let mut dep = task("dependency");
5123 q.put(&mut dep).unwrap();
5124 std::fs::remove_file(q.path_of(&dep.id)).unwrap();
5125 assert_eq!(
5126 missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
5127 [dep.id.clone()]
5128 );
5129 assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
5130 }
5131
5132 #[test]
5133 fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
5134 let mut t = task("held");
5135 t.hold_machine(Some("because".to_owned()));
5136 t.blocked_by = vec!["gone".to_owned()];
5137 assert!(!t.dependency_deleted("gone"));
5138 assert_eq!(t.status, TaskStatus::Held);
5139 }
5140
5141 #[test]
5142 fn a_failed_record_removal_takes_the_tombstone_back() {
5143 let (_dir, q) = queue();
5144 let mut t = task("stays");
5145 q.put(&mut t).unwrap();
5146 q.write_tombstone(&t.id).unwrap();
5147 let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
5148 assert!(err.is_err());
5149 assert!(!q.was_deleted(&t.id));
5152 }
5153
5154 fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
5155 let p = dir.join(name);
5156 std::fs::write(&p, body).unwrap();
5157 p
5158 }
5159
5160 #[test]
5161 fn an_attachment_copy_survives_deleting_its_source() {
5162 let (dir, q) = queue();
5163 let src = source_file(dir.path(), "shot.png", "pixels");
5164 let mut t = task("with a picture");
5165 let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
5166 q.put(&mut t).unwrap();
5167 std::fs::remove_file(&src).unwrap();
5168 assert_eq!(names, ["shot.png"]);
5169 let loaded = q.get(&t.id).unwrap();
5170 let paths = q.attachment_paths(&loaded);
5171 assert_eq!(paths.len(), 1);
5172 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
5173 }
5174
5175 #[test]
5176 fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
5177 let (dir, q) = queue();
5178 let mut t = task("bad names");
5179 for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
5180 let src = source_file(dir.path(), name, "x");
5181 assert!(
5182 q.attach(&mut t, &[src]).is_err(),
5183 "`{name}` must be refused"
5184 );
5185 }
5186 assert!(!crate::ask::valid_asset_name("C:foo.png"));
5190 #[cfg(not(windows))]
5191 {
5192 let src = source_file(dir.path(), "C:foo.png", "x");
5193 assert!(q.attach(&mut t, &[src]).is_err());
5194 }
5195 let long = format!("{}.png", "a".repeat(70));
5196 let src = source_file(dir.path(), &long, "x");
5197 assert!(q.attach(&mut t, &[src]).is_err());
5198 assert!(t.attachments.is_empty());
5199 assert!(!q.attachments_dir(&t.id).exists());
5200 }
5201
5202 #[test]
5203 fn a_taken_attachment_name_is_numbered_not_overwritten() {
5204 let (dir, q) = queue();
5205 let a = source_file(dir.path(), "shot.png", "one");
5206 let sub = dir.path().join("other");
5207 std::fs::create_dir_all(&sub).unwrap();
5208 let b = source_file(&sub, "shot.png", "two");
5209 let mut t = task("collision");
5210 q.attach(&mut t, &[a]).unwrap();
5211 q.attach(&mut t, &[b]).unwrap();
5212 assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
5213 let paths = q.attachment_paths(&t);
5214 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
5215 assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
5216 }
5217
5218 #[test]
5219 fn a_renumbered_name_stays_inside_the_length_bound() {
5220 let (dir, q) = queue();
5221 let name = format!("{}.png", "a".repeat(60));
5222 assert_eq!(name.len(), 64);
5223 let a = source_file(dir.path(), &name, "one");
5224 let sub = dir.path().join("other");
5225 std::fs::create_dir_all(&sub).unwrap();
5226 let b = source_file(&sub, &name, "two");
5227 let mut t = task("long");
5228 q.attach(&mut t, &[a, b]).unwrap();
5229 assert_eq!(t.attachments.len(), 2);
5230 assert!(
5231 t.attachments
5232 .iter()
5233 .all(|n| crate::ask::valid_asset_name(n))
5234 );
5235 assert!(t.attachments[1].ends_with("-2.png"));
5236 }
5237
5238 #[test]
5239 fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
5240 let (dir, q) = queue();
5241 let good = source_file(dir.path(), "good.png", "ok");
5242 let mut t = task("partial");
5243 q.attach(&mut t, &[good]).unwrap();
5244 let more = source_file(dir.path(), "more.png", "ok");
5245 let missing = dir.path().join("missing.png");
5246 assert!(q.attach(&mut t, &[more, missing]).is_err());
5247 assert_eq!(t.attachments, ["good.png"]);
5248 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
5249 .unwrap()
5250 .flatten()
5251 .collect();
5252 assert_eq!(on_disk.len(), 1);
5253 }
5254
5255 fn block_put(q: &Queue, t: &Task) -> PathBuf {
5257 let tmp = q.path_of(&t.id).with_extension("json.tmp");
5258 std::fs::create_dir_all(&tmp).unwrap();
5259 tmp
5260 }
5261
5262 #[test]
5263 fn a_failed_put_leaves_no_new_attachment_directory() {
5264 let (dir, q) = queue();
5265 let mut t = task("fresh");
5266 let tmp = block_put(&q, &t);
5267 let src = source_file(dir.path(), "shot.png", "x");
5268 assert!(q.attach_and_put(&mut t, &[src]).is_err());
5269 assert!(t.attachments.is_empty());
5270 assert!(!q.attachments_dir(&t.id).exists());
5271 assert!(!q.path_of(&t.id).exists());
5272 std::fs::remove_dir(tmp).unwrap();
5273 }
5274
5275 #[test]
5276 fn a_failed_put_removes_only_the_copy_it_just_made() {
5277 let (dir, q) = queue();
5278 let mut t = task("edited");
5279 let first = source_file(dir.path(), "first.png", "1");
5280 q.attach_and_put(&mut t, &[first]).unwrap();
5281 block_put(&q, &t);
5282 let second = source_file(dir.path(), "second.png", "2");
5283 assert!(q.attach_and_put(&mut t, &[second]).is_err());
5284 assert_eq!(t.attachments, ["first.png"]);
5285 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
5286 .unwrap()
5287 .flatten()
5288 .map(|e| e.file_name().to_string_lossy().into_owned())
5289 .collect();
5290 assert_eq!(on_disk, ["first.png"]);
5291 assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
5292 }
5293
5294 #[test]
5295 fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
5296 let (dir, q) = queue();
5297 let questions = Questions::at(dir.path().join("questions"));
5298 let gone = task("gone");
5299 let mut other = task("other");
5300 let mut live = task("live");
5301 q.put(&mut other).unwrap();
5302 q.put(&mut live).unwrap();
5303 let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
5306 std::fs::create_dir_all(&orphan).unwrap();
5307 std::fs::write(orphan.join("shot.png"), "x").unwrap();
5308 let busy = q.root.join(format!("{}.attachments.removing", live.id));
5311 std::fs::create_dir_all(&busy).unwrap();
5312
5313 q.remove(&other.id, false, &questions).unwrap();
5314 assert!(!orphan.exists(), "an orphan is swept");
5315 assert!(busy.exists(), "a removal in progress is left alone");
5316 }
5317
5318 #[test]
5319 fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
5320 let (dir, q) = queue();
5321 let questions = Questions::at(dir.path().join("questions"));
5322 let mut t = task("stuck");
5323 let src = source_file(dir.path(), "shot.png", "x");
5324 q.attach_and_put(&mut t, &[src]).unwrap();
5325 let aside = q.root.join(format!("{}.attachments.removing", t.id));
5328 std::fs::create_dir_all(&aside).unwrap();
5329 std::fs::write(aside.join("old.png"), "o").unwrap();
5330 assert!(q.remove(&t.id, false, &questions).is_err());
5331 assert!(q.path_of(&t.id).exists());
5332 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5333 }
5334
5335 #[test]
5336 fn a_failed_record_removal_puts_the_attachments_back() {
5337 let (dir, q) = queue();
5338 let mut t = task("rollback");
5339 let src = source_file(dir.path(), "shot.png", "x");
5340 q.attach_and_put(&mut t, &[src]).unwrap();
5341 let err = q
5342 .remove_record_with_attachments(&t.id, |_| {
5343 Err(std::io::Error::other("injected failure"))
5344 })
5345 .unwrap_err();
5346 assert!(format!("{err:#}").contains("injected failure"));
5347 assert!(q.path_of(&t.id).exists());
5348 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5349 assert!(
5350 !q.root
5351 .join(format!("{}.attachments.removing", t.id))
5352 .exists()
5353 );
5354 }
5355
5356 #[test]
5357 fn editing_a_task_keeps_its_attachments() {
5358 let (dir, q) = queue();
5359 let src = source_file(dir.path(), "shot.png", "x");
5360 let mut t = task("editable");
5361 q.attach(&mut t, &[src]).unwrap();
5362 t.edit("new".to_owned(), "new text".to_owned()).unwrap();
5363 q.put(&mut t).unwrap();
5364 assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
5365 }
5366
5367 #[test]
5368 fn removing_a_task_deletes_its_attachments() {
5369 let (dir, q) = queue();
5370 let questions = Questions::at(dir.path().join("questions"));
5371 let src = source_file(dir.path(), "shot.png", "x");
5372 let mut t = task("doomed");
5373 q.attach(&mut t, &[src]).unwrap();
5374 q.put(&mut t).unwrap();
5375 assert!(q.attachments_dir(&t.id).is_dir());
5376 q.remove(&t.id, false, &questions).unwrap();
5377 assert!(!q.attachments_dir(&t.id).exists());
5378 assert!(q.list().is_empty());
5379 }
5380
5381 #[test]
5382 fn a_task_written_before_attachments_still_reads() {
5383 let (_dir, q) = queue();
5384 let mut t = task("old");
5385 q.put(&mut t).unwrap();
5386 let path = q.path_of(&t.id);
5387 let mut v: serde_json::Value =
5388 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
5389 v.as_object_mut().unwrap().remove("attachments");
5390 std::fs::write(&path, v.to_string()).unwrap();
5391 assert!(q.get(&t.id).unwrap().attachments.is_empty());
5392 }
5393
5394 #[test]
5395 fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
5396 let q = Queue::at(PathBuf::from("relative-queue"));
5397 let mut t = task("rel");
5398 t.attachments.push("shot.png".to_owned());
5399 let paths = q.attachment_paths(&t);
5400 assert!(paths[0].is_absolute(), "{}", paths[0].display());
5401 assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
5402 }
5403
5404 fn upgrade_parked(id: &str, started: &str) -> Task {
5405 let mut t = task(id);
5406 t.id = id.into();
5407 t.start(format!("run-{id}"));
5408 t.park("upgrade");
5409 t.record_pass_at(
5410 &format!("run-{id}"),
5411 0,
5412 Some(started.parse().unwrap()),
5413 1,
5414 Some(crate::run::ParkCause::Upgrade {
5415 from: "old".into(),
5416 to: None,
5417 }),
5418 );
5419 t
5420 }
5421
5422 #[test]
5423 fn upgrade_resume_order_follows_dispatch_claim_not_graph_start() {
5424 let stamp = |id: &str, dispatched: &str, started: &str| {
5426 let mut t = task(id);
5427 t.id = id.into();
5428 t.start(format!("run-{id}"));
5429 t.park("upgrade");
5430 t.dispatched_at = Some(dispatched.parse().unwrap());
5431 t.record_pass_at(
5432 &format!("run-{id}"),
5433 0,
5434 Some(started.parse().unwrap()),
5435 1,
5436 Some(crate::run::ParkCause::Upgrade {
5437 from: "old".into(),
5438 to: None,
5439 }),
5440 );
5441 t
5442 };
5443 let a = stamp("a", "2026-10-22T00:00:01Z", "2026-10-22T00:00:09Z");
5444 let b = stamp("b", "2026-10-22T00:00:02Z", "2026-10-22T00:00:03Z");
5445 assert!(dispatch_order(&a, &b).is_lt());
5446 }
5447
5448 #[test]
5449 fn upgrade_resume_order_survives_queue_reload_and_reversed_settlement() {
5450 let (_dir, q) = queue();
5451 let mut later = upgrade_parked("later", "2026-10-22T00:00:02Z");
5452 let mut earlier = upgrade_parked("earlier", "2026-10-22T00:00:01Z");
5453 earlier.priority = -10;
5454 let mut fresh = task("fresh");
5455 fresh.id = "newer".into();
5456 fresh.priority = 100;
5457 for t in [&mut fresh, &mut later, &mut earlier] {
5458 q.put(t).unwrap();
5459 }
5460 let reloaded = Queue::at(q.root().to_owned());
5461 let mut tasks = reloaded.list();
5462 tasks.sort_unstable_by(dispatch_order);
5463 assert_eq!(
5464 tasks.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
5465 vec!["earlier", "later", "newer"]
5466 );
5467 assert_eq!(reloaded.next_runnable().unwrap().id, "earlier");
5468 assert_eq!(reloaded.get("earlier").unwrap().attempts, 0);
5469 }
5470
5471 #[test]
5472 fn upgrade_resume_legacy_order_uses_settlement_and_id() {
5473 let mut a = upgrade_parked("a", "2026-10-22T00:00:02Z");
5474 let mut b = upgrade_parked("b", "2026-10-22T00:00:01Z");
5475 for t in [&mut a, &mut b] {
5476 let mut value = serde_json::to_value(&*t).unwrap();
5477 value["passes"][0]
5478 .as_object_mut()
5479 .unwrap()
5480 .remove("started_at");
5481 *t = serde_json::from_value(value).unwrap();
5482 }
5483 assert!(dispatch_order(&a, &b).is_lt());
5484 b.passes[0].settled_at = a.passes[0].settled_at;
5485 assert!(dispatch_order(&a, &b).is_lt());
5486 }
5487
5488 #[test]
5489 fn upgrade_resume_key_excludes_other_parks_and_explicit_task_changes() {
5490 let original = upgrade_parked("a", "2026-10-22T00:00:01Z");
5491 assert!(original.upgrade_resume_key().is_some());
5492 for cause in [
5493 crate::run::ParkCause::Operator,
5494 crate::run::ParkCause::Question,
5495 crate::run::ParkCause::Interrupt {
5496 reason: "interrupt".into(),
5497 },
5498 crate::run::ParkCause::Unspecified,
5499 ] {
5500 let mut t = original.clone();
5501 t.passes[0].parked = Some(cause);
5502 assert!(t.upgrade_resume_key().is_none());
5503 let mut fresh = task("fresh");
5504 fresh.priority = 1;
5505 assert!(dispatch_order(&fresh, &t).is_lt());
5506 }
5507 let mut t = original.clone();
5508 t.release();
5509 assert!(t.upgrade_resume_key().is_none());
5510 t.edit("edited".into(), "new instruction".into()).unwrap();
5511 assert!(t.upgrade_resume_key().is_none());
5512 let mut t = original.clone();
5513 t.requeue();
5514 t.status = TaskStatus::Parked;
5515 assert!(t.upgrade_resume_key().is_none());
5516 let mut t = original.clone();
5517 t.start("different-run".into());
5518 t.park("no upgrade");
5519 assert!(t.upgrade_resume_key().is_none());
5520 }
5521
5522 #[test]
5523 fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
5524 let dir = tempfile::tempdir().unwrap();
5525 let q = Queue::at(dir.path().join("queue"));
5526 let mut t = task("resume");
5527 t.start("A".into());
5528 t.record_pass("A", 0, 1, None);
5529 t.start("B".into());
5530 q.put(&mut t).unwrap();
5531 assert_eq!(t.runs, vec!["A", "B"]);
5532 t.start("A".into());
5533 t.record_pass("A", 1, 3, None);
5534 q.put(&mut t).unwrap();
5535 let saved = q.get(&t.id).unwrap();
5536 assert_eq!(saved.runs, vec!["B", "A"]);
5537 assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
5538 assert_eq!(saved.attempts, 3);
5539 assert_eq!(saved.passes.len(), 2);
5540 assert_eq!(saved.passes[0].run, "A");
5541 assert_eq!(saved.passes[1].run, "A");
5542 }
5543
5544 #[test]
5545 fn repeated_start_records_one_run_and_separate_passes() {
5546 let mut t = task("resume");
5547 t.start("same-run".into());
5548 t.record_pass("same-run", 0, 1, None);
5549 t.start("same-run".into());
5550 t.record_pass("same-run", 1, 2, None);
5551 assert_eq!(t.runs, vec!["same-run"]);
5552 assert_eq!(t.attempts, 2);
5553 assert_eq!(t.passes.len(), 2);
5554 }
5555
5556 #[test]
5557 fn link_run_adds_a_run_once_and_touches_nothing_else() {
5558 let dir = tempfile::tempdir().unwrap();
5559 let queue = Queue::at(dir.path().join("queue"));
5560 let mut t = Task::new(
5561 "t".to_owned(),
5562 "do it".to_owned(),
5563 PathBuf::from("."),
5564 Source::Human,
5565 );
5566 queue.put(&mut t).unwrap();
5567 let before = queue.get(&t.id).unwrap();
5568
5569 let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
5570 assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
5571 assert_eq!(linked.status, before.status);
5572 assert_eq!(linked.attempts, before.attempts);
5573 assert_eq!(linked.interrupt, before.interrupt);
5574
5575 let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5576 assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
5577 assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
5578 assert!(queue.link_run("no-such-task", "r").is_err());
5579 }
5580
5581 #[test]
5582 fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
5583 let dir = tempfile::tempdir().unwrap();
5584 let queue = Queue::at(dir.path().join("queue"));
5585 let mut t = Task::new(
5586 "t".to_owned(),
5587 "do it".to_owned(),
5588 PathBuf::from("."),
5589 Source::Human,
5590 );
5591 queue.put(&mut t).unwrap();
5592 let mut snapshot = queue.get(&t.id).unwrap();
5594 queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5595
5596 snapshot.start("20260930-000000-aaaa".to_owned());
5597 queue.put(&mut snapshot).unwrap();
5598
5599 let stored = queue.get(&t.id).unwrap();
5600 assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
5601 assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
5602 assert_eq!(stored.attempts, 1);
5603 }
5604
5605 #[test]
5606 fn concurrent_links_and_daemon_saves_lose_nothing() {
5607 let dir = tempfile::tempdir().unwrap();
5608 let queue = Queue::at(dir.path().join("queue"));
5609 let mut t = Task::new(
5610 "t".to_owned(),
5611 "do it".to_owned(),
5612 PathBuf::from("."),
5613 Source::Human,
5614 );
5615 queue.put(&mut t).unwrap();
5616 let id = t.id.clone();
5617
5618 let linkers: Vec<_> = (0..4)
5619 .map(|n| {
5620 let (queue, id) = (queue.clone(), id.clone());
5621 std::thread::spawn(move || {
5622 for k in 0..10 {
5623 queue
5624 .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
5625 .unwrap();
5626 }
5627 })
5628 })
5629 .collect();
5630 let mut mine = queue.get(&id).unwrap();
5633 for k in 0..10 {
5634 mine.start(format!("20260930-000009-d{k:03}"));
5635 queue.put(&mut mine).unwrap();
5636 }
5637 for l in linkers {
5638 l.join().unwrap();
5639 }
5640
5641 let stored = queue.get(&id).unwrap();
5642 assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
5643 assert_eq!(
5644 stored.attempts, 10,
5645 "linking never rewinds the daemon's work"
5646 );
5647 }
5648
5649 fn age_lock(path: &Path) {
5650 let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
5651 f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
5652 .unwrap();
5653 }
5654
5655 #[test]
5656 fn concurrent_stale_takeover_yields_one_holder() {
5657 use std::sync::atomic::{AtomicUsize, Ordering};
5658 use std::sync::{Arc, Barrier};
5659 for _ in 0..5 {
5660 let dir = tempfile::tempdir().unwrap();
5661 let q = Queue::at(dir.path().to_path_buf());
5662 let lock = dir.path().join("t.write-lock");
5663 std::fs::write(&lock, "dead-0000").unwrap();
5664 age_lock(&lock);
5665 let n = 6;
5666 let barrier = Arc::new(Barrier::new(n));
5667 let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
5668 let handles: Vec<_> = (0..n)
5669 .map(|_| {
5670 let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
5671 std::thread::spawn(move || {
5672 b.wait();
5673 let g = q.lock_task("t").unwrap();
5674 let held = now.fetch_add(1, Ordering::SeqCst) + 1;
5675 max.fetch_max(held, Ordering::SeqCst);
5676 std::thread::sleep(std::time::Duration::from_millis(20));
5677 now.fetch_sub(1, Ordering::SeqCst);
5678 drop(g);
5679 })
5680 })
5681 .collect();
5682 for h in handles {
5683 h.join().unwrap();
5684 }
5685 assert_eq!(max.load(Ordering::SeqCst), 1);
5686 assert!(!lock.exists());
5687 }
5688 }
5689
5690 #[test]
5691 fn dropping_a_replaced_lock_leaves_the_new_holders_token() {
5692 let dir = tempfile::tempdir().unwrap();
5693 let q = Queue::at(dir.path().to_path_buf());
5694 let lock = dir.path().join("t.write-lock");
5695 let a = q.lock_task("t").unwrap();
5696 let replacement = owner_token();
5699 std::fs::write(&lock, &replacement).unwrap();
5700 drop(a);
5701 assert_eq!(std::fs::read_to_string(&lock).unwrap(), replacement);
5702 }
5703
5704 #[test]
5705 fn aged_live_task_lock_excludes_updates_until_retirement_releases_it() {
5706 let dir = tempfile::tempdir().unwrap();
5707 let q = Queue::at(dir.path().to_path_buf());
5708 let held = q.lock_task("t").unwrap();
5709 let path = dir.path().join("t.write-lock");
5710 age_lock(&path);
5711 assert!(!break_stale(&path, &held.token));
5712 let (started_tx, started_rx) = std::sync::mpsc::channel();
5713 let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
5714 let updater = std::thread::spawn(move || {
5715 started_tx.send(()).unwrap();
5716 let _write = q.lock_task("t").unwrap();
5717 acquired_tx.send(()).unwrap();
5718 });
5719 started_rx.recv().unwrap();
5720 assert!(
5721 acquired_rx
5722 .recv_timeout(std::time::Duration::from_millis(100))
5723 .is_err()
5724 );
5725 assert_eq!(std::fs::read_to_string(&path).unwrap(), held.token);
5726 drop(held);
5727 acquired_rx
5728 .recv_timeout(std::time::Duration::from_secs(2))
5729 .unwrap();
5730 updater.join().unwrap();
5731 assert!(!path.exists());
5732 }
5733
5734 #[test]
5735 fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
5736 let dir = tempfile::tempdir().unwrap();
5737 let lock = dir.path().join("t.write-lock");
5738 std::fs::write(&lock, "new-token").unwrap();
5739 assert!(!break_stale(&lock, "old-token"));
5740 assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
5741 assert!(!break_stale(&lock, "new-token"));
5743 assert!(lock.exists());
5744 age_lock(&lock);
5745 assert!(break_stale(&lock, "new-token"));
5746 assert!(!lock.exists());
5747 }
5748
5749 #[test]
5750 fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
5751 let dir = tempfile::tempdir().unwrap();
5752 let lock = dir.path().join("t.write-lock");
5753 std::fs::write(&lock, "dead-1").unwrap();
5754 age_lock(&lock);
5755 let marker = break_marker(&lock, "dead-1");
5756 std::fs::write(&marker, "crashed-remover").unwrap();
5757 assert!(!break_stale(&lock, "dead-1"));
5759 assert!(lock.exists());
5760 age_lock(&marker);
5762 assert!(break_stale(&lock, "dead-1"));
5763 assert!(!lock.exists());
5764 assert!(!marker.exists());
5765 }
5766
5767 fn chat_task() -> Task {
5768 Task::new(
5769 "t".to_owned(),
5770 "do t".to_owned(),
5771 PathBuf::from("."),
5772 Source::Agent {
5773 run: "20260902-000000-beef".to_owned(),
5774 node: CHAT_NODE.to_owned(),
5775 },
5776 )
5777 }
5778
5779 #[test]
5780 fn a_blocked_hold_keeps_its_anchor_but_a_new_run_starts_a_new_hold() {
5781 let mut t = chat_task();
5782 t.hold_machine(Some("waiting".into()));
5783 let old: Timestamp = "2020-01-01T00:00:00Z".parse().unwrap();
5784 t.held_at = Some(old);
5785 t.block(vec!["decision".into()], None);
5786 t.hold_machine(Some("same waiting decision".into()));
5787 assert_eq!(t.held_at, Some(old));
5788 t.start("real-new-run".into());
5789 t.fail("new failure", 1);
5790 assert_ne!(
5791 t.held_at,
5792 Some(old),
5793 "old answers cannot approve a new hold"
5794 );
5795 }
5796
5797 #[test]
5798 fn legacy_delivery_survives_upgrade_and_a_no_progress_run_link() {
5799 let (_dir, queue) = queue();
5800 let mut t = chat_task();
5801 t.start("old-run".into());
5802 t.hold_machine(Some("operator is considering recovery".into()));
5803 t.chat_notice = Some(ChatNotice {
5804 key: format!("held#{}", t.held_at.unwrap()),
5805 at: Timestamp::now(),
5806 outcome: ChatNoticeOutcome::Posted,
5807 });
5808 let mut raw = serde_json::to_value(&t).unwrap();
5809 raw.as_object_mut().unwrap().remove("chat_notices");
5810 raw.as_object_mut().unwrap().remove("hold_notice_cause");
5811 std::fs::create_dir_all(queue.path_of(&t.id).parent().unwrap()).unwrap();
5812 std::fs::write(queue.path_of(&t.id), serde_json::to_string(&raw).unwrap()).unwrap();
5813 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5814 queue
5815 .modify(&t.id, |task| {
5816 task.requeue();
5817 task.link_run("unprepared-run");
5818 task.startup_notice_cause("startup-source");
5819 task.fail("same unresolved preflight", 0);
5820 true
5821 })
5822 .unwrap();
5823 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5824 queue
5825 .modify(&t.id, |task| {
5826 task.succeed();
5827 true
5828 })
5829 .unwrap();
5830 assert_eq!(
5831 queue.get(&t.id).unwrap().chat_notice_due().as_deref(),
5832 Some("done")
5833 );
5834 }
5835
5836 #[test]
5837 fn a_dependency_wait_does_not_change_when_released_and_blocked_again() {
5838 let mut t = chat_task();
5839 t.block(vec!["a".into(), "b".into()], Some("waiting".into()));
5840 let key = t.chat_notice_key().unwrap();
5841 t.chat_notices.push(key);
5842 t.release();
5843 t.block(
5844 vec!["b".into(), "a".into(), "a".into()],
5845 Some("still waiting on prerequisites".into()),
5846 );
5847 assert_eq!(t.chat_notice_due(), None);
5848 t.block(
5849 vec!["new-question".into()],
5850 Some("different decision".into()),
5851 );
5852 assert!(t.chat_notice_due().is_some());
5853 }
5854
5855 #[test]
5856 fn running_startup_failures_preserve_stage_and_report_distinct_blockers() {
5857 let mut t = chat_task();
5858 t.start("existing-run".into());
5859 t.fail("implementation failed", 1);
5860 let failed = t.chat_notice_key().unwrap();
5861 t.chat_notices.push(failed.clone());
5862 t.release_operator("test", None);
5863 t.start("existing-run".into());
5864 t.fail_startup("startup-source", "same unresolved recovery source", 1);
5865 assert_eq!(
5866 t.chat_notice_due(),
5867 None,
5868 "no-progress recovery stays quiet"
5869 );
5870 t.release_operator("test", None);
5871 t.start("existing-run".into());
5872 t.fail_startup("startup-attachments", "attachment missing", 1);
5873 let attachments = t.chat_notice_due().expect("new attachment blocker");
5874 assert_ne!(attachments, failed);
5875 assert_eq!(
5876 t.hold_notice_cause.as_ref().unwrap().stage,
5877 "startup-attachments"
5878 );
5879 t.chat_notices.push(attachments);
5880 t.release_operator("test", None);
5881 t.start("existing-run".into());
5882 t.fail_startup(
5883 "startup-attachments",
5884 "same missing attachment, reworded",
5885 1,
5886 );
5887 assert_eq!(
5888 t.chat_notice_due(),
5889 None,
5890 "same stage after attempt counter reset"
5891 );
5892 t.release_operator("test", None);
5893 t.start("existing-run".into());
5894 t.fail_startup("startup-config", "different preflight stage", 1);
5895 assert!(t.chat_notice_due().is_some());
5896 assert_eq!(
5897 t.hold_notice_cause.as_ref().unwrap().stage,
5898 "startup-config"
5899 );
5900 }
5901
5902 #[test]
5903 fn structured_causes_preserve_no_progress_retries_and_report_new_passes() {
5904 let mut t = chat_task();
5905 t.hold_machine(Some("startup source missing".into()));
5906 t.startup_notice_cause("startup-source");
5907 let first = t.chat_notice_key().unwrap();
5908 t.chat_notices.push(first.clone());
5909 t.release_operator("test", None);
5910 t.startup_notice_cause("startup-source");
5911 t.hold_machine(Some("same source cannot be read".into()));
5912 assert_eq!(t.chat_notice_due(), None);
5913 t.startup_notice_cause("startup-config");
5914 assert!(t.chat_notice_due().is_some(), "different startup stage");
5915 t.notice_cause("failed", Some("existing-run"), Some(0));
5916 let run_failure = t.chat_notice_key().unwrap();
5917 t.chat_notices.push(run_failure.clone());
5918 t.release_operator("test", None);
5919 t.startup_notice_cause("startup-source");
5920 t.hold_machine(Some("no new work happened".into()));
5921 assert_eq!(t.chat_notice_due(), None);
5922 t.notice_cause("failed", Some("existing-run"), Some(1));
5923 assert!(t.chat_notice_due().is_some(), "a real resumed pass failed");
5924 assert_ne!(t.chat_notice_key().unwrap(), run_failure);
5925 }
5926
5927 #[test]
5928 fn legacy_hold_baseline_does_not_replay_or_hide_new_failures_and_completion() {
5929 let mut t = chat_task();
5930 t.hold_machine(Some("still waiting".into()));
5931 t.chat_notice = Some(ChatNotice {
5932 key: "held#2026-10-14T00:00:00Z".into(),
5933 at: Timestamp::now(),
5934 outcome: ChatNoticeOutcome::Posted,
5935 });
5936 assert_eq!(t.chat_notice_due(), None, "old hold baseline");
5937 t.import_chat_notice();
5938 t.notice_cause("failed", Some("new-run"), Some(0));
5939 assert!(t.chat_notice_due().is_some(), "new producer evidence");
5940 t.succeed();
5941 assert_eq!(t.chat_notice_due().as_deref(), Some("done"));
5942 }
5943
5944 #[test]
5945 fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
5946 let mut t = chat_task();
5947 assert_eq!(t.chat_notice_due(), None, "queued");
5948 t.start("r1".to_owned());
5949 assert_eq!(t.chat_notice_due(), None, "running");
5950 t.fail("boom", 3);
5951 assert_eq!(t.status, TaskStatus::Failed);
5952 assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
5953 t.start("r2".to_owned());
5954 t.park("upgrade");
5955 assert_eq!(t.chat_notice_due(), None, "parked");
5956 t.fail("boom", 1);
5957 assert_eq!(t.status, TaskStatus::Held);
5958 let key = t
5959 .chat_notice_due()
5960 .expect("held on the last attempt is final");
5961 t.chat_notice = Some(ChatNotice {
5962 key,
5963 at: Timestamp::now(),
5964 outcome: ChatNoticeOutcome::Posted,
5965 });
5966 assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
5967 t.release();
5968 assert_eq!(t.chat_notice_due(), None, "requeued after a release");
5969 t.start("r3".to_owned());
5970 t.succeed();
5971 assert!(t.chat_notice_due().is_some(), "done is a new final state");
5972 }
5973
5974 #[test]
5975 fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
5976 for key in ["done", "done#0", "done#1", "done#12"] {
5977 let mut t = chat_task();
5978 t.start("r1".to_owned());
5979 t.succeed();
5980 t.chat_notice = Some(ChatNotice {
5981 key: key.to_owned(),
5982 at: Timestamp::now(),
5983 outcome: ChatNoticeOutcome::Posted,
5984 });
5985 t.requeue();
5986 t.start("r2".to_owned());
5987 t.succeed();
5988 assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
5989 assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
5990 }
5991 }
5992
5993 #[test]
5994 fn a_release_without_progress_keeps_the_unresolved_notice_cause() {
5995 let mut t = chat_task();
5996 t.hold_manual(Some("wait".to_owned()));
5997 let first = t.chat_notice_due().expect("held");
5998 t.chat_notice = Some(ChatNotice {
5999 key: first.clone(),
6000 at: Timestamp::now(),
6001 outcome: ChatNoticeOutcome::Posted,
6002 });
6003 t.hold_manual(Some("still waiting".to_owned()));
6004 assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
6005 t.release();
6006 std::thread::sleep(std::time::Duration::from_millis(5));
6007 t.hold_manual(None);
6008 assert_eq!(t.chat_notice_due(), None, "no progress, same cause");
6009 }
6010
6011 #[test]
6012 fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
6013 let mut human = task("h");
6014 human.succeed();
6015 assert_eq!(human.chat_notice_due(), None);
6016 let mut inherited = task("i");
6018 inherited.origin_chat = Some("20260902-000000-beef".to_owned());
6019 inherited.succeed();
6020 assert_eq!(inherited.chat_notice_due(), None);
6021 let mut old = chat_task();
6022 old.schema = 12;
6023 old.status = TaskStatus::Done;
6024 assert_eq!(old.chat_notice_due(), None, "history is not replayed");
6025 let mut live = chat_task();
6026 live.schema = 12;
6027 live.start("r1".to_owned());
6028 live.succeed();
6029 assert!(live.chat_notice_due().is_some(), "started by this build");
6030 let mut held = chat_task();
6031 held.schema = 12;
6032 held.hold_machine(Some("config".to_owned()));
6033 assert!(held.chat_notice_due().is_some(), "held before any start");
6034 }
6035
6036 #[test]
6037 fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
6038 let tmp = tempfile::tempdir().expect("tempdir");
6039 let q = Queue::at(tmp.path().join("queue"));
6040 let mut t = chat_task();
6041 t.succeed();
6042 q.put(&mut t).expect("put");
6043 let mut stale = q.get(&t.id).expect("get");
6044 let key = t.chat_notice_due().expect("due");
6045 q.modify(&t.id, |x| {
6046 x.chat_notice = Some(ChatNotice {
6047 key: key.clone(),
6048 at: Timestamp::now(),
6049 outcome: ChatNoticeOutcome::Posted,
6050 });
6051 true
6052 })
6053 .expect("modify");
6054 q.put(&mut stale).expect("stale put");
6055 assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
6056 }
6057
6058 #[test]
6059 fn empty_sources_are_omitted_so_older_builds_still_read_the_task() {
6060 let mut t = task("plain");
6063 let value = serde_json::to_value(&t).unwrap();
6064 assert!(value.get("sources").is_none(), "{value}");
6065 t.sources.pick_commits.push(crate::refs::Input {
6066 token: "abc1234".into(),
6067 sha: "a".repeat(40),
6068 provenance: crate::refs::Provenance::ExplicitCli,
6069 });
6070 let value = serde_json::to_value(&t).unwrap();
6071 assert_eq!(value["sources"]["pick_commits"][0]["sha"], "a".repeat(40));
6072 let restored: Task = serde_json::from_value(value).unwrap();
6073 assert_eq!(restored.sources, t.sources);
6074 }
6075
6076 fn queue_with_ids(dir: &Path, ids: &[&str]) -> Queue {
6078 let ids: Vec<String> = ids.iter().map(|s| (*s).to_owned()).collect();
6079 let n = std::sync::atomic::AtomicUsize::new(0);
6080 Queue::at(dir.join("queue")).with_id_source(move || {
6081 let i = n.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
6082 ids.get(i)
6083 .cloned()
6084 .unwrap_or_else(|| format!("20260101-000000-f{i:03x}"))
6085 })
6086 }
6087
6088 const TAKEN: &str = "20260101-000000-aaaa";
6089
6090 #[test]
6091 fn a_new_task_never_merges_into_a_record_with_its_candidate_id() {
6092 let dir = tempfile::tempdir().unwrap();
6093 let q = queue_with_ids(dir.path(), &[TAKEN, TAKEN]);
6094 let mut first = task("first");
6095 first.start("run-1".into());
6096 first.succeed();
6097 q.put(&mut first).unwrap();
6098 assert_eq!(first.id, TAKEN);
6099 let stored_before = q.get(&first.id).unwrap();
6100 assert!(!stored_before.runs.is_empty());
6101
6102 let mut second = task("second");
6103 q.put(&mut second).unwrap();
6104
6105 assert_ne!(second.id, first.id);
6106 assert!(second.runs.is_empty());
6107 assert!(second.passes.is_empty());
6108 assert!(second.chat_notices.is_empty());
6109 assert!(second.chat_notice.is_none());
6110 assert_eq!(second.chat_notice_due(), task("fresh").chat_notice_due());
6111 let a = q.get(&first.id).unwrap();
6112 let b = q.get(&second.id).unwrap();
6113 assert_eq!(a.title, "first");
6114 assert_eq!(a.runs, stored_before.runs);
6115 assert_eq!(a.passes, stored_before.passes);
6116 assert_eq!(a.chat_notices, stored_before.chat_notices);
6117 assert_eq!(a.chat_notice, stored_before.chat_notice);
6118 assert_eq!(a.chat_notice_due(), stored_before.chat_notice_due());
6119 assert_eq!(b.title, "second");
6120 assert!(b.runs.is_empty());
6121 assert!(b.passes.is_empty());
6122 assert!(b.chat_notice.is_none());
6123
6124 let mut again = q.get(&second.id).unwrap();
6126 again.title = "second, edited".into();
6127 q.put(&mut again).unwrap();
6128 assert_eq!(q.get(&second.id).unwrap().title, "second, edited");
6129 assert_eq!(q.list().len(), 2);
6130 }
6131
6132 #[test]
6133 fn a_new_task_skips_an_id_with_an_orphaned_claim_or_attachments() {
6134 let dir = tempfile::tempdir().unwrap();
6135 let q = queue_with_ids(dir.path(), &["orphan-lock", "orphan-dir", "free"]);
6136 std::fs::create_dir_all(q.root()).unwrap();
6137 std::fs::write(q.root().join("orphan-lock.lock"), "x").unwrap();
6138 std::fs::create_dir_all(q.attachments_dir("orphan-dir")).unwrap();
6139 let mut t = task("t");
6140 q.put(&mut t).unwrap();
6141 assert_eq!(t.id, "free");
6142 }
6143
6144 #[test]
6145 fn a_claimed_new_task_holds_the_claim_on_the_reserved_id() {
6146 let dir = tempfile::tempdir().unwrap();
6147 let q = queue_with_ids(dir.path(), &[TAKEN, TAKEN]);
6148 let mut first = task("first");
6149 q.put(&mut first).unwrap();
6150 let mut second = task("second");
6151 let claim = q.file_claimed(&mut second).unwrap();
6152 assert_ne!(second.id, first.id);
6153 assert!(q.claim(&second.id).is_err());
6154 assert!(q.claim(&first.id).is_ok(), "the first task was not claimed");
6155 drop(claim);
6156 }
6157
6158 #[test]
6159 fn an_attached_new_task_skips_a_taken_id() {
6160 let dir = tempfile::tempdir().unwrap();
6161 let q = queue_with_ids(dir.path(), &[TAKEN, TAKEN]);
6162 let mut first = task("first");
6163 q.put(&mut first).unwrap();
6164 let src = source_file(dir.path(), "shot.png", "x");
6165 let mut second = task("second");
6166 q.attach_and_put(&mut second, &[src]).unwrap();
6167 assert_ne!(second.id, first.id);
6168 assert!(!q.attachments_dir(&first.id).exists());
6169 assert_eq!(q.attachment_paths(&second).len(), 1);
6170 assert!(q.attachment_paths(&second)[0].is_file());
6171 }
6172
6173 #[test]
6174 fn attaching_before_putting_a_new_task_reserves_the_id() {
6175 let dir = tempfile::tempdir().unwrap();
6176 let q = queue_with_ids(dir.path(), &[TAKEN, TAKEN]);
6177 let mut first = task("first");
6178 q.put(&mut first).unwrap();
6179 let src = source_file(dir.path(), "shot.png", "x");
6180 let mut second = task("second");
6181 q.attach(&mut second, &[src]).unwrap();
6182 assert_ne!(second.id, first.id);
6183 assert!(q.attachments_dir(&second.id).is_dir());
6184 q.put(&mut second).unwrap();
6185 assert_eq!(q.get(&second.id).unwrap().attachments.len(), 1);
6186 assert_eq!(q.get(&first.id).unwrap().title, "first");
6187 assert!(q.get(&first.id).unwrap().attachments.is_empty());
6188 }
6189
6190 #[test]
6191 fn create_new_keeps_its_already_exists_meaning() {
6192 let dir = tempfile::tempdir().unwrap();
6193 let q = Queue::at(dir.path().join("queue"));
6194 let mut a = task("a");
6195 a.id = "fixed-id".into();
6196 assert!(q.create_new(&mut a).unwrap());
6197 let mut b = task("b");
6198 b.id = "fixed-id".into();
6199 assert!(!q.create_new(&mut b).unwrap());
6200 assert_eq!(q.get("fixed-id").unwrap().title, "a");
6201 }
6202
6203 #[test]
6204 fn filing_gives_up_when_no_id_is_ever_free() {
6205 let dir = tempfile::tempdir().unwrap();
6206 let q = Queue::at(dir.path().join("queue")).with_id_source(|| TAKEN.to_owned());
6207 let mut first = task("first");
6208 q.put(&mut first).unwrap();
6209 let mut second = task("second");
6210 assert!(q.put(&mut second).is_err());
6211 assert_eq!(q.get(&first.id).unwrap().title, "first");
6212 }
6213}