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