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;
42
43pub const SCHEMA: u32 = 20;
134
135pub(crate) fn dispatch_order(a: &Task, b: &Task) -> std::cmp::Ordering {
141 match (a.upgrade_resume_key(), b.upgrade_resume_key()) {
142 (Some(a), Some(b)) => a.cmp(&b),
143 (Some(_), None) => std::cmp::Ordering::Less,
144 (None, Some(_)) => std::cmp::Ordering::Greater,
145 (None, None) => b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)),
146 }
147}
148
149#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
151#[serde(rename_all = "lowercase")]
152pub enum HoldSource {
153 Manual,
155 Machine,
157}
158
159impl HoldSource {
160 pub fn label(self) -> &'static str {
162 match self {
163 Self::Manual => "manual",
164 Self::Machine => "machine",
165 }
166 }
167}
168
169#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
172#[serde(tag = "kind", rename_all = "lowercase")]
173pub enum Source {
174 Human,
176 Agent {
179 run: String,
181 node: String,
183 },
184 Issue {
186 number: u64,
188 repo: String,
190 },
191}
192
193pub const CHAT_NODE: &str = "chat";
196
197impl Source {
198 pub fn label(&self) -> String {
200 match self {
201 Self::Human => "human".to_owned(),
202 Self::Agent { run, node } => format!("{node}@{}", short(run)),
203 Self::Issue { number, .. } => format!("issue #{number}"),
204 }
205 }
206}
207
208#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
210#[serde(rename_all = "lowercase")]
211pub enum TaskStatus {
212 Queued,
214 Running,
216 Done,
218 Failed,
220 Held,
222 Blocked,
226 Parked,
231}
232
233impl TaskStatus {
234 pub fn runnable(self) -> bool {
236 matches!(self, Self::Queued | Self::Failed | Self::Parked)
237 }
238
239 pub fn as_str(self) -> &'static str {
241 match self {
242 Self::Queued => "queued",
243 Self::Running => "running",
244 Self::Done => "done",
245 Self::Failed => "failed",
246 Self::Held => "held",
247 Self::Blocked => "blocked",
248 Self::Parked => "parked",
249 }
250 }
251}
252
253#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
256pub struct TaskCounts {
257 pub queued: usize,
259 pub running: usize,
261 pub done: usize,
263 pub failed: usize,
265 pub held: usize,
267 pub blocked: usize,
269 pub parked: usize,
271}
272
273impl TaskCounts {
274 pub fn of(tasks: &[Task]) -> Self {
278 Self::of_statuses(tasks.iter().map(|task| task.status))
279 }
280
281 pub(crate) fn of_statuses(statuses: impl IntoIterator<Item = TaskStatus>) -> Self {
283 let mut counts = Self::default();
284 for status in statuses {
285 match status {
286 TaskStatus::Queued => counts.queued += 1,
287 TaskStatus::Running => counts.running += 1,
288 TaskStatus::Done => counts.done += 1,
289 TaskStatus::Failed => counts.failed += 1,
290 TaskStatus::Held => counts.held += 1,
291 TaskStatus::Blocked => counts.blocked += 1,
292 TaskStatus::Parked => counts.parked += 1,
293 }
294 }
295 counts
296 }
297}
298
299#[derive(Debug, Clone, Serialize, Deserialize)]
301#[serde(deny_unknown_fields)]
302pub struct Task {
303 pub schema: u32,
305 pub id: String,
307 pub title: String,
309 pub instruction: String,
311 #[serde(default, skip_serializing_if = "crate::refs::Sources::is_empty")]
313 pub sources: crate::refs::Sources,
314 pub repo: PathBuf,
316 pub source: Source,
318 #[serde(default)]
320 pub chat: Option<String>,
321 #[serde(default)]
323 pub priority: i32,
324 #[serde(default)]
335 pub solo: bool,
336 pub status: TaskStatus,
338 #[serde(default)]
340 pub attempts: usize,
341 #[serde(default)]
343 pub attempt_limit: Option<NonZeroUsize>,
344 #[serde(default)]
347 pub runs: Vec<String>,
348 #[serde(default)]
350 pub last_error: Option<String>,
351 #[serde(default)]
364 pub hold_reason: Option<String>,
365 #[serde(default)]
368 pub hold_source: Option<HoldSource>,
369 #[serde(default)]
381 pub diagnostic: Option<String>,
382 #[serde(default)]
392 pub blocked_by: Vec<String>,
393 #[serde(default)]
396 pub block_reason: Option<String>,
397 #[serde(default)]
410 pub blocked_from: Option<TaskStatus>,
411 #[serde(default)]
413 pub blocked_at: Option<Timestamp>,
414 #[serde(default)]
425 pub answers: Vec<AnsweredQuestion>,
426 #[serde(default)]
432 pub triage_applied: Vec<String>,
433 #[serde(default)]
439 pub actions_applied: Vec<String>,
440 #[serde(default)]
445 pub resume_override: Option<OperatorResume>,
446 #[serde(default)]
449 pub resumption: Option<Resumption>,
450 #[serde(default)]
452 pub resumption_history: Vec<Resumption>,
453 #[serde(default)]
455 pub operator_decided_at: Option<Timestamp>,
456 #[serde(default)]
468 pub review_branch: Option<String>,
469 #[serde(default)]
471 pub review_instruction: Option<String>,
472 #[serde(default)]
474 pub pr_recovery: Option<crate::pr_recovery::Selection>,
475 #[serde(default)]
478 pub fresh_start: bool,
479 #[serde(default)]
481 pub replacement: Option<crate::merge_decision::Replacement>,
482 #[serde(default)]
493 pub interrupt: bool,
494 #[serde(default)]
517 pub urgent: bool,
518 #[serde(skip)]
523 pub(crate) unfiled: bool,
524 #[serde(default)]
530 pub attachments: Vec<String>,
531 #[serde(default)]
535 pub followup: Option<FollowUp>,
536 #[serde(default)]
542 pub origin_chat: Option<String>,
543 #[serde(default)]
550 pub overrides: Option<RunOverrides>,
551 #[serde(default)]
558 pub review_of: Option<String>,
559 #[serde(default)]
564 pub held_at: Option<Timestamp>,
565 #[serde(default)]
569 pub park_reason: Option<String>,
570 #[serde(default)]
573 pub chat_notice: Option<ChatNotice>,
574 #[serde(default)]
576 pub chat_notices: Vec<String>,
577 #[serde(default)]
580 pub hold_notice_cause: Option<NoticeCause>,
581 #[serde(default)]
586 pub passes: Vec<TaskPass>,
587 #[serde(default)]
591 pub dispatched_at: Option<Timestamp>,
592 #[serde(default)]
594 pub recovery_budget: Option<RecoveryBudget>,
595 #[serde(default)]
598 pub recovery_admission: Option<RecoveryAdmission>,
599 #[serde(default)]
602 pub recovery_generations: Vec<RecoveryBudget>,
603 pub created_at: Timestamp,
605 pub updated_at: Timestamp,
607}
608
609#[derive(Debug, Clone, Serialize, Deserialize)]
611pub struct RecoveryAdmission {
612 pub base: String,
614 pub inputs: String,
616}
617
618#[derive(Debug, Clone, Serialize, Deserialize)]
621pub struct RecoveryBudget {
622 pub instruction: String,
624 pub base: String,
626 pub inputs: String,
628 pub stage: String,
630 pub failures: usize,
632 #[serde(default)]
635 pub progress: Option<String>,
636}
637
638fn recovery_progress(state: &crate::run::RunState) -> Option<String> {
639 use crate::proc::Quiet as _;
640 use std::hash::{Hash, Hasher};
641 let mut patches = Vec::new();
642 for candidate in state.candidates.iter().filter(|c| c.commits > 0) {
643 let output = std::process::Command::new("git")
644 .current_dir(&state.repo)
645 .args(["diff", "--binary", &state.base_commit, &candidate.branch])
646 .quiet()
647 .output()
648 .ok()?;
649 if !output.status.success() {
650 return None;
651 }
652 patches.push(output.stdout);
653 }
654 patches.sort();
655 patches.dedup();
656 let mut hash = std::collections::hash_map::DefaultHasher::new();
657 patches.hash(&mut hash);
658 Some(format!("{:x}", hash.finish()))
659}
660
661pub(crate) fn recovery_inputs(sources: &crate::refs::Sources) -> String {
665 serde_json::to_string(
666 &sources
667 .start_from
668 .iter()
669 .map(|i| ("start", &i.token, &i.sha))
670 .chain(
671 sources
672 .pick_commits
673 .iter()
674 .map(|i| ("pick", &i.token, &i.sha)),
675 )
676 .collect::<Vec<_>>(),
677 )
678 .unwrap_or_default()
679}
680
681#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
683pub struct NoticeCause {
684 pub stage: String,
686 #[serde(default, skip_serializing_if = "Option::is_none")]
689 pub target: Option<String>,
690 pub run: Option<String>,
692 pub pass: Option<usize>,
694}
695
696pub(crate) fn is_done_notice_key(key: &str) -> bool {
698 key == "done"
699 || key
700 .strip_prefix("done#")
701 .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
702}
703
704#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
706#[serde(rename_all = "snake_case")]
707pub enum PassCost {
708 Spent,
710 Refunded,
712}
713
714#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
716pub struct TaskPass {
717 pub run: String,
719 pub pass_index: usize,
721 pub cost: PassCost,
723 pub attempts_after: usize,
725 #[serde(default)]
727 pub started_at: Option<Timestamp>,
728 pub settled_at: Timestamp,
730 #[serde(default)]
732 pub parked: Option<crate::run::ParkCause>,
733}
734
735#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
737#[serde(rename_all = "lowercase")]
738pub enum ChatNoticeOutcome {
739 Posted,
741 Skipped,
743}
744
745#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
747pub struct ChatNotice {
748 pub key: String,
750 pub at: Timestamp,
752 pub outcome: ChatNoticeOutcome,
754}
755
756#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
758pub struct RunOverrides {
759 #[serde(default)]
762 pub inherit_daemon_options: bool,
763 #[serde(default)]
765 pub merge: Option<String>,
766 #[serde(default)]
768 pub candidates: Option<usize>,
769 #[serde(default)]
771 pub judges: Option<usize>,
772 #[serde(default)]
774 pub reviewers: Option<usize>,
775 #[serde(default)]
777 pub review_rounds: Option<usize>,
778 #[serde(default)]
780 pub seed: Option<u64>,
781 #[serde(default)]
783 pub config: Option<PathBuf>,
784}
785
786impl RunOverrides {
787 pub fn merge_over(&mut self, other: &RunOverrides) {
789 macro_rules! take {
790 ($($f:ident),*) => {$(
791 if other.$f.is_some() {
792 self.$f.clone_from(&other.$f);
793 }
794 )*};
795 }
796 take!(
797 merge,
798 candidates,
799 judges,
800 reviewers,
801 review_rounds,
802 seed,
803 config
804 );
805 }
806
807 pub fn apply(&self, config: &mut crate::config::Config) {
809 if let Some(n) = self.candidates {
810 config.graph.implementers = n;
811 }
812 if let Some(n) = self.judges {
813 config.graph.judges = n;
814 }
815 if let Some(n) = self.reviewers {
816 config.graph.reviewers = n;
817 }
818 if let Some(n) = self.review_rounds {
819 config.graph.review_rounds = n;
820 }
821 if let Some(m) = &self.merge
822 && let Ok(mode) = crate::daemon::merge_mode(m)
823 {
824 config.merge.mode = mode;
825 }
826 if let Some(s) = self.seed {
827 config.blind.seed = Some(s);
828 }
829 }
830}
831
832#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
834pub struct FollowUp {
835 pub run: String,
837 #[serde(default)]
839 pub origin_task: Option<String>,
840 pub pr: String,
842 pub findings: Vec<String>,
844 pub generation: u32,
847}
848
849#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
851pub struct Resumption {
852 pub generation: u64,
854 pub source: String,
856 pub question_id: Option<String>,
858 pub at: Timestamp,
860 pub instruction: String,
862 pub answers_through: usize,
864 pub runs_through: usize,
867 pub cleared_reason: Option<String>,
869 #[serde(default)]
871 pub blocking_evidence: Vec<String>,
872}
873
874#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
877pub struct OperatorResume {
878 pub question_id: String,
880 pub at: Timestamp,
882 #[serde(default)]
885 pub conductor_rehold: Option<String>,
886 #[serde(default)]
889 pub forced: bool,
890 #[serde(default)]
897 pub pinned_run: Option<String>,
898}
899
900#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
903pub struct AnsweredQuestion {
904 pub question: String,
906 pub answer: String,
908}
909
910impl Task {
911 pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
913 let now = Timestamp::now();
914 let origin_chat = match &source {
915 Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
916 _ => None,
917 };
918 Self {
919 schema: SCHEMA,
920 id: new_id(),
921 title,
922 instruction,
923 sources: crate::refs::Sources::default(),
924 repo,
925 source,
926 chat: None,
927 priority: 0,
928 solo: false,
929 status: TaskStatus::Queued,
930 attempts: 0,
931 attempt_limit: None,
932 runs: Vec::new(),
933 last_error: None,
934 hold_reason: None,
935 hold_source: None,
936 diagnostic: None,
937 blocked_by: Vec::new(),
938 block_reason: None,
939 blocked_from: None,
940 blocked_at: None,
941 answers: Vec::new(),
942 triage_applied: Vec::new(),
943 actions_applied: Vec::new(),
944 resume_override: None,
945 resumption: None,
946 resumption_history: Vec::new(),
947 operator_decided_at: None,
948 review_branch: None,
949 review_instruction: None,
950 pr_recovery: None,
951 fresh_start: false,
952 replacement: None,
953 interrupt: false,
954 urgent: false,
955 unfiled: true,
956 attachments: Vec::new(),
957 followup: None,
958 origin_chat,
959 overrides: None,
960 review_of: None,
961 held_at: None,
962 park_reason: None,
963 chat_notice: None,
964 chat_notices: Vec::new(),
965 hold_notice_cause: None,
966 passes: Vec::new(),
967 dispatched_at: None,
968 recovery_budget: None,
969 recovery_admission: None,
970 recovery_generations: Vec::new(),
971 created_at: now,
972 updated_at: now,
973 }
974 }
975
976 pub fn chat_talk(&self) -> Option<&str> {
979 if let Some(id) = &self.origin_chat {
980 return Some(id);
981 }
982 match &self.source {
983 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
984 _ => None,
985 }
986 }
987
988 pub fn filed_by_chat(&self) -> Option<&str> {
992 match &self.source {
993 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
994 _ => None,
995 }
996 }
997
998 pub fn chat_notice_key(&self) -> Option<String> {
1001 match self.status {
1002 TaskStatus::Done => Some("done".to_owned()),
1003 TaskStatus::Held => {
1004 let cause = self
1005 .hold_notice_cause
1006 .clone()
1007 .unwrap_or_else(|| self.unresolved_hold_cause());
1008 Some(format!(
1009 "cause:{}",
1010 serde_json::to_string(&cause).expect("cause")
1011 ))
1012 }
1013 TaskStatus::Blocked => {
1014 let ids: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
1015 let evidence = self
1018 .hold_notice_cause
1019 .as_ref()
1020 .filter(|c| c.stage.starts_with("conductor-evidence:"));
1021 Some(format!(
1022 "dependencies:{}",
1023 serde_json::to_string(&(ids, evidence)).expect("ids")
1024 ))
1025 }
1026 _ => None,
1027 }
1028 }
1029
1030 fn unresolved_hold_cause(&self) -> NoticeCause {
1031 NoticeCause {
1032 stage: "hold".into(),
1033 target: None,
1034 run: self.runs.last().cloned(),
1035 pass: None,
1036 }
1037 }
1038
1039 pub(crate) fn legacy_chat_notice_key(&self) -> Option<String> {
1041 let key = match self.status {
1042 TaskStatus::Held if self.hold_notice_cause.is_none() => Some(format!(
1043 "held#{}",
1044 self.held_at.map(|t| t.to_string()).unwrap_or_default()
1045 )),
1046 TaskStatus::Blocked => Some(format!(
1047 "blocked#{}#{}",
1048 self.blocked_by.join(","),
1049 self.block_reason.as_deref().unwrap_or_default()
1050 )),
1051 _ => None,
1052 };
1053 key.filter(|key| self.legacy_notice_matches(key))
1054 }
1055
1056 pub(crate) fn legacy_notice_matches(&self, key: &str) -> bool {
1057 match self.status {
1058 TaskStatus::Held if self.hold_notice_cause.is_none() => key
1059 .strip_prefix("held#")
1060 .is_some_and(|time| time.is_empty() || time.parse::<Timestamp>().is_ok()),
1061 TaskStatus::Blocked
1062 if self
1063 .hold_notice_cause
1064 .as_ref()
1065 .is_none_or(|c| !c.stage.starts_with("conductor-evidence:")) =>
1066 {
1067 key.strip_prefix("blocked#").is_some_and(|tail| {
1068 let Some((ids, _)) = tail.split_once('#') else {
1069 return false;
1070 };
1071 let old: std::collections::BTreeSet<_> =
1072 ids.split(',').filter(|id| !id.is_empty()).collect();
1073 let current: std::collections::BTreeSet<_> =
1074 self.blocked_by.iter().map(String::as_str).collect();
1075 old == current
1076 })
1077 }
1078 _ => false,
1079 }
1080 }
1081
1082 fn import_chat_notice(&mut self) {
1083 if let Some(n) = &self.chat_notice {
1084 let key = if is_done_notice_key(&n.key) {
1085 "done".to_owned()
1086 } else if self.legacy_notice_matches(&n.key) {
1087 self.chat_notice_key().unwrap_or_else(|| n.key.clone())
1088 } else {
1089 n.key.clone()
1090 };
1091 if !self.chat_notices.contains(&key) {
1092 self.chat_notices.push(key);
1093 }
1094 }
1095 }
1096
1097 pub(crate) fn notice_cause(&mut self, stage: &str, run: Option<&str>, pass: Option<usize>) {
1100 self.import_chat_notice();
1101 self.hold_notice_cause = Some(NoticeCause {
1102 stage: stage.to_owned(),
1103 target: None,
1104 run: run.map(str::to_owned),
1105 pass,
1106 });
1107 }
1108
1109 pub(crate) fn startup_notice_cause(&mut self, stage: &str) {
1112 let recovery_only = matches!(
1113 stage,
1114 "startup-source" | "startup-candidates" | "startup-public-head"
1115 );
1116 match self.hold_notice_cause.as_ref() {
1117 Some(cause) if recovery_only && !cause.stage.starts_with("startup-") => return,
1118 None if recovery_only
1119 && self
1120 .chat_notice
1121 .as_ref()
1122 .is_some_and(|n| !is_done_notice_key(&n.key)) =>
1123 {
1124 return;
1125 }
1126 _ => {}
1127 }
1128 self.notice_cause(stage, None, None);
1129 }
1130
1131 pub(crate) fn fail_startup(
1134 &mut self,
1135 stage: &str,
1136 why: impl Into<String>,
1137 max_attempts: usize,
1138 ) {
1139 let unresolved = self.hold_notice_cause.clone();
1140 self.fail(why, max_attempts);
1141 self.hold_notice_cause = unresolved;
1142 self.startup_notice_cause(stage);
1143 }
1144
1145 pub(crate) fn fail_startup_target(
1147 &mut self,
1148 stage: &str,
1149 target: String,
1150 why: impl Into<String>,
1151 max_attempts: usize,
1152 ) {
1153 self.fail_startup(stage, why, max_attempts);
1154 if let Some(cause) = self.hold_notice_cause.as_mut() {
1155 cause.target = Some(target);
1156 }
1157 }
1158
1159 pub fn chat_notice_due(&self) -> Option<String> {
1161 self.filed_by_chat()?;
1162 if self.schema < 13 {
1163 return None;
1164 }
1165 let key = self.chat_notice_key()?;
1166 if self.chat_notices.contains(&key) {
1167 return None;
1168 }
1169 match &self.chat_notice {
1170 Some(n)
1171 if n.key == key
1172 || (key == "done" && is_done_notice_key(&n.key))
1173 || self.legacy_notice_matches(&n.key) =>
1174 {
1175 None
1176 }
1177 _ => Some(key),
1178 }
1179 }
1180
1181 pub fn short(&self) -> &str {
1183 short(&self.id)
1184 }
1185
1186 pub fn mark_triage_applied(&mut self, question_id: &str) {
1188 if !self.triage_applied(question_id) {
1189 self.triage_applied.push(question_id.to_owned());
1190 }
1191 }
1192
1193 pub fn action_applied(&self, question_id: &str) -> bool {
1196 self.actions_applied.iter().any(|id| id == question_id)
1197 }
1198
1199 pub fn mark_action_applied(&mut self, question_id: &str) {
1201 if !self.action_applied(question_id) {
1202 self.actions_applied.push(question_id.to_owned());
1203 }
1204 }
1205
1206 pub fn triage_applied(&self, question_id: &str) -> bool {
1208 self.triage_applied.iter().any(|id| id == question_id)
1209 }
1210
1211 pub fn start(&mut self, run: String) {
1224 self.stamp_schema();
1225 self.status = TaskStatus::Running;
1226 self.attempts += 1;
1227 self.runs.retain(|id| id != &run);
1228 self.runs.push(run);
1229 self.last_error = None;
1230 self.park_reason = None;
1231 self.fresh_start = false;
1232 self.interrupt = false;
1233 self.resume_override = None;
1235 }
1236
1237 pub fn link_run(&mut self, run: &str) -> bool {
1242 if self.runs.iter().any(|r| r == run) {
1243 return false;
1244 }
1245 self.runs.push(run.to_owned());
1246 true
1247 }
1248
1249 pub fn succeed(&mut self) {
1259 self.stamp_schema();
1260 self.status = TaskStatus::Done;
1261 self.resume_override = None;
1262 self.last_error = None;
1263 self.park_reason = None;
1264 self.hold_reason = None;
1265 self.hold_source = None;
1266 self.diagnostic = None;
1267 self.blocked_by.clear();
1268 self.block_reason = None;
1269 self.blocked_from = None;
1270 }
1271
1272 pub fn already_landed(&mut self, note: impl Into<String>) {
1280 self.succeed();
1281 self.attempts = self.attempts.saturating_sub(1);
1282 self.last_error = Some(note.into());
1283 }
1284
1285 pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
1315 if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
1316 return &[];
1317 }
1318 &self.runs[..self.runs.len() - 1]
1319 }
1320
1321 pub fn successor_of(&self, run: &str) -> Option<&String> {
1329 let pos = self.runs.iter().rposition(|r| r == run)?;
1330 self.runs.get(pos + 1)
1331 }
1332
1333 pub fn earlier_attempts(&self) -> &[String] {
1340 &self.runs
1341 }
1342
1343 fn stamp_schema(&mut self) {
1350 self.schema = self.schema.max(SCHEMA);
1351 }
1352
1353 fn note_held(&mut self) {
1354 self.stamp_schema();
1355 let continuing = self.status == TaskStatus::Held
1356 || (self.status == TaskStatus::Blocked
1357 && (self.blocked_from == Some(TaskStatus::Held)
1358 || (self.blocked_from.is_none()
1359 && (self.hold_reason.is_some() || self.hold_source.is_some()))));
1360 if !continuing || self.held_at.is_none() {
1361 self.held_at = Some(Timestamp::now());
1362 }
1363 }
1364
1365 pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
1381 let why = why.into();
1382 if self.status == TaskStatus::Running {
1383 let run = self.runs.last().cloned();
1384 self.notice_cause("run-failed", run.as_deref(), Some(self.attempts));
1385 }
1386 self.diagnostic = None;
1387 self.park_reason = None;
1388 self.status = if self.attempts >= self.effective_attempt_limit(max_attempts) {
1389 self.note_held();
1390 self.hold_source = Some(HoldSource::Machine);
1391 self.hold_reason = Some(why.clone());
1392 TaskStatus::Held
1393 } else {
1394 TaskStatus::Failed
1395 };
1396 self.last_error = Some(why);
1397 }
1398
1399 pub fn stall(&mut self, why: impl Into<String>) {
1409 self.last_error = Some(why.into());
1410 self.diagnostic = None;
1411 self.attempts = self.attempts.saturating_sub(1);
1412 self.status = TaskStatus::Failed;
1413 }
1414
1415 pub fn park(&mut self, why: impl Into<String>) {
1421 self.diagnostic = None;
1422 self.attempts = self.attempts.saturating_sub(1);
1423 self.status = TaskStatus::Parked;
1424 self.park_reason = Some(why.into());
1425 self.schema = self.schema.max(SCHEMA);
1428 }
1429
1430 pub fn record_pass(
1432 &mut self,
1433 run: &str,
1434 pass_index: usize,
1435 attempts_before: usize,
1436 parked: Option<crate::run::ParkCause>,
1437 ) {
1438 self.record_pass_at(run, pass_index, None, attempts_before, parked);
1439 }
1440
1441 pub fn record_pass_at(
1444 &mut self,
1445 run: &str,
1446 pass_index: usize,
1447 started_at: Option<Timestamp>,
1448 attempts_before: usize,
1449 parked: Option<crate::run::ParkCause>,
1450 ) {
1451 if self
1452 .passes
1453 .iter()
1454 .any(|p| p.run == run && p.pass_index == pass_index)
1455 {
1456 return;
1457 }
1458 let cost = if self.attempts < attempts_before {
1459 PassCost::Refunded
1460 } else {
1461 PassCost::Spent
1462 };
1463 self.passes.push(TaskPass {
1464 run: run.to_owned(),
1465 pass_index,
1466 started_at: self.dispatched_at.or(started_at),
1467 cost,
1468 attempts_after: self.attempts,
1469 settled_at: Timestamp::now(),
1470 parked,
1471 });
1472 self.schema = self.schema.max(SCHEMA);
1473 }
1474
1475 pub(crate) fn upgrade_resume_key(&self) -> Option<(Timestamp, &str)> {
1478 if self.status != TaskStatus::Parked || self.fresh_start {
1479 return None;
1480 }
1481 let pass = self.passes.last()?;
1482 if self.runs.last() != Some(&pass.run)
1483 || !matches!(pass.parked, Some(crate::run::ParkCause::Upgrade { .. }))
1484 {
1485 return None;
1486 }
1487 Some((pass.started_at.unwrap_or(pass.settled_at), &self.id))
1490 }
1491
1492 pub fn operator_held(&self) -> bool {
1498 self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
1499 }
1500
1501 pub fn hold_manual(&mut self, reason: Option<String>) {
1511 self.operator_decided_at = Some(Timestamp::now());
1513 self.hold_unrequested_retry(reason);
1514 }
1515
1516 pub(crate) fn hold_unrequested_retry(&mut self, reason: Option<String>) {
1519 self.park_reason = None;
1520 self.note_held();
1521 self.status = TaskStatus::Held;
1522 if reason.is_some() {
1523 self.hold_reason = reason;
1524 }
1525 self.hold_source = Some(HoldSource::Manual);
1526 self.blocked_by.clear();
1527 self.block_reason = None;
1528 self.blocked_from = None;
1529 }
1530
1531 pub fn hold_machine(&mut self, reason: Option<String>) {
1536 self.park_reason = None;
1537 self.note_held();
1538 self.status = TaskStatus::Held;
1539 if reason.is_some() {
1540 self.hold_reason = reason;
1541 }
1542 self.hold_source = Some(HoldSource::Machine);
1543 self.blocked_by.clear();
1544 self.block_reason = None;
1545 self.blocked_from = None;
1546 }
1547
1548 pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
1558 self.stamp_schema();
1559 let before: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
1560 let after: std::collections::BTreeSet<_> = blocked_by.iter().collect();
1561 if self.status != TaskStatus::Blocked || before != after {
1562 self.blocked_at = Some(Timestamp::now());
1563 }
1564 if self.status != TaskStatus::Blocked {
1565 self.blocked_from = Some(self.status);
1566 }
1567 self.status = TaskStatus::Blocked;
1568 self.blocked_by = blocked_by;
1569 self.block_reason = reason;
1570 }
1571
1572 pub fn unblock(&mut self, resolved_id: &str) {
1595 if self.status != TaskStatus::Blocked {
1596 return;
1597 }
1598 self.blocked_by.retain(|id| id != resolved_id);
1599 self.restore_if_unblocked();
1600 }
1601
1602 pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
1613 if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
1614 return false;
1615 }
1616 self.blocked_by.retain(|id| id != deleted_id);
1617 self.restore_if_unblocked();
1618 true
1619 }
1620
1621 fn restore_if_unblocked(&mut self) {
1625 if self.blocked_by.is_empty() {
1626 self.status = match self.blocked_from {
1627 Some(TaskStatus::Running) => TaskStatus::Queued,
1628 Some(other) => other,
1629 None if self.hold_reason.is_some() || self.hold_source.is_some() => {
1630 TaskStatus::Held
1631 }
1632 None => TaskStatus::Queued,
1633 };
1634 self.block_reason = None;
1635 self.blocked_from = None;
1636 }
1637 }
1638
1639 pub fn record_answer(&mut self, question: String, answer: String) {
1644 self.answers.push(AnsweredQuestion { question, answer });
1645 }
1646
1647 pub fn request_review(&mut self, branch: String) {
1651 self.release();
1652 self.review_branch = Some(branch);
1653 self.review_instruction = Some(self.instruction.clone());
1654 }
1655
1656 pub fn requeue(&mut self) {
1659 self.release();
1660 self.review_branch = None;
1662 self.review_instruction = None;
1663 self.fresh_start = true;
1664 }
1665
1666 pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
1675 if branch.is_some() {
1676 self.review_branch = branch;
1677 self.review_instruction = Some(self.instruction.clone());
1678 }
1679 self.hold_machine(Some(reason));
1680 }
1681
1682 pub fn set_urgent(&mut self, urgent: bool) -> Result<()> {
1684 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1685 bail!(
1686 "task {} is {}; urgent can only change while queued or held; \
1687 wait for the active attempt to finish and requeue or hold the task first",
1688 self.short(),
1689 self.status.as_str()
1690 );
1691 }
1692 self.urgent = urgent;
1693 Ok(())
1694 }
1695
1696 pub fn set_priority(&mut self, priority: i32) -> Result<()> {
1705 if self.status == TaskStatus::Running {
1706 bail!(
1707 "task {} is running; its priority cannot be changed until \
1708 this attempt finishes",
1709 self.short()
1710 );
1711 }
1712 self.priority = priority;
1713 Ok(())
1714 }
1715
1716 pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
1731 if interrupt && !self.status.runnable() {
1732 bail!(
1733 "task {} is {}; only a queued or failed task can be marked \
1734 to interrupt",
1735 self.short(),
1736 self.status.as_str()
1737 );
1738 }
1739 self.interrupt = interrupt;
1740 Ok(())
1741 }
1742
1743 pub fn effective_attempt_limit(&self, daemon_limit: usize) -> usize {
1745 self.attempt_limit.map_or(daemon_limit, NonZeroUsize::get)
1746 }
1747
1748 pub fn change_attempt_limit(&mut self, limit: Option<NonZeroUsize>) -> Result<()> {
1751 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1752 bail!("only queued or held tasks can change the attempt limit");
1753 }
1754 self.stamp_schema();
1755 self.attempt_limit = limit;
1756 Ok(())
1757 }
1758
1759 pub fn implementers(&self) -> Option<usize> {
1761 if self.solo {
1762 Some(1)
1763 } else {
1764 self.overrides.as_ref().and_then(|o| o.candidates)
1765 }
1766 }
1767
1768 pub fn change_implementers(&mut self, count: u8) -> Result<()> {
1771 self.check_implementers_editable()?;
1772 let explicit = self.overrides.as_ref().and_then(|o| o.config.as_deref());
1773 let (mut config, _) = crate::config::Config::discover(&self.repo, explicit)?;
1774 if let Some(o) = &self.overrides {
1775 o.apply(&mut config);
1776 }
1777 self.set_implementers(count, &config)
1778 }
1779
1780 fn check_implementers_editable(&self) -> Result<()> {
1781 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1782 bail!(
1783 "task {} is {}; only queued or held tasks can change implementers",
1784 self.short(),
1785 self.status.as_str()
1786 );
1787 }
1788 Ok(())
1789 }
1790
1791 pub fn set_implementers(&mut self, count: u8, config: &crate::config::Config) -> Result<()> {
1793 self.check_implementers_editable()?;
1794 crate::talk::check_implementers(count, config).map_err(anyhow::Error::msg)?;
1795 self.stamp_schema();
1796 self.solo = count == 1;
1797 let mut overrides = self.overrides.clone().unwrap_or_else(|| RunOverrides {
1798 inherit_daemon_options: true,
1799 ..Default::default()
1800 });
1801 overrides.candidates = (count > 1).then_some(usize::from(count));
1802 let mut settings = overrides.clone();
1803 settings.inherit_daemon_options = false;
1804 self.overrides = if overrides.inherit_daemon_options && settings == RunOverrides::default()
1806 {
1807 None
1808 } else {
1809 Some(overrides)
1810 };
1811 Ok(())
1812 }
1813
1814 pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
1826 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1827 bail!(
1828 "task {} is {}; only a queued or held task's instruction can \
1829 be edited",
1830 self.short(),
1831 self.status.as_str()
1832 );
1833 }
1834 if self.review_branch.is_some() && self.review_instruction.is_none() {
1837 self.review_instruction = Some(self.instruction.clone());
1838 }
1839 if self.instruction != instruction {
1840 self.resume_override = None;
1843 }
1844 self.title = title;
1845 self.instruction = instruction;
1846 Ok(())
1847 }
1848
1849 pub fn handed_off(&mut self, why: impl Into<String>) {
1869 let why = why.into();
1870 self.diagnostic = None;
1871 self.note_held();
1872 self.status = TaskStatus::Held;
1873 self.hold_source = Some(HoldSource::Machine);
1874 self.hold_reason = Some(why.clone());
1875 self.last_error = Some(why);
1876 }
1877
1878 pub fn release_operator(&mut self, source: &str, question_id: Option<&str>) {
1880 self.recovery_budget = None;
1881 self.recovery_generations.clear();
1882 self.recovery_admission = None;
1883 if self
1884 .resumption
1885 .as_ref()
1886 .is_some_and(|r| r.instruction != self.instruction)
1887 {
1888 self.resume_override = None;
1891 }
1892 let approval = Resumption {
1893 generation: self.resumption.as_ref().map_or(1, |r| r.generation + 1),
1894 source: source.to_owned(),
1895 question_id: question_id.map(str::to_owned),
1896 at: Timestamp::now(),
1897 instruction: self.instruction.clone(),
1898 answers_through: self.answers.len(),
1899 runs_through: self.runs.len(),
1900 blocking_evidence: Vec::new(),
1901 cleared_reason: self
1902 .hold_reason
1903 .clone()
1904 .or_else(|| self.block_reason.clone()),
1905 };
1906 self.release();
1907 self.operator_decided_at = Some(approval.at);
1908 if let Some(previous) = self.resumption.replace(approval) {
1909 self.resumption_history.push(previous);
1910 }
1911 }
1912
1913 pub(crate) fn recovery_failed(&mut self, state: &crate::run::RunState) {
1915 let (base, inputs) = self.recovery_admission.as_ref().map_or_else(
1916 || (state.base_commit.clone(), recovery_inputs(&state.sources)),
1917 |a| (a.base.clone(), a.inputs.clone()),
1918 );
1919 let progress = recovery_progress(state);
1920 let stage = if state.prep_failure.is_some() {
1921 "preparation"
1922 } else {
1923 "execution"
1924 };
1925 let matches = |b: &RecoveryBudget| {
1926 b.instruction == self.instruction
1927 && b.base == base
1928 && b.inputs == inputs
1929 && (progress.is_none() || b.progress == progress)
1930 };
1931 if !self.recovery_budget.as_ref().is_some_and(matches) {
1932 let previous = self
1933 .recovery_generations
1934 .iter()
1935 .position(matches)
1936 .map(|index| self.recovery_generations.remove(index));
1937 if let Some(old) = self.recovery_budget.take() {
1938 self.recovery_generations.push(old);
1939 }
1940 self.recovery_budget = previous;
1941 }
1942 if let Some(budget) = &mut self.recovery_budget {
1943 budget.failures = budget.failures.saturating_add(1);
1944 budget.stage = stage.to_owned();
1945 } else {
1946 self.recovery_budget = Some(RecoveryBudget {
1947 instruction: self.instruction.clone(),
1948 base,
1949 inputs,
1950 stage: stage.to_owned(),
1951 failures: 1,
1952 progress,
1953 });
1954 }
1955 }
1956
1957 pub(crate) fn recovery_exhausted(
1958 &self,
1959 base: &str,
1960 inputs: &str,
1961 limit: usize,
1962 ) -> Option<String> {
1963 let same_input = |b: &&RecoveryBudget| {
1964 b.instruction == self.instruction && b.base == base && b.inputs == inputs
1965 };
1966 let budget = self
1967 .recovery_budget
1968 .as_ref()
1969 .filter(same_input)
1970 .or_else(|| self.recovery_generations.iter().rev().find(same_input))?;
1971 (budget.instruction == self.instruction && budget.base == base && budget.inputs == inputs
1972 && budget.failures >= limit.max(1)).then(|| format!(
1973 "[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.",
1974 budget.failures, budget.stage))
1975 }
1976
1977 pub fn release(&mut self) {
1981 self.import_chat_notice();
1984 if self.hold_notice_cause.is_none()
1985 && matches!(self.status, TaskStatus::Held | TaskStatus::Blocked)
1986 {
1987 self.hold_notice_cause = Some(self.unresolved_hold_cause());
1988 }
1989 let refused_handover = self.status == TaskStatus::Held
1990 && self.hold_source == Some(HoldSource::Machine)
1991 && self.review_branch.is_some()
1992 && self
1993 .review_instruction
1994 .as_ref()
1995 .is_none_or(|i| *i == self.instruction);
1996 self.status = TaskStatus::Queued;
1997 self.held_at = None;
1998 self.attempts = 0;
1999 self.last_error = None;
2000 self.park_reason = None;
2001 self.hold_reason = None;
2004 self.hold_source = None;
2005 self.diagnostic = None;
2006 self.blocked_by.clear();
2011 self.block_reason = None;
2012 self.blocked_from = None;
2013 if !refused_handover {
2017 self.review_branch = None;
2018 self.review_instruction = None;
2019 }
2020 self.fresh_start = false;
2021 }
2022}
2023
2024fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
2028 use std::io::ErrorKind;
2029 let (stem, ext) = match name.rfind('.') {
2030 Some(i) if i > 0 => (&name[..i], &name[i..]),
2031 _ => (name, ""),
2032 };
2033 for n in 1u32.. {
2034 let candidate = if n == 1 {
2035 name.to_owned()
2036 } else {
2037 let suffix = format!("-{n}");
2038 let room = 64usize.saturating_sub(suffix.len() + ext.len());
2039 let stem: String = stem.chars().take(room).collect();
2040 format!("{stem}{suffix}{ext}")
2041 };
2042 if !crate::ask::valid_asset_name(&candidate) {
2043 bail!("no valid attachment name is left for `{name}`");
2044 }
2045 let path = dir.join(&candidate);
2046 match std::fs::OpenOptions::new()
2047 .write(true)
2048 .create_new(true)
2049 .open(&path)
2050 {
2051 Ok(mut out) => {
2052 let copied = std::fs::File::open(src)
2053 .and_then(|mut input| std::io::copy(&mut input, &mut out));
2054 if let Err(e) = copied {
2055 drop(out);
2056 let _ = std::fs::remove_file(&path);
2057 return Err(e).with_context(|| format!("copy {}", src.display()));
2058 }
2059 return Ok((candidate, path));
2060 }
2061 Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
2062 Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
2063 }
2064 }
2065 unreachable!("the counter never runs out")
2066}
2067
2068const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
2071
2072pub(crate) struct TaskLock {
2075 path: PathBuf,
2076 token: String,
2077}
2078
2079fn owner_token() -> String {
2082 static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
2083 let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2084 let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
2085 format!("{}-{:016x}", std::process::id(), r.next_u64())
2086}
2087
2088fn break_marker(path: &Path, token: &str) -> PathBuf {
2091 let mut h: u64 = 0xcbf29ce484222325;
2092 for b in token.bytes() {
2093 h ^= u64::from(b);
2094 h = h.wrapping_mul(0x100000001b3);
2095 }
2096 let mut name = path.as_os_str().to_owned();
2097 name.push(format!(".break-{h:016x}"));
2098 PathBuf::from(name)
2099}
2100
2101fn older_than_stale(path: &Path) -> bool {
2102 std::fs::metadata(path)
2103 .and_then(|m| m.modified())
2104 .ok()
2105 .and_then(|t| t.elapsed().ok())
2106 .is_some_and(|age| age > TASK_LOCK_STALE)
2107}
2108
2109fn writer_may_be_gone(token: &str) -> bool {
2110 let Some(pid) = token
2111 .split_once('-')
2112 .and_then(|(pid, _)| pid.parse::<u32>().ok())
2113 else {
2114 return true;
2117 };
2118 if pid == std::process::id() {
2119 return false;
2120 }
2121 crate::proc::pid_status(pid) == Some(false)
2123}
2124
2125const MAX_MARKER_DEPTH: u8 = 3;
2129
2130fn take_marker(
2136 path: &Path,
2137 token: &str,
2138 depth: u8,
2139 retry: crate::lock_file::Retry,
2140) -> Option<(PathBuf, String)> {
2141 use std::io::Write;
2142 let marker = break_marker(path, token);
2143 for _ in 0..2 {
2144 match retry.create(&marker) {
2145 Ok(mut f) => {
2146 let mine = owner_token();
2147 if f.write_all(mine.as_bytes()).is_err() {
2148 drop(f);
2149 let _ = std::fs::remove_file(&marker);
2150 return None;
2151 }
2152 return Some((marker, mine));
2153 }
2154 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2155 let judged = std::fs::read_to_string(&marker).ok();
2156 match judged.filter(|_| older_than_stale(&marker)) {
2157 Some(judged) if depth < MAX_MARKER_DEPTH => {
2158 remove_lock_if(&marker, &judged, true, depth + 1, retry)?;
2159 }
2160 _ => return None,
2161 }
2162 }
2163 Err(_) => return None,
2164 }
2165 }
2166 None
2167}
2168
2169fn remove_lock_if(
2174 path: &Path,
2175 token: &str,
2176 require_stale: bool,
2177 depth: u8,
2178 retry: crate::lock_file::Retry,
2179) -> Option<bool> {
2180 let (marker, mine) = take_marker(path, token, depth, retry)?;
2181 let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
2182 && (!require_stale || older_than_stale(path) && writer_may_be_gone(token));
2183 if still {
2184 let _ = std::fs::remove_file(path);
2185 }
2186 if depth >= MAX_MARKER_DEPTH {
2188 if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
2189 let _ = std::fs::remove_file(&marker);
2190 }
2191 } else {
2192 let _ = remove_lock_if(&marker, &mine, false, depth + 1, retry);
2193 }
2194 Some(still)
2195}
2196
2197#[cfg(test)]
2200fn break_stale(path: &Path, judged: &str) -> bool {
2201 break_stale_with(path, judged, crate::lock_file::Retry::new(None))
2202}
2203
2204fn break_stale_with(path: &Path, judged: &str, retry: crate::lock_file::Retry) -> bool {
2205 remove_lock_if(path, judged, true, 0, retry).unwrap_or(false)
2206}
2207
2208impl Drop for TaskLock {
2209 fn drop(&mut self) {
2210 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
2211 let retry = crate::lock_file::Retry::new(Some(deadline));
2212 loop {
2213 match remove_lock_if(&self.path, &self.token, false, 0, retry) {
2214 Some(_) => return,
2215 None if std::time::Instant::now() > deadline => return,
2216 None => std::thread::sleep(std::time::Duration::from_millis(5)),
2217 }
2218 }
2219 }
2220}
2221
2222#[derive(Debug, Clone)]
2224pub struct Queue {
2225 root: PathBuf,
2226 ids: IdSource,
2227}
2228
2229#[derive(Clone)]
2232struct IdSource(Option<std::sync::Arc<dyn Fn() -> String + Send + Sync>>);
2233
2234impl std::fmt::Debug for IdSource {
2235 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2236 f.write_str(if self.0.is_some() {
2237 "IdSource(custom)"
2238 } else {
2239 "IdSource"
2240 })
2241 }
2242}
2243
2244fn path_exists(path: &Path) -> Result<bool> {
2247 match std::fs::symlink_metadata(path) {
2248 Ok(_) => Ok(true),
2249 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
2250 Err(e) => Err(e).with_context(|| format!("stat {}", path.display())),
2251 }
2252}
2253
2254fn validate_attachments(sources: &[PathBuf]) -> Result<Vec<(&PathBuf, &str)>> {
2255 let mut wanted = Vec::new();
2256 for src in sources {
2257 let name = src
2258 .file_name()
2259 .and_then(|n| n.to_str())
2260 .with_context(|| format!("`{}` has no usable file name", src.display()))?;
2261 if !crate::ask::valid_asset_name(name) {
2262 bail!(
2263 "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
2264 with no `..`; rename the file and try again"
2265 );
2266 }
2267 if !src.is_file() {
2268 bail!("attachment `{}` is not a file", src.display());
2269 }
2270 wanted.push((src, name));
2271 }
2272 Ok(wanted)
2273}
2274
2275const FILE_ATTEMPTS: usize = 256;
2277
2278impl Queue {
2279 pub fn open() -> Self {
2281 Self::at(crate::run::home().join("queue"))
2282 }
2283
2284 pub fn at(root: PathBuf) -> Self {
2287 Self {
2288 root,
2289 ids: IdSource(None),
2290 }
2291 }
2292
2293 #[doc(hidden)]
2296 pub fn with_id_source(mut self, f: impl Fn() -> String + Send + Sync + 'static) -> Self {
2297 self.ids = IdSource(Some(std::sync::Arc::new(f)));
2298 self
2299 }
2300
2301 fn mint_id(&self) -> String {
2302 match &self.ids.0 {
2303 Some(f) => f(),
2304 None => new_id(),
2305 }
2306 }
2307
2308 fn file_with(
2319 &self,
2320 task: &mut Task,
2321 hook: impl FnMut(&mut Task) -> Result<bool>,
2322 undo: impl FnMut(&mut Task),
2323 ) -> Result<()> {
2324 self.reserve_id(task, hook, undo, true)
2325 }
2326
2327 fn reserve_id(
2335 &self,
2336 task: &mut Task,
2337 mut hook: impl FnMut(&mut Task) -> Result<bool>,
2338 mut undo: impl FnMut(&mut Task),
2339 store: bool,
2340 ) -> Result<()> {
2341 let redraw_first = self.ids.0.is_some() && task.attachments.is_empty();
2342 for attempt in 0..FILE_ATTEMPTS {
2343 if attempt > 0 || redraw_first {
2344 task.id = self.mint_id();
2345 }
2346 let _lock = self.lock_task(&task.id)?;
2347 if path_exists(&self.path_of(&task.id))? {
2348 continue;
2349 }
2350 if !hook(task)? {
2351 continue;
2352 }
2353 if store {
2354 if let Err(e) = self.write_record(task) {
2355 undo(task);
2356 return Err(e);
2357 }
2358 task.unfiled = false;
2359 }
2360 return Ok(());
2361 }
2362 bail!("no free task id after {FILE_ATTEMPTS} attempts")
2363 }
2364
2365 fn id_in_use(&self, id: &str, check_attachments: bool) -> Result<bool> {
2370 if path_exists(&self.lock_path(id))? {
2371 return Ok(true);
2372 }
2373 Ok(check_attachments && path_exists(&self.attachments_dir(id))?)
2374 }
2375
2376 fn plain_filing_hook(&self, task: &Task) -> Result<bool> {
2380 Ok(!self.id_in_use(&task.id, task.attachments.is_empty())?)
2381 }
2382
2383 pub fn file_claimed(&self, task: &mut Task) -> Result<Claim> {
2386 let mut claim = None;
2387 self.file_with(
2388 task,
2389 |t| {
2390 if self.id_in_use(&t.id, true)? {
2391 return Ok(false);
2392 }
2393 claim = Some(self.claim(&t.id)?);
2394 Ok(true)
2395 },
2396 |_| {},
2397 )?;
2398 claim.context("claim missing after filing")
2399 }
2400
2401 pub fn root(&self) -> &Path {
2403 &self.root
2404 }
2405
2406 pub fn path_of(&self, id: &str) -> PathBuf {
2408 self.root.join(format!("{id}.json"))
2409 }
2410
2411 pub fn attachments_dir(&self, id: &str) -> PathBuf {
2414 self.root.join(format!("{id}.attachments"))
2415 }
2416
2417 pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2428 let wanted = validate_attachments(sources)?;
2429 if task.unfiled && task.attachments.is_empty() {
2430 let names = std::cell::RefCell::new(Vec::new());
2433 self.reserve_id(
2434 task,
2435 |t| {
2436 if self.id_in_use(&t.id, true)? {
2437 return Ok(false);
2438 }
2439 *names.borrow_mut() = self.copy_attachments(t, &wanted)?;
2440 Ok(true)
2441 },
2442 |_| {},
2443 false,
2444 )?;
2445 return Ok(names.into_inner());
2446 }
2447 self.copy_attachments(task, &wanted)
2448 }
2449
2450 fn copy_attachments(
2452 &self,
2453 task: &mut Task,
2454 wanted: &[(&PathBuf, &str)],
2455 ) -> Result<Vec<String>> {
2456 let dir = self.attachments_dir(&task.id);
2457 let existed = dir.is_dir();
2458 std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
2459 let mut created: Vec<PathBuf> = Vec::new();
2460 let mut names = Vec::new();
2461 let mut copy_all = || -> Result<()> {
2462 for (src, name) in wanted {
2463 let (stored, path) = copy_new(&dir, src, name)?;
2464 created.push(path);
2465 names.push(stored);
2466 }
2467 Ok(())
2468 };
2469 if let Err(e) = copy_all() {
2470 for path in &created {
2471 let _ = std::fs::remove_file(path);
2472 }
2473 if !existed {
2474 let _ = std::fs::remove_dir(&dir);
2475 }
2476 return Err(e);
2477 }
2478 task.attachments.extend(names.iter().cloned());
2479 Ok(names)
2480 }
2481
2482 pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2493 if task.unfiled && task.attachments.is_empty() {
2494 return self.attach_and_file(task, sources);
2495 }
2496 let dir = self.attachments_dir(&task.id);
2497 let existed = dir.is_dir();
2498 let before = task.attachments.len();
2499 let names = self.attach(task, sources)?;
2500 if let Err(e) = self.put(task) {
2501 for name in &names {
2502 let _ = std::fs::remove_file(dir.join(name));
2503 }
2504 if !existed {
2505 let _ = std::fs::remove_dir(&dir);
2506 }
2507 task.attachments.truncate(before);
2508 return Err(e);
2509 }
2510 Ok(names)
2511 }
2512
2513 fn attach_and_file(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2517 let wanted = validate_attachments(sources)?;
2518 let before = task.attachments.len();
2519 let names = std::cell::RefCell::new(Vec::new());
2520 let result = self.file_with(
2521 task,
2522 |t| {
2523 if self.id_in_use(&t.id, true)? {
2524 return Ok(false);
2525 }
2526 *names.borrow_mut() = self.copy_attachments(t, &wanted)?;
2527 Ok(true)
2528 },
2529 |t| {
2530 let dir = self.attachments_dir(&t.id);
2531 for name in names.borrow().iter() {
2532 let _ = std::fs::remove_file(dir.join(name));
2533 }
2534 let _ = std::fs::remove_dir(&dir);
2536 t.attachments.truncate(before);
2537 },
2538 );
2539 result.map(|()| names.into_inner())
2540 }
2541
2542 pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
2546 let dir = self.attachments_dir(&task.id);
2547 task.attachments
2548 .iter()
2549 .map(|n| {
2550 let p = dir.join(n);
2551 std::path::absolute(&p).unwrap_or(p)
2552 })
2553 .collect()
2554 }
2555
2556 pub fn put(&self, task: &mut Task) -> Result<()> {
2565 if task.unfiled {
2566 return self.file_with(task, |t| self.plain_filing_hook(t), |_| {});
2568 }
2569 let _lock = self.lock_task(&task.id)?;
2570 self.put_unlocked(task)
2571 }
2572
2573 pub fn create_new(&self, task: &mut Task) -> Result<bool> {
2578 let _lock = self.lock_task(&task.id)?;
2579 if path_exists(&self.path_of(&task.id))? {
2580 return Ok(false);
2581 }
2582 self.write_record(task)?;
2583 task.unfiled = false;
2584 Ok(true)
2585 }
2586
2587 fn put_unlocked(&self, task: &mut Task) -> Result<()> {
2589 self.merge_stored(task);
2590 self.write_record(task)
2591 }
2592
2593 fn merge_stored(&self, task: &mut Task) {
2595 if let Ok(mut stored) = read_path(&self.path_of(&task.id)) {
2596 stored.import_chat_notice();
2597 if task.chat.is_none() {
2600 task.chat = stored.chat.clone();
2601 }
2602 for key in stored.chat_notices {
2603 if !task.chat_notices.contains(&key) {
2604 task.chat_notices.push(key);
2605 }
2606 }
2607 for run in stored.runs {
2608 if !task.runs.contains(&run) {
2609 task.runs.push(run);
2610 }
2611 }
2612 for p in stored.passes {
2613 if !task
2614 .passes
2615 .iter()
2616 .any(|q| q.run == p.run && q.pass_index == p.pass_index)
2617 {
2618 task.passes.push(p);
2619 }
2620 }
2621 if let Some(n) = stored.chat_notice
2624 && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
2625 {
2626 task.chat_notice = Some(n);
2627 }
2628 }
2629 }
2630
2631 fn write_record(&self, task: &mut Task) -> Result<()> {
2633 task.import_chat_notice();
2634 task.updated_at = Timestamp::now();
2635 std::fs::create_dir_all(&self.root)
2636 .with_context(|| format!("create {}", self.root.display()))?;
2637 let body = serde_json::to_string_pretty(task).context("serialize task")?;
2638 let path = self.path_of(&task.id);
2639 let tmp = path.with_extension("json.tmp");
2640 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
2641 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
2642 if let (Some(notice), Some(home)) = (
2647 crate::notices::task_held(task),
2648 self.root.parent().filter(|p| !p.as_os_str().is_empty()),
2649 ) {
2650 crate::notices::raise_in(home, notice);
2651 }
2652 Ok(())
2653 }
2654
2655 pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
2660 let id = self.resolve_id(id)?;
2661 let _lock = self.lock_task(&id)?;
2662 let mut task = self.get(&id)?;
2663 if !f(&mut task) {
2664 return Ok(false);
2665 }
2666 self.put_unlocked(&mut task)?;
2667 Ok(true)
2668 }
2669
2670 pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
2675 let id = self.resolve_id(id)?;
2676 let _lock = self.lock_task(&id)?;
2679 let mut task = self.get(&id)?;
2680 if task.link_run(run) {
2681 self.put_unlocked(&mut task)?;
2682 }
2683 Ok(task)
2684 }
2685
2686 pub(crate) fn lock_task(&self, id: &str) -> Result<TaskLock> {
2694 std::fs::create_dir_all(&self.root)
2695 .with_context(|| format!("create {}", self.root.display()))?;
2696 let path = self.root.join(format!("{id}.write-lock"));
2697 let started = std::time::Instant::now();
2698 let retry = crate::lock_file::Retry::task_write(started + TASK_LOCK_STALE);
2699 let marker_retry = crate::lock_file::Retry::new(Some(started + TASK_LOCK_STALE));
2700 loop {
2701 match retry.create(&path) {
2702 Ok(mut file) => {
2703 use std::io::Write;
2704 let token = owner_token();
2705 if let Err(e) = file.write_all(token.as_bytes()) {
2706 drop(file);
2707 let _ = std::fs::remove_file(&path);
2708 return Err(e).with_context(|| format!("lock {}", path.display()));
2709 }
2710 return Ok(TaskLock { path, token });
2711 }
2712 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2713 let judged = std::fs::read_to_string(&path).ok();
2716 let broken = judged
2717 .filter(|_| older_than_stale(&path))
2718 .is_some_and(|judged| break_stale_with(&path, &judged, marker_retry));
2719 if !broken {
2723 if started.elapsed() > TASK_LOCK_STALE {
2724 bail!("could not lock task {id}");
2725 }
2726 std::thread::sleep(std::time::Duration::from_millis(15));
2727 }
2728 }
2729 Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
2730 }
2731 }
2732 }
2733
2734 pub fn get(&self, id: &str) -> Result<Task> {
2736 let resolved = self.resolve_id(id)?;
2737 read_path(&self.path_of(&resolved))
2738 }
2739
2740 pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
2772 let resolved = self.resolve_id(id)?;
2773 if in_flight {
2774 bail!("task {resolved} is being run by a live daemon right now");
2775 }
2776 self.write_tombstone(&resolved)?;
2777 if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
2778 {
2779 if self.path_of(&resolved).exists() {
2783 let _ = std::fs::remove_file(self.tombstone_path(&resolved));
2784 }
2785 return Err(e);
2786 }
2787 let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
2788 Ok(Removal {
2789 id: resolved,
2790 released,
2791 still_blocked,
2792 })
2793 }
2794
2795 fn tombstone_path(&self, id: &str) -> PathBuf {
2798 self.root.join(format!("{id}.removed"))
2799 }
2800
2801 fn write_tombstone(&self, id: &str) -> Result<()> {
2802 let path = self.tombstone_path(id);
2803 let tmp = self.root.join(format!("{id}.removed.tmp"));
2804 std::fs::write(&tmp, b"")
2805 .and_then(|()| std::fs::rename(&tmp, &path))
2806 .with_context(|| format!("write {}", path.display()))
2807 }
2808
2809 fn was_deleted(&self, id: &str) -> bool {
2813 self.tombstone_path(id).is_file() && !self.path_of(id).exists()
2814 }
2815
2816 pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
2820 let deleted = deleted_blockers(self, questions, &task.blocked_by);
2821 deleted
2822 .into_iter()
2823 .filter(|id| task.dependency_deleted(id))
2824 .collect()
2825 }
2826
2827 pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
2831 let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
2832 return;
2833 };
2834 let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
2835 let waiting = dependent.status == TaskStatus::Blocked;
2836 let message = match (ja, waiting) {
2837 (true, false) => format!(
2838 "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
2839 dependent.short(),
2840 short(deleted)
2841 ),
2842 (true, true) => format!(
2843 "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
2844 dependent.short(),
2845 short(deleted)
2846 ),
2847 (false, false) => format!(
2848 "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
2849 dependent.short(),
2850 short(deleted)
2851 ),
2852 (false, true) => format!(
2853 "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
2854 dependent.short(),
2855 short(deleted)
2856 ),
2857 };
2858 crate::notices::raise_in(
2859 home,
2860 crate::notices::Notice::info(
2861 &format!("unblocked:{}:{}", dependent.id, deleted),
2862 message,
2863 )
2864 .link(crate::notices::Link::Task {
2865 id: dependent.id.clone(),
2866 }),
2867 );
2868 }
2869
2870 fn release_dependents_of(
2875 &self,
2876 dependency: &str,
2877 questions: &Questions,
2878 ) -> (Vec<String>, Vec<String>) {
2879 let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
2880 for listed in self.list() {
2881 if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
2882 matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
2883 }) {
2884 continue;
2885 }
2886 let Ok(_claim) = self.claim(&listed.id) else {
2887 continue;
2888 };
2889 let Ok(mut task) = self.get(&listed.id) else {
2890 continue;
2891 };
2892 let references: Vec<String> = task.blocked_by.iter().filter(|id| {
2893 matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
2894 }).cloned().collect();
2895 if references.is_empty() {
2896 continue;
2897 }
2898 for reference in references {
2899 task.dependency_deleted(&reference);
2900 }
2901 if self.put(&mut task).is_err() {
2902 continue;
2903 }
2904 self.note_dependency_deleted(&task, dependency);
2905 if task.status == TaskStatus::Blocked {
2906 still_blocked.push(task.id.clone());
2907 } else {
2908 released.push(task.id.clone());
2909 }
2910 }
2911 (released, still_blocked)
2912 }
2913
2914 fn remove_record_with_attachments(
2918 &self,
2919 resolved: &str,
2920 remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
2921 ) -> Result<()> {
2922 self.sweep_removed_attachments();
2929 let attachments = self.attachments_dir(resolved);
2930 let aside = self.root.join(format!("{resolved}.attachments.removing"));
2931 let moved = match std::fs::rename(&attachments, &aside) {
2932 Ok(()) => true,
2933 Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
2934 Err(e) => {
2935 return Err(e).with_context(|| format!("remove {}", attachments.display()));
2936 }
2937 };
2938 let path = self.path_of(resolved);
2939 if let Err(e) = remove_record(&path) {
2940 if moved {
2941 let _ = std::fs::rename(&aside, &attachments);
2942 }
2943 return Err(e).with_context(|| format!("remove {}", path.display()));
2944 }
2945 if moved {
2946 if let Err(e) = std::fs::remove_dir_all(&aside) {
2947 tracing::warn!("leftover attachments {}: {e}", aside.display());
2948 }
2949 }
2950 let lock = self.lock_path(resolved);
2951 if let Err(e) = std::fs::remove_file(&lock) {
2952 if e.kind() != std::io::ErrorKind::NotFound {
2953 return Err(e).with_context(|| format!("remove {}", lock.display()));
2954 }
2955 }
2956 Ok(())
2957 }
2958
2959 fn sweep_removed_attachments(&self) {
2965 let Ok(entries) = std::fs::read_dir(&self.root) else {
2966 return;
2967 };
2968 for entry in entries.flatten() {
2969 let name = entry.file_name();
2970 let name = name.to_string_lossy();
2971 let Some(id) = name.strip_suffix(".attachments.removing") else {
2972 continue;
2973 };
2974 if !self.path_of(id).exists() {
2977 if let Err(e) = std::fs::remove_dir_all(entry.path()) {
2978 tracing::warn!("leftover attachments {}: {e}", entry.path().display());
2979 }
2980 }
2981 }
2982 self.sweep_tombstones();
2983 }
2984
2985 fn sweep_tombstones(&self) {
2994 let Ok(entries) = std::fs::read_dir(&self.root) else {
2995 return;
2996 };
2997 let tasks = self.list();
2998 for entry in entries.flatten() {
2999 let name = entry.file_name();
3000 let name = name.to_string_lossy();
3001 let Some(id) = name.strip_suffix(".removed") else {
3002 continue;
3003 };
3004 if self.path_of(id).exists()
3005 || tasks.iter().any(|t| {
3006 t.blocked_by
3007 .iter()
3008 .any(|b| id.starts_with(b) || id.ends_with(b))
3009 })
3010 {
3011 continue;
3012 }
3013 let old = entry
3014 .metadata()
3015 .and_then(|m| m.modified())
3016 .ok()
3017 .and_then(|m| m.elapsed().ok())
3018 .is_some_and(|age| age >= TOMBSTONE_GRACE);
3019 if old {
3020 let _ = std::fs::remove_file(entry.path());
3021 }
3022 }
3023 }
3024
3025 fn lock_path(&self, id: &str) -> PathBuf {
3028 self.root.join(format!("{id}.lock"))
3029 }
3030
3031 pub fn list(&self) -> Vec<Task> {
3044 let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
3045 .into_iter()
3046 .flatten()
3047 .flatten()
3048 .map(|e| e.path())
3049 .filter(|p| p.extension().is_some_and(|x| x == "json"))
3050 .filter_map(|p| read_path(&p).ok())
3051 .collect();
3052 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
3053 tasks
3054 }
3055
3056 pub fn unreadable(&self) -> Vec<PathBuf> {
3060 std::fs::read_dir(&self.root)
3061 .into_iter()
3062 .flatten()
3063 .flatten()
3064 .map(|e| e.path())
3065 .filter(|p| p.extension().is_some_and(|x| x == "json"))
3066 .filter(|p| read_path(p).is_err())
3067 .collect()
3068 }
3069
3070 pub fn superseded(&self) -> HashMap<String, String> {
3083 let mut by = HashMap::new();
3084 for task in self.list() {
3085 for earlier in &task.runs {
3086 if let Some(later) = task.successor_of(earlier) {
3087 by.insert(earlier.clone(), later.clone());
3088 }
3089 }
3090 }
3091 by
3092 }
3093
3094 pub fn superseded_by(&self, run: &str) -> Option<String> {
3102 for task in self.list() {
3103 if task.runs.iter().any(|r| r == run) {
3104 return task.successor_of(run).cloned();
3105 }
3106 }
3107 None
3108 }
3109
3110 pub fn latest_attempt(&self, run: &str) -> Option<String> {
3121 for task in self.list() {
3122 if task.runs.iter().any(|r| r == run) {
3123 return task.runs.last().filter(|last| **last != run).cloned();
3124 }
3125 }
3126 None
3127 }
3128
3129 pub fn next_runnable(&self) -> Option<Task> {
3134 let mut runnable: Vec<Task> = self
3135 .list()
3136 .into_iter()
3137 .filter(|t| t.status.runnable())
3138 .collect();
3139 runnable.sort_unstable_by(dispatch_order);
3140 runnable.into_iter().next()
3141 }
3142
3143 pub fn claim(&self, id: &str) -> Result<Claim> {
3150 std::fs::create_dir_all(&self.root)
3151 .with_context(|| format!("create {}", self.root.display()))?;
3152 let path = self.lock_path(id);
3153 let retry = crate::lock_file::Retry::new(None);
3154 match retry.create(&path) {
3155 Ok(mut f) => {
3156 use std::io::Write as _;
3157 let _ = writeln!(f, "{}", std::process::id());
3159 Ok(Claim { path })
3160 }
3161 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
3162 bail!("task {id} is already claimed ({} exists)", path.display())
3163 }
3164 Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
3165 }
3166 }
3167
3168 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
3170 if !prefix.is_empty()
3171 && prefix != "."
3172 && prefix != ".."
3173 && !prefix.contains(['/', '\\'])
3174 && self.path_of(prefix).is_file()
3175 {
3176 return Ok(prefix.to_owned());
3177 }
3178 let ids = record_ids(&self.root, "json")?;
3179 match resolve_reference(prefix, ids.iter().map(String::as_str)) {
3180 Reference::Found(id) => Ok(id),
3181 Reference::Missing => bail!("no task matches `{prefix}`"),
3182 Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
3183 }
3184 }
3185
3186 pub fn revision(&self) -> u64 {
3193 self.revision_excluding(&std::collections::BTreeSet::new())
3194 }
3195
3196 pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
3199 use std::hash::{Hash as _, Hasher as _};
3200
3201 let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
3202 .into_iter()
3203 .flatten()
3204 .flatten()
3205 .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
3206 .filter(|e| {
3207 let path = e.path();
3208 !path
3209 .file_stem()
3210 .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
3211 })
3212 .filter_map(|e| {
3213 let name = e.file_name().to_string_lossy().into_owned();
3214 let mtime = e
3215 .metadata()
3216 .ok()?
3217 .modified()
3218 .ok()?
3219 .duration_since(std::time::UNIX_EPOCH)
3220 .ok()?
3221 .as_millis() as u64;
3222 Some((name, mtime))
3223 })
3224 .collect();
3225
3226 if entries.is_empty() {
3227 return 0;
3228 }
3229
3230 entries.sort_unstable();
3231 let mut hasher = std::hash::DefaultHasher::new();
3232 for (name, mtime) in &entries {
3233 name.hash(&mut hasher);
3234 mtime.hash(&mut hasher);
3235 }
3236 let h = hasher.finish();
3237 if h == 0 { 1 } else { h }
3238 }
3239}
3240
3241const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
3244
3245#[derive(Debug, Clone)]
3247pub struct Removal {
3248 pub id: String,
3250 pub released: Vec<String>,
3254 pub still_blocked: Vec<String>,
3257}
3258
3259#[derive(Debug)]
3261pub struct Claim {
3262 path: PathBuf,
3263}
3264
3265impl Drop for Claim {
3266 fn drop(&mut self) {
3267 let _ = std::fs::remove_file(&self.path);
3268 }
3269}
3270
3271pub fn title_from(instruction: &str, max: usize) -> String {
3274 let Some(line) = first_line(instruction) else {
3275 return "(empty task)".to_owned();
3276 };
3277 if line.chars().count() <= max {
3278 return line.to_owned();
3279 }
3280 let head: String = line.chars().take(max.saturating_sub(1)).collect();
3281 format!("{head}…")
3282}
3283
3284pub(crate) fn first_line(instruction: &str) -> Option<&str> {
3290 let line = instruction
3291 .lines()
3292 .map(str::trim)
3293 .find(|l| !l.is_empty())?
3294 .trim_start_matches(['#', '-', '*', '>', ' '])
3295 .trim();
3296 (!line.is_empty()).then_some(line)
3297}
3298
3299fn read_path(path: &Path) -> Result<Task> {
3300 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
3301 let task: Task =
3302 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
3303 if task.schema > SCHEMA {
3309 bail!(
3310 "task {} was written by a different magi (schema {}, this build \
3311 speaks {SCHEMA})",
3312 task.id,
3313 task.schema
3314 );
3315 }
3316 Ok(task)
3317}
3318
3319#[derive(Debug, PartialEq, Eq)]
3322pub(crate) enum Reference {
3323 Found(String),
3324 Missing,
3325 Ambiguous,
3326}
3327
3328pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
3329 let mut hits = Vec::new();
3330 for candidate in ids {
3331 if candidate == id {
3332 return Reference::Found(candidate.to_owned());
3333 }
3334 if candidate.starts_with(id) || candidate.ends_with(id) {
3335 hits.push(candidate);
3336 }
3337 }
3338 match hits.as_slice() {
3339 [] => Reference::Missing,
3340 [one] => Reference::Found((*one).to_owned()),
3341 _ => Reference::Ambiguous,
3342 }
3343}
3344
3345pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
3348 let entries = match std::fs::read_dir(root) {
3349 Ok(entries) => entries,
3350 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
3351 Err(e) => return Err(e.into()),
3352 };
3353 let mut ids = Vec::new();
3354 for entry in entries {
3355 let path = entry?.path();
3356 if path.extension().is_some_and(|e| e == extension)
3357 && let Some(id) = path.file_stem().and_then(|s| s.to_str())
3358 {
3359 ids.push(id.to_owned());
3360 }
3361 }
3362 Ok(ids)
3363}
3364
3365#[derive(Debug, PartialEq, Eq)]
3368pub enum Dependency {
3369 Task(String),
3371 Question(String),
3373 Deleted(String),
3375 Missing,
3377 Ambiguous,
3379 Unknown,
3381}
3382
3383pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
3385 let (Ok(tasks), Ok(qs), Ok(removed)) = (
3386 record_ids(queue.root(), "json"),
3387 record_ids(questions.root(), "json"),
3388 record_ids(queue.root(), "removed"),
3389 ) else {
3390 return Dependency::Unknown;
3391 };
3392 let mut ids = tasks
3395 .iter()
3396 .chain(qs.iter())
3397 .chain(removed.iter())
3398 .cloned()
3399 .collect::<Vec<_>>();
3400 ids.sort();
3401 ids.dedup();
3402 match resolve_reference(id, ids.iter().map(String::as_str)) {
3403 Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
3404 Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
3405 Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
3406 Reference::Found(_) => Dependency::Unknown,
3407 Reference::Missing => Dependency::Missing,
3408 Reference::Ambiguous => Dependency::Ambiguous,
3409 }
3410}
3411
3412pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
3414 for id in ids {
3415 if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
3416 dependency(queue, questions, id)
3417 {
3418 *id = full;
3419 }
3420 }
3421}
3422
3423pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
3426 let inventory = crate::blockers::Inventory::load(queue, questions);
3427 let mut text = String::from(
3428 "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
3429 );
3430 match record_ids(queue.root(), "json") {
3431 Ok(ids) => {
3432 for id in ids {
3433 match queue.get(&id) {
3434 Ok(task) => text.push_str(&format!(
3435 "{}: {}; blocked_by={:?}; chain={:?}\n",
3436 task.id,
3437 task.status.as_str(),
3438 task.blocked_by,
3439 inventory.waits_on(&task)
3440 )),
3441 Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
3442 }
3443 }
3444 }
3445 Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
3446 }
3447 text
3448}
3449
3450pub fn missing_blockers(
3452 queue: &Queue,
3453 questions: &Questions,
3454 blocked_by: &[String],
3455) -> Vec<String> {
3456 blocked_by
3457 .iter()
3458 .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
3459 .cloned()
3460 .collect()
3461}
3462
3463pub fn deleted_blockers(
3466 queue: &Queue,
3467 questions: &Questions,
3468 blocked_by: &[String],
3469) -> Vec<String> {
3470 blocked_by
3471 .iter()
3472 .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
3473 .cloned()
3474 .collect()
3475}
3476
3477pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
3489 missing_blocker_hold_reason_in(blocked_by, missing, "en")
3490}
3491
3492pub fn missing_blocker_hold_reason_in(
3494 blocked_by: &[String],
3495 missing: &[String],
3496 language: &str,
3497) -> String {
3498 if crate::lang::is_japanese(language) {
3499 format!(
3500 "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
3501 blocked_by.join(", "),
3502 missing.join(", "),
3503 )
3504 } else {
3505 format!(
3506 "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
3507 blocked_by.join(", "),
3508 missing.join(", "),
3509 )
3510 }
3511}
3512
3513pub fn short(id: &str) -> &str {
3515 id.split('-').next_back().unwrap_or(id)
3516}
3517
3518fn new_id() -> String {
3519 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
3520 let seed = crate::rng::entropy();
3521 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
3522}
3523
3524#[cfg(test)]
3525mod tests {
3526 #[test]
3527 fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
3528 let t = Task::new(
3529 "retry".into(),
3530 "do it".into(),
3531 PathBuf::from("."),
3532 Source::Human,
3533 );
3534 let mut value = serde_json::to_value(&t).unwrap();
3535 for field in ["resumption", "resumption_history", "operator_decided_at"] {
3536 value.as_object_mut().unwrap().remove(field);
3537 }
3538 value["schema"] = serde_json::json!(15);
3539 let mut legacy: Task = serde_json::from_value(value).unwrap();
3540 legacy.hold_machine(Some("automatic retry".into()));
3541 legacy.release();
3542 assert!(legacy.resumption.is_none());
3543 assert!(legacy.resumption_history.is_empty());
3544 assert!(legacy.operator_decided_at.is_none());
3545 legacy.release_operator("cli", None);
3546 legacy.start("run".into());
3547 let reloaded: Task =
3548 serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
3549 assert!(reloaded.resume_override.is_none());
3550 assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
3551 }
3552
3553 #[test]
3554 fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
3555 let mut base = RunOverrides {
3556 merge: Some("pr".to_owned()),
3557 candidates: Some(3),
3558 ..RunOverrides::default()
3559 };
3560 base.merge_over(&RunOverrides {
3561 merge: Some("none".to_owned()),
3562 seed: Some(7),
3563 ..RunOverrides::default()
3564 });
3565 assert_eq!(base.merge.as_deref(), Some("none"));
3566 assert_eq!(base.candidates, Some(3));
3567 assert_eq!(base.seed, Some(7));
3568 }
3569
3570 #[test]
3571 fn an_already_landed_task_is_done_with_its_attempt_refunded() {
3572 let mut t = task("relanded");
3573 t.attempts = 1;
3574 t.status = TaskStatus::Running;
3575 t.already_landed("already in main as 0e368de");
3576 assert_eq!(t.status, TaskStatus::Done);
3577 assert_eq!(t.attempts, 0);
3578 assert!(t.hold_reason.is_none());
3579 assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
3580 }
3581
3582 #[test]
3583 fn missing_blocker_reason_follows_the_language() {
3584 let b = vec!["a".to_owned()];
3585 let en = missing_blocker_hold_reason_in(&b, &b, "en");
3586 assert_eq!(en, missing_blocker_hold_reason(&b, &b));
3587 assert!(en.starts_with("blocked on a"));
3588 assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
3589 assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
3590 }
3591
3592 use super::*;
3593
3594 #[test]
3595 fn triage_applied_survives_release_and_old_records_read_as_empty() {
3596 let mut t = Task::new(
3597 "t".to_owned(),
3598 "i".to_owned(),
3599 PathBuf::from("r"),
3600 Source::Human,
3601 );
3602 t.mark_triage_applied("q1");
3603 t.mark_triage_applied("q1");
3604 t.hold_machine(Some("x".to_owned()));
3605 t.release();
3606 assert_eq!(t.triage_applied, ["q1"]);
3607 assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
3608
3609 let mut v = serde_json::to_value(&t).unwrap();
3610 v.as_object_mut().unwrap().remove("triage_applied");
3611 let old: Task = serde_json::from_value(v).unwrap();
3612 assert!(old.triage_applied.is_empty());
3613 }
3614
3615 #[test]
3616 fn task_counts_of_empty_is_all_zero() {
3617 assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
3618 }
3619
3620 #[test]
3621 fn task_counts_of_tallies_every_status() {
3622 let mut queued = Task::new(
3623 "q".to_owned(),
3624 "i".to_owned(),
3625 PathBuf::from("."),
3626 Source::Human,
3627 );
3628 queued.status = TaskStatus::Queued;
3629 let mut running = queued.clone();
3630 running.status = TaskStatus::Running;
3631 let mut done = queued.clone();
3632 done.status = TaskStatus::Done;
3633 let mut failed = queued.clone();
3634 failed.status = TaskStatus::Failed;
3635 let mut held = queued.clone();
3636 held.status = TaskStatus::Held;
3637 let mut blocked = queued.clone();
3638 blocked.status = TaskStatus::Blocked;
3639
3640 let mut parked = queued.clone();
3641 parked.status = TaskStatus::Parked;
3642
3643 let counts = TaskCounts::of(&[
3644 queued,
3645 running,
3646 done.clone(),
3647 done,
3648 failed,
3649 held,
3650 blocked,
3651 parked,
3652 ]);
3653 assert_eq!(
3654 counts,
3655 TaskCounts {
3656 queued: 1,
3657 running: 1,
3658 done: 2,
3659 failed: 1,
3660 held: 1,
3661 blocked: 1,
3662 parked: 1,
3663 }
3664 );
3665 }
3666
3667 #[test]
3668 fn task_implementers_persist_normalized_without_losing_history() {
3669 let (_dir, q) = queue();
3670 let mut t = task("keep me");
3671 t.hold_manual(Some("waiting".into()));
3672 t.runs.push("earlier-run".into());
3673 t.attachments.push("reference.txt".into());
3674 let cfg: crate::config::Config =
3675 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
3676 .unwrap();
3677 q.put(&mut t).unwrap();
3678 let original = serde_json::to_value(&t).unwrap();
3679 for n in [2, 1, 2] {
3680 t.set_implementers(n, &cfg).unwrap();
3681 q.put(&mut t).unwrap();
3682 t = q.get(&t.id).unwrap();
3683 assert_eq!(t.implementers(), Some(usize::from(n)));
3684 assert_eq!(t.solo, n == 1);
3685 if n == 1 {
3686 assert!(t.overrides.is_none());
3687 }
3688 let mut actual = serde_json::to_value(&t).unwrap();
3689 for field in ["solo", "overrides", "updated_at"] {
3690 actual[field] = original[field].clone();
3691 }
3692 assert_eq!(actual, original);
3693 }
3694 t.overrides.as_mut().unwrap().judges = Some(2);
3695 t.set_implementers(1, &cfg).unwrap();
3696 assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
3697 assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
3698 }
3699
3700 #[test]
3701 fn task_implementers_old_records_keep_inheritance_until_changed() {
3702 let mut value = serde_json::to_value(task("old task")).unwrap();
3703 value.as_object_mut().unwrap().remove("solo");
3704 value.as_object_mut().unwrap().remove("overrides");
3705 let mut old: Task = serde_json::from_value(value).unwrap();
3706 assert_eq!(old.implementers(), None);
3707 old.set_implementers(1, &crate::config::Config::default())
3708 .unwrap();
3709 assert!(old.solo);
3710 assert!(old.overrides.is_none());
3711 }
3712
3713 #[test]
3714 fn task_implementers_reject_invalid_counts_and_states() {
3715 let cfg = crate::config::Config::default();
3716 let mut t = task("unchanged");
3717 for n in [0, 4, 255] {
3718 assert!(t.set_implementers(n, &cfg).is_err());
3719 }
3720 for status in [
3721 TaskStatus::Running,
3722 TaskStatus::Done,
3723 TaskStatus::Failed,
3724 TaskStatus::Blocked,
3725 TaskStatus::Parked,
3726 ] {
3727 t.status = status;
3728 let before = serde_json::to_value(&t).unwrap();
3729 assert!(t.set_implementers(1, &cfg).is_err());
3730 assert_eq!(serde_json::to_value(&t).unwrap(), before);
3731 }
3732 }
3733
3734 fn queue() -> (tempfile::TempDir, Queue) {
3737 let dir = tempfile::tempdir().unwrap();
3738 let q = Queue::at(dir.path().join("queue"));
3739 (dir, q)
3740 }
3741
3742 #[test]
3743 fn a_stale_write_does_not_drop_the_recorded_task_chat() {
3744 let (_dir, q) = queue();
3745 let mut stale = task("chat");
3746 q.put(&mut stale).unwrap();
3747 let mut fresh = q.get(&stale.id).unwrap();
3748 fresh.chat = Some("talk-1".to_owned());
3749 q.put(&mut fresh).unwrap();
3750 q.put(&mut stale).unwrap();
3751 assert_eq!(q.get(&stale.id).unwrap().chat.as_deref(), Some("talk-1"));
3752 }
3753
3754 #[test]
3755 fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
3756 let dir = tempfile::tempdir().unwrap();
3757 let q = Queue::at(dir.path().join("queue"));
3758 let mut t = task("held");
3759 q.put(&mut t).unwrap();
3760 assert_eq!(
3761 crate::notices::Notices::at(dir.path().join("notifications"))
3762 .list()
3763 .len(),
3764 0
3765 );
3766 t.hold_machine(Some("out of attempts".to_owned()));
3767 q.put(&mut t).unwrap();
3768 let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
3769 assert_eq!(listed.len(), 1);
3770 assert!(listed[0].message.contains("out of attempts"));
3771 }
3772
3773 fn task(title: &str) -> Task {
3774 Task::new(
3775 title.to_owned(),
3776 format!("do {title}"),
3777 PathBuf::from("."),
3778 Source::Human,
3779 )
3780 }
3781
3782 #[test]
3783 fn attempt_limit_inherits_old_records_and_controls_failure() {
3784 let mut t = task("budget");
3785 let mut old = serde_json::to_value(&t).unwrap();
3786 old.as_object_mut().unwrap().remove("attempt_limit");
3787 old["schema"] = serde_json::json!(15);
3788 let legacy: Task = serde_json::from_value(old).unwrap();
3789 assert_eq!(legacy.attempt_limit, None);
3790 assert_eq!(legacy.effective_attempt_limit(2), 2);
3791 t.attempts = 2;
3792 t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
3793 t.fail("retry", 1);
3794 assert_eq!(t.status, TaskStatus::Failed);
3795 t.attempts = 3;
3796 t.fail("stop", 10);
3797 assert_eq!(t.status, TaskStatus::Held);
3798 t.change_attempt_limit(None).unwrap();
3799 assert_eq!(t.status, TaskStatus::Held);
3800 assert_eq!(t.attempts, 3);
3801 t.fail("inherit", 4);
3802 assert_eq!(t.status, TaskStatus::Failed);
3803 for status in [
3804 TaskStatus::Running,
3805 TaskStatus::Done,
3806 TaskStatus::Failed,
3807 TaskStatus::Blocked,
3808 TaskStatus::Parked,
3809 ] {
3810 t.status = status;
3811 assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
3812 assert_eq!(t.attempt_limit, None);
3813 }
3814 }
3815
3816 #[test]
3817 fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
3818 let mut t = task("retried");
3819 assert!(
3820 t.earlier_attempts().is_empty(),
3821 "a first attempt takes nothing over"
3822 );
3823 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3824 assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
3825 assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
3827 assert_eq!(t.successor_of("bbbb"), None);
3828 assert_eq!(t.successor_of("zzzz"), None);
3829 }
3830
3831 #[test]
3832 fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
3833 let (_dir, q) = queue();
3834 let mut t = task("resumed");
3835 for (runs, successor) in [
3836 (vec!["aaaa", "aaaa"], None),
3837 (vec!["aaaa", "bbbb", "aaaa"], None),
3838 (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
3839 (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
3840 ] {
3841 t.runs = runs.into_iter().map(str::to_owned).collect();
3842 q.put(&mut t).unwrap();
3843
3844 assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
3845 assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
3846 assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
3847 }
3848 }
3849
3850 #[test]
3851 fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
3852 let (_dir, q) = queue();
3853 let mut t = task("retried");
3854 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3855 q.put(&mut t).unwrap();
3856
3857 assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
3858 assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
3859 assert_eq!(
3860 q.superseded_by("cccc"),
3861 None,
3862 "the latest attempt replaces nothing"
3863 );
3864 assert_eq!(
3865 q.superseded_by("never-heard-of-it"),
3866 None,
3867 "a run belonging to no task on this queue is not superseded"
3868 );
3869
3870 let mut by = HashMap::new();
3871 by.insert("aaaa".to_owned(), "bbbb".to_owned());
3872 by.insert("bbbb".to_owned(), "cccc".to_owned());
3873 assert_eq!(
3874 q.superseded(),
3875 by,
3876 "the whole-map and single-run forms must agree"
3877 );
3878 }
3879
3880 #[test]
3881 fn superseded_attempts_is_empty_until_the_task_is_done() {
3882 let mut t = task("retried");
3883 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3884 t.status = TaskStatus::Failed;
3885 assert_eq!(
3886 t.superseded_attempts(true),
3887 &[] as &[String],
3888 "a task still retrying has no attempt yet that a later one made moot"
3889 );
3890
3891 t.status = TaskStatus::Running;
3892 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3893 }
3894
3895 #[test]
3896 fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
3897 let mut t = task("retried");
3898 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3899 t.status = TaskStatus::Done;
3900 assert_eq!(
3901 t.superseded_attempts(true),
3902 &["aaaa".to_owned(), "bbbb".to_owned()],
3903 "cccc is the attempt whose success made the task done, and stays out"
3904 );
3905 }
3906
3907 #[test]
3908 fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
3909 let mut t = task("first try landed");
3910 t.runs = vec!["aaaa".to_owned()];
3911 t.status = TaskStatus::Done;
3912 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3913 }
3914
3915 #[test]
3916 fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
3917 let mut t = task("closed by hand after a manual merge");
3924 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3925 t.status = TaskStatus::Done;
3926 assert_eq!(
3927 t.superseded_attempts(false),
3928 &[] as &[String],
3929 "nothing here is provably why the task is done, so nothing is superseded"
3930 );
3931 }
3932
3933 #[test]
3934 fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
3935 let (_dir, q) = queue();
3936 let mut t = task("retried twice");
3937 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3938 q.put(&mut t).unwrap();
3939
3940 assert_eq!(
3941 q.latest_attempt("aaaa"),
3942 Some("cccc".to_owned()),
3943 "an old attempt points straight at the chain's current head, not the \
3944 next attempt in the middle of it"
3945 );
3946 assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
3947 assert_eq!(
3948 q.latest_attempt("cccc"),
3949 None,
3950 "the latest attempt is not superseded by anything"
3951 );
3952 assert_eq!(
3953 q.latest_attempt("never-heard-of-it"),
3954 None,
3955 "a run belonging to no task on this queue is not superseded"
3956 );
3957 }
3958
3959 #[test]
3960 fn a_markdown_heading_is_the_title_not_decoration() {
3961 assert_eq!(
3966 title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
3967 "Rework the config loader"
3968 );
3969 assert_eq!(title_from("- fix the thing", 40), "fix the thing");
3970 assert_eq!(title_from("> quoted task", 40), "quoted task");
3971 assert_eq!(title_from(" \n\n", 40), "(empty task)");
3973 assert_eq!(title_from("###\n", 40), "(empty task)");
3974 }
3975
3976 #[test]
3977 fn a_long_title_is_elided_by_characters_not_bytes() {
3978 let long = "課題".repeat(30);
3980 let title = title_from(&long, 10);
3981 assert_eq!(title.chars().count(), 10);
3982 assert!(title.ends_with('…'));
3983 }
3984
3985 #[test]
3986 fn priority_wins_and_ties_break_oldest_first() {
3987 let (_dir, q) = queue();
3988 let mut a = task("first");
3989 let mut b = task("second");
3990 let mut c = task("urgent");
3991 a.id = "20260101-000001-aaaa".to_owned();
3993 b.id = "20260101-000002-bbbb".to_owned();
3994 c.id = "20260101-000003-cccc".to_owned();
3995 c.priority = 5;
3996 for t in [&mut a, &mut b, &mut c] {
3997 q.put(t).unwrap();
3998 }
3999
4000 assert_eq!(q.next_runnable().unwrap().id, c.id);
4002 c.hold_machine(None);
4003 q.put(&mut c).unwrap();
4004 assert_eq!(q.next_runnable().unwrap().id, a.id);
4006 assert_eq!(q.list().len(), 3, "b is still waiting its turn");
4007 }
4008
4009 #[test]
4010 fn a_blocked_task_never_starves_another_runnable_one() {
4011 let (_dir, q) = queue();
4012 let mut blocked = task("blocked");
4013 blocked.block(vec!["something".to_owned()], None);
4014 q.put(&mut blocked).unwrap();
4015
4016 let mut runnable = task("free to go");
4017 q.put(&mut runnable).unwrap();
4018
4019 let next = q.next_runnable().expect("a runnable task is still offered");
4020 assert_eq!(next.id, runnable.id);
4021 }
4022
4023 #[test]
4024 fn a_held_task_is_never_offered_to_the_loop() {
4025 let (_dir, q) = queue();
4026 let mut t = task("held");
4027 q.put(&mut t).unwrap();
4028 assert!(q.next_runnable().is_some());
4029
4030 t.hold_machine(None);
4031 q.put(&mut t).unwrap();
4032 assert!(
4033 q.next_runnable().is_none(),
4034 "a held task must wait for a human"
4035 );
4036
4037 t.status = TaskStatus::Failed;
4039 q.put(&mut t).unwrap();
4040 assert!(q.next_runnable().is_some());
4041 }
4042
4043 #[test]
4044 fn attempts_are_capped_and_then_the_task_is_held() {
4045 let mut t = task("doomed");
4046
4047 t.start("run-1".to_owned());
4048 t.fail("gate red", 2);
4049 assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
4050
4051 t.start("run-2".to_owned());
4052 t.fail("gate red", 2);
4053 assert_eq!(
4054 t.status,
4055 TaskStatus::Held,
4056 "out of attempts: stop spending money on it"
4057 );
4058 assert_eq!(t.runs, ["run-1", "run-2"]);
4059 assert_eq!(t.last_error.as_deref(), Some("gate red"));
4060 assert_eq!(
4061 t.hold_reason.as_deref(),
4062 Some("gate red"),
4063 "the hold must say why, not leave hold_reason null next to a \
4064 populated last_error"
4065 );
4066 }
4067
4068 #[test]
4069 fn handing_off_a_task_records_a_hold_reason_too() {
4070 let mut t = task("left a pull request");
4071 t.start("run-1".to_owned());
4072 t.handed_off("run ended with a pull request open [run run-1]");
4073 assert_eq!(t.status, TaskStatus::Held);
4074 assert_eq!(t.hold_source, Some(HoldSource::Machine));
4075 assert_eq!(
4076 t.hold_reason.as_deref(),
4077 Some("run ended with a pull request open [run run-1]")
4078 );
4079 assert_eq!(t.hold_reason, t.last_error);
4080 }
4081
4082 #[test]
4083 fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
4084 let mut t = task("stalled by quota");
4085
4086 t.start("run-1".to_owned());
4087 assert_eq!(t.attempts, 1);
4088 t.stall("judge-1, judge-2 out of quota");
4089 assert_eq!(
4090 t.attempts, 0,
4091 "a closed quota window must not spend the task's retry budget"
4092 );
4093 assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
4094 assert_eq!(
4095 t.last_error.as_deref(),
4096 Some("judge-1, judge-2 out of quota")
4097 );
4098
4099 for _ in 0..20 {
4102 t.start("run-n".to_owned());
4103 t.stall("still out of quota");
4104 }
4105 t.start("run-real".to_owned());
4106 t.fail("gate red", 2);
4107 assert_eq!(
4108 t.status,
4109 TaskStatus::Failed,
4110 "the first attempt that was really judged is attempt one"
4111 );
4112 }
4113
4114 #[test]
4115 fn releasing_a_held_task_gives_it_a_real_second_chance() {
4116 let mut t = task("retry me");
4117 t.start("run-1".to_owned());
4118 t.fail("gate red", 1);
4119 assert_eq!(t.status, TaskStatus::Held);
4120
4121 t.release();
4122 assert_eq!(t.status, TaskStatus::Queued);
4123 assert_eq!(t.attempts, 0);
4126 assert!(t.last_error.is_none());
4127 assert_eq!(
4128 t.runs.len(),
4129 1,
4130 "history is kept: attempts reset, evidence does not"
4131 );
4132 }
4133
4134 #[test]
4135 fn a_hold_reason_survives_and_a_release_clears_it() {
4136 let mut t = task("waiting on something else");
4137 t.hold_manual(Some(
4138 "waiting for 20260101-000000-aaaa to land first".to_owned(),
4139 ));
4140 assert_eq!(t.status, TaskStatus::Held);
4141 assert_eq!(
4142 t.hold_reason.as_deref(),
4143 Some("waiting for 20260101-000000-aaaa to land first")
4144 );
4145
4146 t.hold_manual(None);
4148 assert_eq!(
4149 t.hold_reason.as_deref(),
4150 Some("waiting for 20260101-000000-aaaa to land first"),
4151 "a bare re-hold keeps whatever a human already wrote down"
4152 );
4153
4154 let mut plain = task("no reason given");
4156 plain.hold_manual(None);
4157 assert_eq!(plain.status, TaskStatus::Held);
4158 assert!(plain.hold_reason.is_none());
4159
4160 t.release();
4161 assert_eq!(t.status, TaskStatus::Queued);
4162 assert!(
4163 t.hold_reason.is_none(),
4164 "a stale reason must not greet the next person who holds this task"
4165 );
4166 }
4167
4168 #[test]
4169 fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
4170 let mut t = task("landed by hand while held");
4175 t.hold_manual(Some("waiting on 3ed9".to_owned()));
4176 assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
4177
4178 t.succeed();
4179 assert_eq!(t.status, TaskStatus::Done);
4180 assert!(
4181 t.hold_reason.is_none(),
4182 "a done task cannot still be waiting on something"
4183 );
4184 }
4185
4186 #[test]
4187 fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
4188 let mut held = task("held straight out of blocked");
4195 held.block(
4196 vec!["20260101-000000-dead".to_owned()],
4197 Some("waiting on the migration script".to_owned()),
4198 );
4199 assert_eq!(held.status, TaskStatus::Blocked);
4200
4201 held.hold_manual(None);
4202 assert_eq!(held.status, TaskStatus::Held);
4203 assert!(
4204 held.blocked_by.is_empty(),
4205 "hold overrides the wait, same as release"
4206 );
4207 assert!(held.block_reason.is_none());
4208
4209 let mut done = task("closed straight out of blocked");
4210 done.block(
4211 vec!["20260101-000000-dead".to_owned()],
4212 Some("waiting on the migration script".to_owned()),
4213 );
4214 done.succeed();
4215 assert_eq!(done.status, TaskStatus::Done);
4216 assert!(
4217 done.blocked_by.is_empty(),
4218 "a done task cannot still be waiting on a dependency"
4219 );
4220 assert!(done.block_reason.is_none());
4221 }
4222
4223 #[test]
4224 fn a_blocked_task_is_never_offered_to_the_loop() {
4225 let mut t = task("blocked");
4226 assert!(t.status.runnable());
4227 t.block(
4228 vec!["dep-id".to_owned()],
4229 Some("waits on dep-id".to_owned()),
4230 );
4231 assert_eq!(t.status, TaskStatus::Blocked);
4232 assert!(!t.status.runnable());
4233 assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
4234 }
4235
4236 #[test]
4237 fn unblocking_the_last_dependency_returns_the_task_to_queued() {
4238 let mut t = task("blocked on two");
4239 t.block(
4240 vec!["a".to_owned(), "b".to_owned()],
4241 Some("waits on a and b".to_owned()),
4242 );
4243
4244 t.unblock("a");
4245 assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
4246 assert_eq!(t.blocked_by, ["b"]);
4247
4248 t.unblock("b");
4249 assert_eq!(t.status, TaskStatus::Queued);
4250 assert!(t.blocked_by.is_empty());
4251 assert!(t.block_reason.is_none());
4252 }
4253
4254 #[test]
4255 fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
4256 let mut t = task("never blocked");
4257 t.unblock("whatever");
4258 assert_eq!(t.status, TaskStatus::Queued);
4259 }
4260
4261 #[test]
4262 fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
4263 let mut t = task("held, then asked about");
4268 t.hold_machine(Some("out of attempts".to_owned()));
4269 assert_eq!(t.status, TaskStatus::Held);
4270
4271 t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
4272 assert_eq!(t.status, TaskStatus::Blocked);
4273
4274 t.record_answer("what now?".to_owned(), "leave it held".to_owned());
4275 t.unblock("q1");
4276 assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
4277 assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
4278 assert_eq!(t.hold_source, Some(HoldSource::Machine));
4279 assert!(t.blocked_from.is_none(), "consumed once restored");
4280 }
4281
4282 #[test]
4283 fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
4284 let mut t = task("manually held, then asked about");
4285 t.hold_manual(Some("waiting on a dependency".to_owned()));
4286
4287 t.block(vec!["q1".to_owned()], None);
4288 t.unblock("q1");
4289
4290 assert_eq!(t.status, TaskStatus::Held);
4291 assert_eq!(t.hold_source, Some(HoldSource::Manual));
4292 }
4293
4294 #[test]
4295 fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
4296 let mut t = task("held, blocked twice");
4300 t.hold_machine(None);
4301 t.block(vec!["q1".to_owned()], Some("first".to_owned()));
4302 t.block(
4303 vec!["q1".to_owned(), "q2".to_owned()],
4304 Some("second".to_owned()),
4305 );
4306
4307 t.unblock("q1");
4308 assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
4309 t.unblock("q2");
4310 assert_eq!(t.status, TaskStatus::Held);
4311 }
4312
4313 #[test]
4314 fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
4315 let mut t = task("blocked mid-run");
4319 t.start("run-1".to_owned());
4320 assert_eq!(t.status, TaskStatus::Running);
4321
4322 t.block(vec!["q1".to_owned()], None);
4323 t.unblock("q1");
4324 assert_eq!(t.status, TaskStatus::Queued);
4325 }
4326
4327 #[test]
4328 fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
4329 let mut t = task("legacy record, held before it was blocked");
4335 t.hold_source = Some(HoldSource::Machine);
4336 t.hold_reason = Some("legacy hold reason".to_owned());
4337 t.status = TaskStatus::Blocked;
4338 t.blocked_by = vec!["q1".to_owned()];
4339 t.blocked_from = None;
4340
4341 t.unblock("q1");
4342 assert_eq!(t.status, TaskStatus::Held);
4343 }
4344
4345 #[test]
4346 fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
4347 let mut t = task("legacy record, ordinary dependency block");
4348 t.status = TaskStatus::Blocked;
4349 t.blocked_by = vec!["dep".to_owned()];
4350 t.blocked_from = None;
4351
4352 t.unblock("dep");
4353 assert_eq!(t.status, TaskStatus::Queued);
4354 }
4355
4356 #[test]
4357 fn answering_a_question_is_recorded_and_survives_a_release() {
4358 let mut t = task("asked something");
4359 t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
4360 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
4361 t.unblock("q1");
4362 assert_eq!(t.status, TaskStatus::Queued);
4363 assert_eq!(t.answers.len(), 1);
4364 assert_eq!(t.answers[0].answer, "SQLite");
4365
4366 t.release();
4370 assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
4371 }
4372
4373 #[test]
4374 fn resumption_review_branch_requires_unchanged_authored_criteria() {
4375 for source in ["cli", "web"] {
4376 let mut t = task("review old candidate");
4377 t.request_review("candidate-A".into());
4378 t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
4379 let mut unchanged = t.clone();
4380 unchanged.release_operator(source, None);
4381 assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
4382 t.edit(
4383 "review approved head".into(),
4384 "Reuse measured PR head B; fix Clippy; review".into(),
4385 )
4386 .unwrap();
4387 t.release_operator(source, None);
4388 assert!(t.review_branch.is_none());
4389 assert!(t.review_instruction.is_none());
4390 let mut legacy = unchanged;
4392 legacy.review_instruction = None;
4393 legacy.hold_for_handover(None, "still checked out".into());
4394 legacy
4395 .edit("new plan".into(), "Reuse measured PR head B".into())
4396 .unwrap();
4397 legacy.release_operator(source, None);
4398 assert!(legacy.review_branch.is_none());
4399 }
4400 }
4401
4402 #[test]
4403 fn a_refused_handover_keeps_the_review_branch_across_release() {
4404 let mut t = task("refused takeover");
4405 t.start("run-1".to_owned());
4406 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
4407 assert_eq!(t.status, TaskStatus::Held);
4408 assert_eq!(t.attempts, 1);
4409 t.release();
4410 assert_eq!(t.status, TaskStatus::Queued);
4411 assert_eq!(t.attempts, 0);
4412 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4413
4414 t.hold_for_handover(None, "again".to_owned());
4416 t.requeue();
4417 assert!(t.review_branch.is_none());
4418
4419 let mut m = task("manual");
4421 m.review_branch = Some("magi/x/A".to_owned());
4422 m.hold_manual(None);
4423 m.release();
4424 assert!(m.review_branch.is_none());
4425 }
4426
4427 #[test]
4428 fn requesting_review_requeues_the_task_and_remembers_the_branch() {
4429 let mut t = task("blocked run with a surviving branch");
4430 t.start("run-1".to_owned());
4431 t.fail("blocked with major findings", 5);
4432 assert_eq!(t.status, TaskStatus::Failed);
4433
4434 t.request_review("magi/eba2/A".to_owned());
4435 assert_eq!(t.status, TaskStatus::Queued);
4436 assert_eq!(t.attempts, 0);
4437 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4438
4439 t.release();
4441 assert!(t.review_branch.is_none());
4442 }
4443
4444 #[test]
4445 fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
4446 let mut t = task("retry");
4447 t.start("run-1".to_owned());
4448 t.requeue();
4449 assert!(t.fresh_start);
4450
4451 t.release();
4452 assert!(!t.fresh_start);
4453 }
4454
4455 #[test]
4456 fn urgent_only_changes_queued_or_held_and_preserves_settings() {
4457 for status in [
4458 TaskStatus::Queued,
4459 TaskStatus::Held,
4460 TaskStatus::Running,
4461 TaskStatus::Done,
4462 TaskStatus::Failed,
4463 TaskStatus::Blocked,
4464 TaskStatus::Parked,
4465 ] {
4466 let mut t = task("urgent settings");
4467 t.status = status;
4468 t.attempts = 2;
4469 t.hold_reason = Some("waiting".into());
4470 t.priority = 7;
4471 t.interrupt = true;
4472 let before = serde_json::to_value(&t).unwrap();
4473 for urgent in [true, true, false, false] {
4474 let result = t.set_urgent(urgent);
4475 let mut expected = before.clone();
4476 if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
4477 result.unwrap();
4478 expected["urgent"] = urgent.into();
4479 } else {
4480 let error = result.unwrap_err().to_string();
4481 assert!(error.contains(status.as_str()), "{error}");
4482 assert!(error.contains("queued or held"), "{error}");
4483 }
4484 assert_eq!(serde_json::to_value(&t).unwrap(), expected);
4485 }
4486 }
4487 }
4488
4489 #[test]
4490 fn priority_can_be_changed_while_queued_but_not_while_running() {
4491 let mut t = task("reprioritise me");
4492 t.set_priority(5).unwrap();
4493 assert_eq!(t.priority, 5);
4494
4495 t.start("run-1".to_owned());
4496 let err = t.set_priority(9).unwrap_err().to_string();
4497 assert!(err.contains("running"), "{err}");
4498 assert_eq!(t.priority, 5, "the rejected write must not partially apply");
4499 }
4500
4501 #[test]
4502 fn interrupt_can_be_marked_while_queued_but_not_while_running() {
4503 let mut t = task("interrupt me");
4504 assert!(!t.interrupt, "off unless asked, same as any other task");
4505
4506 t.set_interrupt(true).unwrap();
4507 assert!(t.interrupt);
4508
4509 t.start("run-1".to_owned());
4510 assert!(
4511 !t.interrupt,
4512 "the mark is one-shot: dispatching the task fulfils it, \
4513 whatever the run that follows ends up doing"
4514 );
4515 let err = t.set_interrupt(true).unwrap_err().to_string();
4516 assert!(err.contains("running"), "{err}");
4517 t.set_interrupt(false).unwrap();
4520 assert!(!t.interrupt);
4521 }
4522
4523 #[test]
4527 fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
4528 let mut t = task("interrupt me");
4529 t.set_interrupt(true).unwrap();
4530 t.start("run-1".to_owned());
4531 t.fail("mock failure", 5);
4532 assert_eq!(t.status, TaskStatus::Failed);
4533 assert!(
4534 !t.interrupt,
4535 "one attempt already spent the mark; a retry is an ordinary \
4536 requeue, not a fresh interrupt request"
4537 );
4538 }
4539
4540 #[test]
4541 fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
4542 let (_dir, q) = queue();
4543 let mut a = task("first filed");
4544 let mut b = task("second filed");
4545 a.id = "20260101-000001-aaaa".to_owned();
4546 b.id = "20260101-000002-bbbb".to_owned();
4547 q.put(&mut a).unwrap();
4548 q.put(&mut b).unwrap();
4549
4550 assert_eq!(
4551 q.next_runnable().unwrap().id,
4552 a.id,
4553 "with equal priority the older task goes first, so a burst of \
4554 new work cannot starve it"
4555 );
4556 assert_eq!(
4557 q.list()[0].id,
4558 b.id,
4559 "but the list an operator reads is newest first, the same as \
4560 before priority existed - a's turn to run does not make it the \
4561 newest task"
4562 );
4563
4564 let mut a = q.get(&a.id).unwrap();
4565 a.set_priority(10).unwrap();
4566 q.put(&mut a).unwrap();
4567
4568 assert_eq!(
4569 q.next_runnable().unwrap().id,
4570 a.id,
4571 "a raised priority must be reflected the moment it is saved"
4572 );
4573 assert_eq!(
4577 q.list()[0].id,
4578 a.id,
4579 "the raised task must sort first in the list an operator reads, \
4580 not only in next_runnable's own ordering"
4581 );
4582 }
4583
4584 #[test]
4585 fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
4586 let mut t = Task::new(
4587 "old title".to_owned(),
4588 "old instruction".to_owned(),
4589 PathBuf::from("/repo"),
4590 Source::Agent {
4591 run: "20260101-000000-beef".to_owned(),
4592 node: "implement".to_owned(),
4593 },
4594 );
4595 let id = t.id.clone();
4596 let created_at = t.created_at;
4597 t.runs.push("20260101-000000-beef".to_owned());
4598
4599 t.edit("new title".to_owned(), "new instruction".to_owned())
4600 .unwrap();
4601
4602 assert_eq!(t.title, "new title");
4603 assert_eq!(t.instruction, "new instruction");
4604 assert_eq!(t.id, id, "editing must not mint a new id");
4605 assert_eq!(t.created_at, created_at);
4606 assert_eq!(
4607 t.source,
4608 Source::Agent {
4609 run: "20260101-000000-beef".to_owned(),
4610 node: "implement".to_owned(),
4611 },
4612 "editing must not turn agent attribution into human"
4613 );
4614 assert_eq!(t.runs, ["20260101-000000-beef"]);
4615 }
4616
4617 #[test]
4618 fn editing_is_refused_once_a_task_is_running_or_finished() {
4619 let mut running = task("in flight");
4620 running.start("run-1".to_owned());
4621 let err = running
4622 .edit("x".to_owned(), "y".to_owned())
4623 .unwrap_err()
4624 .to_string();
4625 assert!(err.contains("running"), "{err}");
4626
4627 let mut done = task("finished");
4628 done.succeed();
4629 let err = done
4630 .edit("x".to_owned(), "y".to_owned())
4631 .unwrap_err()
4632 .to_string();
4633 assert!(err.contains("done"), "{err}");
4634
4635 let mut queued = task("waiting");
4637 queued.edit("x".to_owned(), "y".to_owned()).unwrap();
4638 let mut held = task("parked");
4639 held.hold_machine(None);
4640 held.edit("x".to_owned(), "y".to_owned()).unwrap();
4641 }
4642
4643 #[test]
4644 fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
4645 let (_dir, q) = queue();
4646 let path = q.path_of("20260101-000000-aaaa");
4647 std::fs::create_dir_all(q.root()).unwrap();
4648 std::fs::write(
4649 &path,
4650 serde_json::json!({
4651 "schema": SCHEMA,
4652 "id": "20260101-000000-aaaa",
4653 "title": "from before hold reasons existed",
4654 "instruction": "from before hold reasons existed",
4655 "repo": ".",
4656 "source": { "kind": "human" },
4657 "status": "held",
4658 "created_at": Timestamp::now().to_string(),
4659 "updated_at": Timestamp::now().to_string(),
4660 })
4661 .to_string(),
4662 )
4663 .unwrap();
4664
4665 let task = q.get("20260101-000000-aaaa").expect("must still read");
4666 assert!(task.hold_reason.is_none());
4667 assert!(task.operator_held());
4668 }
4669
4670 #[test]
4671 fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
4672 let (_dir, q) = queue();
4673 let path = q.path_of("20260101-000000-bbbb");
4674 std::fs::create_dir_all(q.root()).unwrap();
4675 std::fs::write(
4676 &path,
4677 serde_json::json!({
4678 "schema": 2,
4679 "id": "20260101-000000-bbbb",
4680 "title": "old manual recovery",
4681 "instruction": "old manual recovery",
4682 "repo": ".",
4683 "source": { "kind": "human" },
4684 "status": "held",
4685 "hold_reason": "active manual recovery run20260912-224242-daf5",
4686 "created_at": Timestamp::now().to_string(),
4687 "updated_at": Timestamp::now().to_string(),
4688 })
4689 .to_string(),
4690 )
4691 .unwrap();
4692
4693 let task = q.get("20260101-000000-bbbb").expect("must still read");
4694 assert_eq!(task.hold_source, None);
4695 assert!(task.operator_held());
4696 }
4697
4698 #[test]
4699 fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
4700 let (_dir, q) = queue();
4701 let path = q.path_of("20260101-000000-aaaa");
4702 std::fs::create_dir_all(q.root()).unwrap();
4703 std::fs::write(
4704 &path,
4705 serde_json::json!({
4706 "schema": SCHEMA,
4707 "id": "20260101-000000-aaaa",
4708 "title": "from before diagnostics existed",
4709 "instruction": "from before diagnostics existed",
4710 "repo": ".",
4711 "source": { "kind": "human" },
4712 "status": "held",
4713 "created_at": Timestamp::now().to_string(),
4714 "updated_at": Timestamp::now().to_string(),
4715 })
4716 .to_string(),
4717 )
4718 .unwrap();
4719
4720 let task = q.get("20260101-000000-aaaa").expect("must still read");
4721 assert!(task.diagnostic.is_none());
4722 }
4723
4724 #[test]
4725 fn a_schema_1_task_with_no_blocking_fields_still_reads() {
4726 let (_dir, q) = queue();
4730 let path = q.path_of("20260101-000000-aaaa");
4731 std::fs::create_dir_all(q.root()).unwrap();
4732 std::fs::write(
4733 &path,
4734 serde_json::json!({
4735 "schema": 1,
4736 "id": "20260101-000000-aaaa",
4737 "title": "from before blocking existed",
4738 "instruction": "from before blocking existed",
4739 "repo": ".",
4740 "source": { "kind": "human" },
4741 "status": "queued",
4742 "created_at": Timestamp::now().to_string(),
4743 "updated_at": Timestamp::now().to_string(),
4744 })
4745 .to_string(),
4746 )
4747 .unwrap();
4748
4749 let task = q.get("20260101-000000-aaaa").expect("must still read");
4750 assert!(task.blocked_by.is_empty());
4751 assert!(task.block_reason.is_none());
4752 assert!(task.answers.is_empty());
4753 assert!(task.review_branch.is_none());
4754 }
4755
4756 #[test]
4757 fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
4758 let mut held = task("diagnosed");
4763 held.start("run-1".to_owned());
4764 held.fail("gate red", 1);
4765 held.diagnostic = Some("cargo test failed: ...".to_owned());
4766 assert_eq!(held.status, TaskStatus::Held);
4767
4768 held.release();
4769 assert!(held.diagnostic.is_none());
4770
4771 held.diagnostic = Some("cargo test failed: ...".to_owned());
4772 held.succeed();
4773 assert!(held.diagnostic.is_none());
4774 }
4775
4776 #[test]
4777 fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
4778 let mut t = task("retried");
4779 t.start("run-1".to_owned());
4780 t.diagnostic = Some("stale evidence from a previous hold".to_owned());
4781 t.fail("unrelated config error", 5);
4782 assert_eq!(t.status, TaskStatus::Failed);
4783 assert!(
4784 t.diagnostic.is_none(),
4785 "fail() must not let an old diagnostic outlive the run that produced it"
4786 );
4787 }
4788
4789 #[test]
4790 fn a_claim_is_exclusive_and_releases_on_drop() {
4791 let (_dir, q) = queue();
4792 let mut t = task("contended");
4793 q.put(&mut t).unwrap();
4794
4795 let held = q.claim(&t.id).unwrap();
4796 assert!(
4797 q.claim(&t.id).is_err(),
4798 "two daemons must not drive one task into two runs"
4799 );
4800 drop(held);
4801 assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
4802 }
4803
4804 #[test]
4805 fn a_round_trip_survives_disk() {
4806 let (_dir, q) = queue();
4807 let mut t = Task::new(
4808 "titled".to_owned(),
4809 "body".to_owned(),
4810 PathBuf::from("/repo"),
4811 Source::Agent {
4812 run: "20260101-000000-beef".to_owned(),
4813 node: "implement".to_owned(),
4814 },
4815 );
4816 t.priority = 3;
4817 q.put(&mut t).unwrap();
4818
4819 let back = q.get(&t.id).unwrap();
4820 assert_eq!(back.id, t.id);
4821 assert_eq!(back.priority, 3);
4822 assert_eq!(back.source.label(), "implement@beef");
4823 assert_eq!(q.get(t.short()).unwrap().id, t.id);
4825 }
4826
4827 #[test]
4828 fn an_unreadable_task_does_not_take_the_queue_down() {
4829 let (_dir, q) = queue();
4830 let mut t = task("fine");
4831 q.put(&mut t).unwrap();
4832 std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
4833
4834 let listed = q.list();
4835 assert_eq!(listed.len(), 1, "the readable task still lists");
4836 assert_eq!(listed[0].id, t.id);
4837 }
4838
4839 #[test]
4840 fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
4841 let (_dir, q) = queue();
4842 let path = q.path_of("20260101-000000-aaaa");
4843 std::fs::create_dir_all(q.root()).unwrap();
4844 std::fs::write(
4845 &path,
4846 serde_json::json!({
4847 "schema": SCHEMA,
4848 "id": "20260101-000000-aaaa",
4849 "title": "from before solo existed",
4850 "instruction": "from before solo existed",
4851 "repo": ".",
4852 "source": { "kind": "human" },
4853 "status": "queued",
4854 "created_at": Timestamp::now().to_string(),
4855 "updated_at": Timestamp::now().to_string(),
4856 })
4857 .to_string(),
4858 )
4859 .unwrap();
4860
4861 let task = q.get("20260101-000000-aaaa").expect("must still read");
4862 assert!(!task.solo, "a queue file with no `solo` field means false");
4863 }
4864
4865 #[test]
4866 fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
4867 let (_dir, q) = queue();
4868 let path = q.path_of("20260101-000000-bbbb");
4869 std::fs::create_dir_all(q.root()).unwrap();
4870 std::fs::write(
4871 &path,
4872 serde_json::json!({
4873 "schema": SCHEMA,
4874 "id": "20260101-000000-bbbb",
4875 "title": "from before urgent existed",
4876 "instruction": "from before urgent existed",
4877 "repo": ".",
4878 "source": { "kind": "human" },
4879 "status": "queued",
4880 "created_at": Timestamp::now().to_string(),
4881 "updated_at": Timestamp::now().to_string(),
4882 })
4883 .to_string(),
4884 )
4885 .unwrap();
4886
4887 let task = q.get("20260101-000000-bbbb").expect("must still read");
4888 assert!(
4889 !task.urgent,
4890 "a queue file with no `urgent` field means false, same as `solo`"
4891 );
4892 }
4893
4894 #[test]
4895 fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
4896 let (_dir, q) = queue();
4897 let mut t = task("from the future");
4898 q.put(&mut t).unwrap();
4899 let path = q.path_of(&t.id);
4900 let body = std::fs::read_to_string(&path)
4901 .unwrap()
4902 .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
4903 std::fs::write(&path, body).unwrap();
4904
4905 let err = q.get(&t.id).unwrap_err().to_string();
4906 assert!(err.contains("schema 99"), "{err}");
4907 }
4908
4909 #[test]
4910 fn revision_moves_when_the_queue_changes() {
4911 let (_dir, q) = queue();
4912 assert_eq!(q.revision(), 0, "an empty queue has no revision");
4913 let mut t = task("first");
4914 q.put(&mut t).unwrap();
4915 assert!(q.revision() > 0, "a written task moves the revision");
4916 }
4917
4918 #[test]
4919 fn revision_moves_when_deleting_an_older_task() {
4920 let (dir, q) = queue();
4921 let questions = Questions::at(dir.path().join("questions"));
4922 let mut t1 = task("older");
4923 q.put(&mut t1).unwrap();
4924 std::thread::sleep(std::time::Duration::from_millis(10));
4926 let mut t2 = task("newer");
4927 q.put(&mut t2).unwrap();
4928
4929 let rev_before = q.revision();
4930 q.remove(&t1.id, false, &questions).unwrap();
4931 let rev_after = q.revision();
4932
4933 assert_ne!(
4934 rev_before, rev_after,
4935 "deleting an older task must change the revision so other clients see the deletion"
4936 );
4937 }
4938
4939 #[test]
4940 fn removing_a_task_takes_it_out_of_the_listing() {
4941 let (dir, q) = queue();
4942 let questions = Questions::at(dir.path().join("questions"));
4943 let mut t = task("delete me");
4944 q.put(&mut t).unwrap();
4945 let removed = q.remove(t.short(), false, &questions).unwrap();
4946 assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
4947 assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
4948 assert!(q.list().is_empty());
4949 assert!(
4950 q.remove(&t.id, false, &questions).is_err(),
4951 "removing twice is an error"
4952 );
4953 }
4954
4955 #[test]
4956 fn removing_a_task_takes_its_stale_lock_with_it() {
4957 let (dir, q) = queue();
4958 let questions = Questions::at(dir.path().join("questions"));
4959 let mut t = task("interrupted");
4960 q.put(&mut t).unwrap();
4961
4962 let claim = q.claim(&t.id).unwrap();
4965 std::mem::forget(claim);
4966 assert!(
4967 q.claim(&t.id).is_err(),
4968 "the orphaned lock is what makes the task look claimed"
4969 );
4970
4971 let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
4973 assert!(err.contains("live daemon"), "{err}");
4974 assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
4975
4976 q.remove(&t.id, false, &questions).unwrap();
4978 assert!(q.list().is_empty());
4979 let mut again = task("interrupted");
4980 again.id = t.id.clone();
4981 q.put(&mut again).unwrap();
4982 assert!(
4983 q.claim(&t.id).is_ok(),
4984 "a task that comes back must be claimable, which a left-behind lock would prevent"
4985 );
4986 }
4987
4988 fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
4989 crate::notices::Notices::at(dir.join("notifications")).list()
4990 }
4991
4992 #[test]
4993 fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
4994 let (dir, q) = queue();
4995 let questions = Questions::at(dir.path().join("questions"));
4996 let mut dep = task("dependency");
4997 q.put(&mut dep).unwrap();
4998 let mut blocked = task("waiting");
4999 blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
5000 q.put(&mut blocked).unwrap();
5001
5002 let removed = q.remove(&dep.id, false, &questions).unwrap();
5003 assert_eq!(removed.released, [blocked.id.clone()]);
5004 assert!(removed.still_blocked.is_empty());
5005
5006 let after = q.get(&blocked.id).unwrap();
5007 assert_eq!(after.status, TaskStatus::Queued);
5008 assert!(after.blocked_by.is_empty());
5009 assert!(after.block_reason.is_none());
5010 let notes = notices_of(dir.path());
5011 assert_eq!(notes.len(), 1, "{notes:?}");
5012 assert_eq!(notes[0].severity, crate::notices::Severity::Info);
5013 assert_eq!(
5014 notes[0].link,
5015 Some(crate::notices::Link::Task {
5016 id: blocked.id.clone(),
5017 })
5018 );
5019 }
5020
5021 #[test]
5022 fn removing_one_of_two_dependencies_keeps_the_other() {
5023 let (dir, q) = queue();
5024 let questions = Questions::at(dir.path().join("questions"));
5025 let mut dep = task("dependency");
5026 q.put(&mut dep).unwrap();
5027 let mut other = task("other");
5028 q.put(&mut other).unwrap();
5029 let mut blocked = task("waiting");
5030 blocked.block(
5031 vec![dep.id.clone(), other.id.clone()],
5032 Some("waits on both".to_owned()),
5033 );
5034 q.put(&mut blocked).unwrap();
5035
5036 let removed = q.remove(&dep.id, false, &questions).unwrap();
5037 assert!(removed.released.is_empty());
5038 assert_eq!(removed.still_blocked, [blocked.id.clone()]);
5039
5040 let after = q.get(&blocked.id).unwrap();
5041 assert_eq!(after.status, TaskStatus::Blocked);
5042 assert_eq!(after.blocked_by, [other.id.clone()]);
5043 assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
5046 }
5047
5048 #[test]
5049 fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
5050 let (dir, q) = queue();
5051 let questions = Questions::at(dir.path().join("questions"));
5052 let mut dep = task("dependency");
5053 q.put(&mut dep).unwrap();
5054 let mut blocked = task("waiting");
5055 blocked.block(vec![dep.id.clone()], None);
5056 q.put(&mut blocked).unwrap();
5057
5058 let claim = q.claim(&blocked.id).unwrap();
5060 let removed = q.remove(&dep.id, false, &questions).unwrap();
5061 assert!(removed.released.is_empty());
5062 drop(claim);
5063
5064 let mut task = q.get(&blocked.id).unwrap();
5065 assert_eq!(task.status, TaskStatus::Blocked);
5066 assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
5067 assert_eq!(
5068 q.apply_deleted_blockers(&mut task, &questions),
5069 [dep.id.clone()]
5070 );
5071 assert_eq!(task.status, TaskStatus::Queued);
5072 }
5073
5074 #[test]
5075 fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
5076 let (dir, q) = queue();
5077 let questions = Questions::at(dir.path().join("questions"));
5078 let mut dep = task("dependency");
5079 dep.id = "20261013-193611-ca5b".into();
5080 q.put(&mut dep).unwrap();
5081 let mut waiting = task("waiting");
5082 waiting.block(vec!["ca5b".into()], None);
5083 q.put(&mut waiting).unwrap();
5084 q.remove(&dep.id, false, &questions).unwrap();
5085 assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
5086 assert_eq!(
5087 dependency(&q, &questions, "ca5b"),
5088 Dependency::Deleted(dep.id)
5089 );
5090 }
5091
5092 #[test]
5093 fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
5094 let (dir, queue) = queue();
5095 let questions = Questions::at(dir.path().join("questions"));
5096 let mut dep = task("dependency");
5097 dep.id = "20261013-193611-ca5b".into();
5098 queue.put(&mut dep).unwrap();
5099 std::fs::create_dir_all(questions.root()).unwrap();
5100 let question_id = "20261014-193611-ca5b";
5101 std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
5102 assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
5103 assert_eq!(
5104 dependency(&queue, &questions, "ca5b"),
5105 Dependency::Ambiguous
5106 );
5107 assert_eq!(
5108 dependency(&queue, &questions, &dep.id),
5109 Dependency::Task(dep.id)
5110 );
5111 assert_eq!(
5112 dependency(&queue, &questions, question_id),
5113 Dependency::Question(question_id.into())
5114 );
5115 assert_eq!(
5116 dependency(&queue, &questions, "missing"),
5117 Dependency::Missing
5118 );
5119 }
5120
5121 #[test]
5122 fn dependency_resolution_directory_errors_are_unknown_not_missing() {
5123 let dir = tempfile::tempdir().unwrap();
5124 let q = Queue::at(dir.path().join("not-a-directory"));
5125 std::fs::write(q.root(), "file").unwrap();
5126 let questions = Questions::at(dir.path().join("questions"));
5127 assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
5128 assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
5129 }
5130
5131 #[test]
5132 fn a_dependency_without_a_tombstone_is_still_missing() {
5133 let (dir, q) = queue();
5134 let questions = Questions::at(dir.path().join("questions"));
5135 let mut dep = task("dependency");
5136 q.put(&mut dep).unwrap();
5137 std::fs::remove_file(q.path_of(&dep.id)).unwrap();
5138 assert_eq!(
5139 missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
5140 [dep.id.clone()]
5141 );
5142 assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
5143 }
5144
5145 #[test]
5146 fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
5147 let mut t = task("held");
5148 t.hold_machine(Some("because".to_owned()));
5149 t.blocked_by = vec!["gone".to_owned()];
5150 assert!(!t.dependency_deleted("gone"));
5151 assert_eq!(t.status, TaskStatus::Held);
5152 }
5153
5154 #[test]
5155 fn a_failed_record_removal_takes_the_tombstone_back() {
5156 let (_dir, q) = queue();
5157 let mut t = task("stays");
5158 q.put(&mut t).unwrap();
5159 q.write_tombstone(&t.id).unwrap();
5160 let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
5161 assert!(err.is_err());
5162 assert!(!q.was_deleted(&t.id));
5165 }
5166
5167 fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
5168 let p = dir.join(name);
5169 std::fs::write(&p, body).unwrap();
5170 p
5171 }
5172
5173 #[test]
5174 fn an_attachment_copy_survives_deleting_its_source() {
5175 let (dir, q) = queue();
5176 let src = source_file(dir.path(), "shot.png", "pixels");
5177 let mut t = task("with a picture");
5178 let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
5179 q.put(&mut t).unwrap();
5180 std::fs::remove_file(&src).unwrap();
5181 assert_eq!(names, ["shot.png"]);
5182 let loaded = q.get(&t.id).unwrap();
5183 let paths = q.attachment_paths(&loaded);
5184 assert_eq!(paths.len(), 1);
5185 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
5186 }
5187
5188 #[test]
5189 fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
5190 let (dir, q) = queue();
5191 let mut t = task("bad names");
5192 for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
5193 let src = source_file(dir.path(), name, "x");
5194 assert!(
5195 q.attach(&mut t, &[src]).is_err(),
5196 "`{name}` must be refused"
5197 );
5198 }
5199 assert!(!crate::ask::valid_asset_name("C:foo.png"));
5203 #[cfg(not(windows))]
5204 {
5205 let src = source_file(dir.path(), "C:foo.png", "x");
5206 assert!(q.attach(&mut t, &[src]).is_err());
5207 }
5208 let long = format!("{}.png", "a".repeat(70));
5209 let src = source_file(dir.path(), &long, "x");
5210 assert!(q.attach(&mut t, &[src]).is_err());
5211 assert!(t.attachments.is_empty());
5212 assert!(!q.attachments_dir(&t.id).exists());
5213 }
5214
5215 #[test]
5216 fn a_taken_attachment_name_is_numbered_not_overwritten() {
5217 let (dir, q) = queue();
5218 let a = source_file(dir.path(), "shot.png", "one");
5219 let sub = dir.path().join("other");
5220 std::fs::create_dir_all(&sub).unwrap();
5221 let b = source_file(&sub, "shot.png", "two");
5222 let mut t = task("collision");
5223 q.attach(&mut t, &[a]).unwrap();
5224 q.attach(&mut t, &[b]).unwrap();
5225 assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
5226 let paths = q.attachment_paths(&t);
5227 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
5228 assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
5229 }
5230
5231 #[test]
5232 fn a_renumbered_name_stays_inside_the_length_bound() {
5233 let (dir, q) = queue();
5234 let name = format!("{}.png", "a".repeat(60));
5235 assert_eq!(name.len(), 64);
5236 let a = source_file(dir.path(), &name, "one");
5237 let sub = dir.path().join("other");
5238 std::fs::create_dir_all(&sub).unwrap();
5239 let b = source_file(&sub, &name, "two");
5240 let mut t = task("long");
5241 q.attach(&mut t, &[a, b]).unwrap();
5242 assert_eq!(t.attachments.len(), 2);
5243 assert!(
5244 t.attachments
5245 .iter()
5246 .all(|n| crate::ask::valid_asset_name(n))
5247 );
5248 assert!(t.attachments[1].ends_with("-2.png"));
5249 }
5250
5251 #[test]
5252 fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
5253 let (dir, q) = queue();
5254 let good = source_file(dir.path(), "good.png", "ok");
5255 let mut t = task("partial");
5256 q.attach(&mut t, &[good]).unwrap();
5257 let more = source_file(dir.path(), "more.png", "ok");
5258 let missing = dir.path().join("missing.png");
5259 assert!(q.attach(&mut t, &[more, missing]).is_err());
5260 assert_eq!(t.attachments, ["good.png"]);
5261 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
5262 .unwrap()
5263 .flatten()
5264 .collect();
5265 assert_eq!(on_disk.len(), 1);
5266 }
5267
5268 fn block_put(q: &Queue, t: &Task) -> PathBuf {
5270 let tmp = q.path_of(&t.id).with_extension("json.tmp");
5271 std::fs::create_dir_all(&tmp).unwrap();
5272 tmp
5273 }
5274
5275 #[test]
5276 fn a_failed_put_leaves_no_new_attachment_directory() {
5277 let (dir, q) = queue();
5278 let mut t = task("fresh");
5279 let tmp = block_put(&q, &t);
5280 let src = source_file(dir.path(), "shot.png", "x");
5281 assert!(q.attach_and_put(&mut t, &[src]).is_err());
5282 assert!(t.attachments.is_empty());
5283 assert!(!q.attachments_dir(&t.id).exists());
5284 assert!(!q.path_of(&t.id).exists());
5285 std::fs::remove_dir(tmp).unwrap();
5286 }
5287
5288 #[test]
5289 fn a_failed_put_removes_only_the_copy_it_just_made() {
5290 let (dir, q) = queue();
5291 let mut t = task("edited");
5292 let first = source_file(dir.path(), "first.png", "1");
5293 q.attach_and_put(&mut t, &[first]).unwrap();
5294 block_put(&q, &t);
5295 let second = source_file(dir.path(), "second.png", "2");
5296 assert!(q.attach_and_put(&mut t, &[second]).is_err());
5297 assert_eq!(t.attachments, ["first.png"]);
5298 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
5299 .unwrap()
5300 .flatten()
5301 .map(|e| e.file_name().to_string_lossy().into_owned())
5302 .collect();
5303 assert_eq!(on_disk, ["first.png"]);
5304 assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
5305 }
5306
5307 #[test]
5308 fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
5309 let (dir, q) = queue();
5310 let questions = Questions::at(dir.path().join("questions"));
5311 let gone = task("gone");
5312 let mut other = task("other");
5313 let mut live = task("live");
5314 q.put(&mut other).unwrap();
5315 q.put(&mut live).unwrap();
5316 let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
5319 std::fs::create_dir_all(&orphan).unwrap();
5320 std::fs::write(orphan.join("shot.png"), "x").unwrap();
5321 let busy = q.root.join(format!("{}.attachments.removing", live.id));
5324 std::fs::create_dir_all(&busy).unwrap();
5325
5326 q.remove(&other.id, false, &questions).unwrap();
5327 assert!(!orphan.exists(), "an orphan is swept");
5328 assert!(busy.exists(), "a removal in progress is left alone");
5329 }
5330
5331 #[test]
5332 fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
5333 let (dir, q) = queue();
5334 let questions = Questions::at(dir.path().join("questions"));
5335 let mut t = task("stuck");
5336 let src = source_file(dir.path(), "shot.png", "x");
5337 q.attach_and_put(&mut t, &[src]).unwrap();
5338 let aside = q.root.join(format!("{}.attachments.removing", t.id));
5341 std::fs::create_dir_all(&aside).unwrap();
5342 std::fs::write(aside.join("old.png"), "o").unwrap();
5343 assert!(q.remove(&t.id, false, &questions).is_err());
5344 assert!(q.path_of(&t.id).exists());
5345 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5346 }
5347
5348 #[test]
5349 fn a_failed_record_removal_puts_the_attachments_back() {
5350 let (dir, q) = queue();
5351 let mut t = task("rollback");
5352 let src = source_file(dir.path(), "shot.png", "x");
5353 q.attach_and_put(&mut t, &[src]).unwrap();
5354 let err = q
5355 .remove_record_with_attachments(&t.id, |_| {
5356 Err(std::io::Error::other("injected failure"))
5357 })
5358 .unwrap_err();
5359 assert!(format!("{err:#}").contains("injected failure"));
5360 assert!(q.path_of(&t.id).exists());
5361 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5362 assert!(
5363 !q.root
5364 .join(format!("{}.attachments.removing", t.id))
5365 .exists()
5366 );
5367 }
5368
5369 #[test]
5370 fn editing_a_task_keeps_its_attachments() {
5371 let (dir, q) = queue();
5372 let src = source_file(dir.path(), "shot.png", "x");
5373 let mut t = task("editable");
5374 q.attach(&mut t, &[src]).unwrap();
5375 t.edit("new".to_owned(), "new text".to_owned()).unwrap();
5376 q.put(&mut t).unwrap();
5377 assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
5378 }
5379
5380 #[test]
5381 fn removing_a_task_deletes_its_attachments() {
5382 let (dir, q) = queue();
5383 let questions = Questions::at(dir.path().join("questions"));
5384 let src = source_file(dir.path(), "shot.png", "x");
5385 let mut t = task("doomed");
5386 q.attach(&mut t, &[src]).unwrap();
5387 q.put(&mut t).unwrap();
5388 assert!(q.attachments_dir(&t.id).is_dir());
5389 q.remove(&t.id, false, &questions).unwrap();
5390 assert!(!q.attachments_dir(&t.id).exists());
5391 assert!(q.list().is_empty());
5392 }
5393
5394 #[test]
5395 fn a_task_written_before_attachments_still_reads() {
5396 let (_dir, q) = queue();
5397 let mut t = task("old");
5398 q.put(&mut t).unwrap();
5399 let path = q.path_of(&t.id);
5400 let mut v: serde_json::Value =
5401 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
5402 v.as_object_mut().unwrap().remove("attachments");
5403 std::fs::write(&path, v.to_string()).unwrap();
5404 assert!(q.get(&t.id).unwrap().attachments.is_empty());
5405 }
5406
5407 #[test]
5408 fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
5409 let q = Queue::at(PathBuf::from("relative-queue"));
5410 let mut t = task("rel");
5411 t.attachments.push("shot.png".to_owned());
5412 let paths = q.attachment_paths(&t);
5413 assert!(paths[0].is_absolute(), "{}", paths[0].display());
5414 assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
5415 }
5416
5417 fn upgrade_parked(id: &str, started: &str) -> Task {
5418 let mut t = task(id);
5419 t.id = id.into();
5420 t.start(format!("run-{id}"));
5421 t.park("upgrade");
5422 t.record_pass_at(
5423 &format!("run-{id}"),
5424 0,
5425 Some(started.parse().unwrap()),
5426 1,
5427 Some(crate::run::ParkCause::Upgrade {
5428 from: "old".into(),
5429 to: None,
5430 }),
5431 );
5432 t
5433 }
5434
5435 #[test]
5436 fn upgrade_resume_order_follows_dispatch_claim_not_graph_start() {
5437 let stamp = |id: &str, dispatched: &str, started: &str| {
5439 let mut t = task(id);
5440 t.id = id.into();
5441 t.start(format!("run-{id}"));
5442 t.park("upgrade");
5443 t.dispatched_at = Some(dispatched.parse().unwrap());
5444 t.record_pass_at(
5445 &format!("run-{id}"),
5446 0,
5447 Some(started.parse().unwrap()),
5448 1,
5449 Some(crate::run::ParkCause::Upgrade {
5450 from: "old".into(),
5451 to: None,
5452 }),
5453 );
5454 t
5455 };
5456 let a = stamp("a", "2026-10-22T00:00:01Z", "2026-10-22T00:00:09Z");
5457 let b = stamp("b", "2026-10-22T00:00:02Z", "2026-10-22T00:00:03Z");
5458 assert!(dispatch_order(&a, &b).is_lt());
5459 }
5460
5461 #[test]
5462 fn upgrade_resume_order_survives_queue_reload_and_reversed_settlement() {
5463 let (_dir, q) = queue();
5464 let mut later = upgrade_parked("later", "2026-10-22T00:00:02Z");
5465 let mut earlier = upgrade_parked("earlier", "2026-10-22T00:00:01Z");
5466 earlier.priority = -10;
5467 let mut fresh = task("fresh");
5468 fresh.id = "newer".into();
5469 fresh.priority = 100;
5470 for t in [&mut fresh, &mut later, &mut earlier] {
5471 q.put(t).unwrap();
5472 }
5473 let reloaded = Queue::at(q.root().to_owned());
5474 let mut tasks = reloaded.list();
5475 tasks.sort_unstable_by(dispatch_order);
5476 assert_eq!(
5477 tasks.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
5478 vec!["earlier", "later", "newer"]
5479 );
5480 assert_eq!(reloaded.next_runnable().unwrap().id, "earlier");
5481 assert_eq!(reloaded.get("earlier").unwrap().attempts, 0);
5482 }
5483
5484 #[test]
5485 fn upgrade_resume_legacy_order_uses_settlement_and_id() {
5486 let mut a = upgrade_parked("a", "2026-10-22T00:00:02Z");
5487 let mut b = upgrade_parked("b", "2026-10-22T00:00:01Z");
5488 for t in [&mut a, &mut b] {
5489 let mut value = serde_json::to_value(&*t).unwrap();
5490 value["passes"][0]
5491 .as_object_mut()
5492 .unwrap()
5493 .remove("started_at");
5494 *t = serde_json::from_value(value).unwrap();
5495 }
5496 assert!(dispatch_order(&a, &b).is_lt());
5497 b.passes[0].settled_at = a.passes[0].settled_at;
5498 assert!(dispatch_order(&a, &b).is_lt());
5499 }
5500
5501 #[test]
5502 fn upgrade_resume_key_excludes_other_parks_and_explicit_task_changes() {
5503 let original = upgrade_parked("a", "2026-10-22T00:00:01Z");
5504 assert!(original.upgrade_resume_key().is_some());
5505 for cause in [
5506 crate::run::ParkCause::Operator,
5507 crate::run::ParkCause::Question,
5508 crate::run::ParkCause::Interrupt {
5509 reason: "interrupt".into(),
5510 },
5511 crate::run::ParkCause::Unspecified,
5512 ] {
5513 let mut t = original.clone();
5514 t.passes[0].parked = Some(cause);
5515 assert!(t.upgrade_resume_key().is_none());
5516 let mut fresh = task("fresh");
5517 fresh.priority = 1;
5518 assert!(dispatch_order(&fresh, &t).is_lt());
5519 }
5520 let mut t = original.clone();
5521 t.release();
5522 assert!(t.upgrade_resume_key().is_none());
5523 t.edit("edited".into(), "new instruction".into()).unwrap();
5524 assert!(t.upgrade_resume_key().is_none());
5525 let mut t = original.clone();
5526 t.requeue();
5527 t.status = TaskStatus::Parked;
5528 assert!(t.upgrade_resume_key().is_none());
5529 let mut t = original.clone();
5530 t.start("different-run".into());
5531 t.park("no upgrade");
5532 assert!(t.upgrade_resume_key().is_none());
5533 }
5534
5535 #[test]
5536 fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
5537 let dir = tempfile::tempdir().unwrap();
5538 let q = Queue::at(dir.path().join("queue"));
5539 let mut t = task("resume");
5540 t.start("A".into());
5541 t.record_pass("A", 0, 1, None);
5542 t.start("B".into());
5543 q.put(&mut t).unwrap();
5544 assert_eq!(t.runs, vec!["A", "B"]);
5545 t.start("A".into());
5546 t.record_pass("A", 1, 3, None);
5547 q.put(&mut t).unwrap();
5548 let saved = q.get(&t.id).unwrap();
5549 assert_eq!(saved.runs, vec!["B", "A"]);
5550 assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
5551 assert_eq!(saved.attempts, 3);
5552 assert_eq!(saved.passes.len(), 2);
5553 assert_eq!(saved.passes[0].run, "A");
5554 assert_eq!(saved.passes[1].run, "A");
5555 }
5556
5557 #[test]
5558 fn repeated_start_records_one_run_and_separate_passes() {
5559 let mut t = task("resume");
5560 t.start("same-run".into());
5561 t.record_pass("same-run", 0, 1, None);
5562 t.start("same-run".into());
5563 t.record_pass("same-run", 1, 2, None);
5564 assert_eq!(t.runs, vec!["same-run"]);
5565 assert_eq!(t.attempts, 2);
5566 assert_eq!(t.passes.len(), 2);
5567 }
5568
5569 #[test]
5570 fn link_run_adds_a_run_once_and_touches_nothing_else() {
5571 let dir = tempfile::tempdir().unwrap();
5572 let queue = Queue::at(dir.path().join("queue"));
5573 let mut t = Task::new(
5574 "t".to_owned(),
5575 "do it".to_owned(),
5576 PathBuf::from("."),
5577 Source::Human,
5578 );
5579 queue.put(&mut t).unwrap();
5580 let before = queue.get(&t.id).unwrap();
5581
5582 let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
5583 assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
5584 assert_eq!(linked.status, before.status);
5585 assert_eq!(linked.attempts, before.attempts);
5586 assert_eq!(linked.interrupt, before.interrupt);
5587
5588 let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5589 assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
5590 assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
5591 assert!(queue.link_run("no-such-task", "r").is_err());
5592 }
5593
5594 #[test]
5595 fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
5596 let dir = tempfile::tempdir().unwrap();
5597 let queue = Queue::at(dir.path().join("queue"));
5598 let mut t = Task::new(
5599 "t".to_owned(),
5600 "do it".to_owned(),
5601 PathBuf::from("."),
5602 Source::Human,
5603 );
5604 queue.put(&mut t).unwrap();
5605 let mut snapshot = queue.get(&t.id).unwrap();
5607 queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5608
5609 snapshot.start("20260930-000000-aaaa".to_owned());
5610 queue.put(&mut snapshot).unwrap();
5611
5612 let stored = queue.get(&t.id).unwrap();
5613 assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
5614 assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
5615 assert_eq!(stored.attempts, 1);
5616 }
5617
5618 #[test]
5619 fn concurrent_links_and_daemon_saves_lose_nothing() {
5620 let dir = tempfile::tempdir().unwrap();
5621 let queue = Queue::at(dir.path().join("queue"));
5622 let mut t = Task::new(
5623 "t".to_owned(),
5624 "do it".to_owned(),
5625 PathBuf::from("."),
5626 Source::Human,
5627 );
5628 queue.put(&mut t).unwrap();
5629 let id = t.id.clone();
5630
5631 let linkers: Vec<_> = (0..4)
5632 .map(|n| {
5633 let (queue, id) = (queue.clone(), id.clone());
5634 std::thread::spawn(move || {
5635 for k in 0..10 {
5636 queue
5637 .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
5638 .unwrap();
5639 }
5640 })
5641 })
5642 .collect();
5643 let mut mine = queue.get(&id).unwrap();
5646 for k in 0..10 {
5647 mine.start(format!("20260930-000009-d{k:03}"));
5648 queue.put(&mut mine).unwrap();
5649 }
5650 for l in linkers {
5651 l.join().unwrap();
5652 }
5653
5654 let stored = queue.get(&id).unwrap();
5655 assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
5656 assert_eq!(
5657 stored.attempts, 10,
5658 "linking never rewinds the daemon's work"
5659 );
5660 }
5661
5662 fn age_lock(path: &Path) {
5663 let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
5664 f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
5665 .unwrap();
5666 }
5667
5668 #[test]
5669 fn concurrent_stale_takeover_yields_one_holder() {
5670 use std::sync::atomic::{AtomicUsize, Ordering};
5671 use std::sync::{Arc, Barrier};
5672 for _ in 0..5 {
5673 let dir = tempfile::tempdir().unwrap();
5674 let q = Queue::at(dir.path().to_path_buf());
5675 let lock = dir.path().join("t.write-lock");
5676 std::fs::write(&lock, "dead-0000").unwrap();
5677 age_lock(&lock);
5678 let n = 6;
5679 let barrier = Arc::new(Barrier::new(n));
5680 let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
5681 let handles: Vec<_> = (0..n)
5682 .map(|_| {
5683 let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
5684 std::thread::spawn(move || {
5685 b.wait();
5686 let g = q.lock_task("t").unwrap();
5687 let held = now.fetch_add(1, Ordering::SeqCst) + 1;
5688 max.fetch_max(held, Ordering::SeqCst);
5689 std::thread::sleep(std::time::Duration::from_millis(20));
5690 now.fetch_sub(1, Ordering::SeqCst);
5691 drop(g);
5692 })
5693 })
5694 .collect();
5695 for h in handles {
5696 h.join().unwrap();
5697 }
5698 assert_eq!(max.load(Ordering::SeqCst), 1);
5699 assert!(!lock.exists());
5700 }
5701 }
5702
5703 #[test]
5704 fn dropping_a_replaced_lock_leaves_the_new_holders_token() {
5705 let dir = tempfile::tempdir().unwrap();
5706 let q = Queue::at(dir.path().to_path_buf());
5707 let lock = dir.path().join("t.write-lock");
5708 let a = q.lock_task("t").unwrap();
5709 let replacement = owner_token();
5712 std::fs::write(&lock, &replacement).unwrap();
5713 drop(a);
5714 assert_eq!(std::fs::read_to_string(&lock).unwrap(), replacement);
5715 }
5716
5717 #[test]
5718 fn aged_live_task_lock_excludes_updates_until_retirement_releases_it() {
5719 let dir = tempfile::tempdir().unwrap();
5720 let q = Queue::at(dir.path().to_path_buf());
5721 let held = q.lock_task("t").unwrap();
5722 let path = dir.path().join("t.write-lock");
5723 age_lock(&path);
5724 assert!(!break_stale(&path, &held.token));
5725 let (started_tx, started_rx) = std::sync::mpsc::channel();
5726 let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
5727 let updater = std::thread::spawn(move || {
5728 started_tx.send(()).unwrap();
5729 let _write = q.lock_task("t").unwrap();
5730 acquired_tx.send(()).unwrap();
5731 });
5732 started_rx.recv().unwrap();
5733 assert!(
5734 acquired_rx
5735 .recv_timeout(std::time::Duration::from_millis(100))
5736 .is_err()
5737 );
5738 assert_eq!(std::fs::read_to_string(&path).unwrap(), held.token);
5739 drop(held);
5740 acquired_rx
5741 .recv_timeout(std::time::Duration::from_secs(2))
5742 .unwrap();
5743 updater.join().unwrap();
5744 assert!(!path.exists());
5745 }
5746
5747 #[test]
5748 fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
5749 let dir = tempfile::tempdir().unwrap();
5750 let lock = dir.path().join("t.write-lock");
5751 std::fs::write(&lock, "new-token").unwrap();
5752 assert!(!break_stale(&lock, "old-token"));
5753 assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
5754 assert!(!break_stale(&lock, "new-token"));
5756 assert!(lock.exists());
5757 age_lock(&lock);
5758 assert!(break_stale(&lock, "new-token"));
5759 assert!(!lock.exists());
5760 }
5761
5762 #[test]
5763 fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
5764 let dir = tempfile::tempdir().unwrap();
5765 let lock = dir.path().join("t.write-lock");
5766 std::fs::write(&lock, "dead-1").unwrap();
5767 age_lock(&lock);
5768 let marker = break_marker(&lock, "dead-1");
5769 std::fs::write(&marker, "crashed-remover").unwrap();
5770 assert!(!break_stale(&lock, "dead-1"));
5772 assert!(lock.exists());
5773 age_lock(&marker);
5775 assert!(break_stale(&lock, "dead-1"));
5776 assert!(!lock.exists());
5777 assert!(!marker.exists());
5778 }
5779
5780 fn chat_task() -> Task {
5781 Task::new(
5782 "t".to_owned(),
5783 "do t".to_owned(),
5784 PathBuf::from("."),
5785 Source::Agent {
5786 run: "20260902-000000-beef".to_owned(),
5787 node: CHAT_NODE.to_owned(),
5788 },
5789 )
5790 }
5791
5792 #[test]
5793 fn a_blocked_hold_keeps_its_anchor_but_a_new_run_starts_a_new_hold() {
5794 let mut t = chat_task();
5795 t.hold_machine(Some("waiting".into()));
5796 let old: Timestamp = "2020-01-01T00:00:00Z".parse().unwrap();
5797 t.held_at = Some(old);
5798 t.block(vec!["decision".into()], None);
5799 t.hold_machine(Some("same waiting decision".into()));
5800 assert_eq!(t.held_at, Some(old));
5801 t.start("real-new-run".into());
5802 t.fail("new failure", 1);
5803 assert_ne!(
5804 t.held_at,
5805 Some(old),
5806 "old answers cannot approve a new hold"
5807 );
5808 }
5809
5810 #[test]
5811 fn legacy_delivery_survives_upgrade_and_a_no_progress_run_link() {
5812 let (_dir, queue) = queue();
5813 let mut t = chat_task();
5814 t.start("old-run".into());
5815 t.hold_machine(Some("operator is considering recovery".into()));
5816 t.chat_notice = Some(ChatNotice {
5817 key: format!("held#{}", t.held_at.unwrap()),
5818 at: Timestamp::now(),
5819 outcome: ChatNoticeOutcome::Posted,
5820 });
5821 let mut raw = serde_json::to_value(&t).unwrap();
5822 raw.as_object_mut().unwrap().remove("chat_notices");
5823 raw.as_object_mut().unwrap().remove("hold_notice_cause");
5824 std::fs::create_dir_all(queue.path_of(&t.id).parent().unwrap()).unwrap();
5825 std::fs::write(queue.path_of(&t.id), serde_json::to_string(&raw).unwrap()).unwrap();
5826 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5827 queue
5828 .modify(&t.id, |task| {
5829 task.requeue();
5830 task.link_run("unprepared-run");
5831 task.startup_notice_cause("startup-source");
5832 task.fail("same unresolved preflight", 0);
5833 true
5834 })
5835 .unwrap();
5836 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5837 queue
5838 .modify(&t.id, |task| {
5839 task.succeed();
5840 true
5841 })
5842 .unwrap();
5843 assert_eq!(
5844 queue.get(&t.id).unwrap().chat_notice_due().as_deref(),
5845 Some("done")
5846 );
5847 }
5848
5849 #[test]
5850 fn a_dependency_wait_does_not_change_when_released_and_blocked_again() {
5851 let mut t = chat_task();
5852 t.block(vec!["a".into(), "b".into()], Some("waiting".into()));
5853 let key = t.chat_notice_key().unwrap();
5854 t.chat_notices.push(key);
5855 t.release();
5856 t.block(
5857 vec!["b".into(), "a".into(), "a".into()],
5858 Some("still waiting on prerequisites".into()),
5859 );
5860 assert_eq!(t.chat_notice_due(), None);
5861 t.block(
5862 vec!["new-question".into()],
5863 Some("different decision".into()),
5864 );
5865 assert!(t.chat_notice_due().is_some());
5866 }
5867
5868 #[test]
5869 fn running_startup_failures_preserve_stage_and_report_distinct_blockers() {
5870 let mut t = chat_task();
5871 t.start("existing-run".into());
5872 t.fail("implementation failed", 1);
5873 let failed = t.chat_notice_key().unwrap();
5874 t.chat_notices.push(failed.clone());
5875 t.release_operator("test", None);
5876 t.start("existing-run".into());
5877 t.fail_startup("startup-source", "same unresolved recovery source", 1);
5878 assert_eq!(
5879 t.chat_notice_due(),
5880 None,
5881 "no-progress recovery stays quiet"
5882 );
5883 t.release_operator("test", None);
5884 t.start("existing-run".into());
5885 t.fail_startup("startup-attachments", "attachment missing", 1);
5886 let attachments = t.chat_notice_due().expect("new attachment blocker");
5887 assert_ne!(attachments, failed);
5888 assert_eq!(
5889 t.hold_notice_cause.as_ref().unwrap().stage,
5890 "startup-attachments"
5891 );
5892 t.chat_notices.push(attachments);
5893 t.release_operator("test", None);
5894 t.start("existing-run".into());
5895 t.fail_startup(
5896 "startup-attachments",
5897 "same missing attachment, reworded",
5898 1,
5899 );
5900 assert_eq!(
5901 t.chat_notice_due(),
5902 None,
5903 "same stage after attempt counter reset"
5904 );
5905 t.release_operator("test", None);
5906 t.start("existing-run".into());
5907 t.fail_startup("startup-config", "different preflight stage", 1);
5908 assert!(t.chat_notice_due().is_some());
5909 assert_eq!(
5910 t.hold_notice_cause.as_ref().unwrap().stage,
5911 "startup-config"
5912 );
5913 }
5914
5915 #[test]
5916 fn structured_causes_preserve_no_progress_retries_and_report_new_passes() {
5917 let mut t = chat_task();
5918 t.hold_machine(Some("startup source missing".into()));
5919 t.startup_notice_cause("startup-source");
5920 let first = t.chat_notice_key().unwrap();
5921 t.chat_notices.push(first.clone());
5922 t.release_operator("test", None);
5923 t.startup_notice_cause("startup-source");
5924 t.hold_machine(Some("same source cannot be read".into()));
5925 assert_eq!(t.chat_notice_due(), None);
5926 t.startup_notice_cause("startup-config");
5927 assert!(t.chat_notice_due().is_some(), "different startup stage");
5928 t.notice_cause("failed", Some("existing-run"), Some(0));
5929 let run_failure = t.chat_notice_key().unwrap();
5930 t.chat_notices.push(run_failure.clone());
5931 t.release_operator("test", None);
5932 t.startup_notice_cause("startup-source");
5933 t.hold_machine(Some("no new work happened".into()));
5934 assert_eq!(t.chat_notice_due(), None);
5935 t.notice_cause("failed", Some("existing-run"), Some(1));
5936 assert!(t.chat_notice_due().is_some(), "a real resumed pass failed");
5937 assert_ne!(t.chat_notice_key().unwrap(), run_failure);
5938 }
5939
5940 #[test]
5941 fn legacy_hold_baseline_does_not_replay_or_hide_new_failures_and_completion() {
5942 let mut t = chat_task();
5943 t.hold_machine(Some("still waiting".into()));
5944 t.chat_notice = Some(ChatNotice {
5945 key: "held#2026-10-14T00:00:00Z".into(),
5946 at: Timestamp::now(),
5947 outcome: ChatNoticeOutcome::Posted,
5948 });
5949 assert_eq!(t.chat_notice_due(), None, "old hold baseline");
5950 t.import_chat_notice();
5951 t.notice_cause("failed", Some("new-run"), Some(0));
5952 assert!(t.chat_notice_due().is_some(), "new producer evidence");
5953 t.succeed();
5954 assert_eq!(t.chat_notice_due().as_deref(), Some("done"));
5955 }
5956
5957 #[test]
5958 fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
5959 let mut t = chat_task();
5960 assert_eq!(t.chat_notice_due(), None, "queued");
5961 t.start("r1".to_owned());
5962 assert_eq!(t.chat_notice_due(), None, "running");
5963 t.fail("boom", 3);
5964 assert_eq!(t.status, TaskStatus::Failed);
5965 assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
5966 t.start("r2".to_owned());
5967 t.park("upgrade");
5968 assert_eq!(t.chat_notice_due(), None, "parked");
5969 t.fail("boom", 1);
5970 assert_eq!(t.status, TaskStatus::Held);
5971 let key = t
5972 .chat_notice_due()
5973 .expect("held on the last attempt is final");
5974 t.chat_notice = Some(ChatNotice {
5975 key,
5976 at: Timestamp::now(),
5977 outcome: ChatNoticeOutcome::Posted,
5978 });
5979 assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
5980 t.release();
5981 assert_eq!(t.chat_notice_due(), None, "requeued after a release");
5982 t.start("r3".to_owned());
5983 t.succeed();
5984 assert!(t.chat_notice_due().is_some(), "done is a new final state");
5985 }
5986
5987 #[test]
5988 fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
5989 for key in ["done", "done#0", "done#1", "done#12"] {
5990 let mut t = chat_task();
5991 t.start("r1".to_owned());
5992 t.succeed();
5993 t.chat_notice = Some(ChatNotice {
5994 key: key.to_owned(),
5995 at: Timestamp::now(),
5996 outcome: ChatNoticeOutcome::Posted,
5997 });
5998 t.requeue();
5999 t.start("r2".to_owned());
6000 t.succeed();
6001 assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
6002 assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
6003 }
6004 }
6005
6006 #[test]
6007 fn a_release_without_progress_keeps_the_unresolved_notice_cause() {
6008 let mut t = chat_task();
6009 t.hold_manual(Some("wait".to_owned()));
6010 let first = t.chat_notice_due().expect("held");
6011 t.chat_notice = Some(ChatNotice {
6012 key: first.clone(),
6013 at: Timestamp::now(),
6014 outcome: ChatNoticeOutcome::Posted,
6015 });
6016 t.hold_manual(Some("still waiting".to_owned()));
6017 assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
6018 t.release();
6019 std::thread::sleep(std::time::Duration::from_millis(5));
6020 t.hold_manual(None);
6021 assert_eq!(t.chat_notice_due(), None, "no progress, same cause");
6022 }
6023
6024 #[test]
6025 fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
6026 let mut human = task("h");
6027 human.succeed();
6028 assert_eq!(human.chat_notice_due(), None);
6029 let mut inherited = task("i");
6031 inherited.origin_chat = Some("20260902-000000-beef".to_owned());
6032 inherited.succeed();
6033 assert_eq!(inherited.chat_notice_due(), None);
6034 let mut old = chat_task();
6035 old.schema = 12;
6036 old.status = TaskStatus::Done;
6037 assert_eq!(old.chat_notice_due(), None, "history is not replayed");
6038 let mut live = chat_task();
6039 live.schema = 12;
6040 live.start("r1".to_owned());
6041 live.succeed();
6042 assert!(live.chat_notice_due().is_some(), "started by this build");
6043 let mut held = chat_task();
6044 held.schema = 12;
6045 held.hold_machine(Some("config".to_owned()));
6046 assert!(held.chat_notice_due().is_some(), "held before any start");
6047 }
6048
6049 #[test]
6050 fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
6051 let tmp = tempfile::tempdir().expect("tempdir");
6052 let q = Queue::at(tmp.path().join("queue"));
6053 let mut t = chat_task();
6054 t.succeed();
6055 q.put(&mut t).expect("put");
6056 let mut stale = q.get(&t.id).expect("get");
6057 let key = t.chat_notice_due().expect("due");
6058 q.modify(&t.id, |x| {
6059 x.chat_notice = Some(ChatNotice {
6060 key: key.clone(),
6061 at: Timestamp::now(),
6062 outcome: ChatNoticeOutcome::Posted,
6063 });
6064 true
6065 })
6066 .expect("modify");
6067 q.put(&mut stale).expect("stale put");
6068 assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
6069 }
6070
6071 #[test]
6072 fn empty_sources_are_omitted_so_older_builds_still_read_the_task() {
6073 let mut t = task("plain");
6076 let value = serde_json::to_value(&t).unwrap();
6077 assert!(value.get("sources").is_none(), "{value}");
6078 t.sources.pick_commits.push(crate::refs::Input {
6079 token: "abc1234".into(),
6080 sha: "a".repeat(40),
6081 provenance: crate::refs::Provenance::ExplicitCli,
6082 });
6083 let value = serde_json::to_value(&t).unwrap();
6084 assert_eq!(value["sources"]["pick_commits"][0]["sha"], "a".repeat(40));
6085 let restored: Task = serde_json::from_value(value).unwrap();
6086 assert_eq!(restored.sources, t.sources);
6087 }
6088
6089 fn queue_with_ids(dir: &Path, ids: &[&str]) -> Queue {
6091 let ids: Vec<String> = ids.iter().map(|s| (*s).to_owned()).collect();
6092 let n = std::sync::atomic::AtomicUsize::new(0);
6093 Queue::at(dir.join("queue")).with_id_source(move || {
6094 let i = n.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
6095 ids.get(i)
6096 .cloned()
6097 .unwrap_or_else(|| format!("20260101-000000-f{i:03x}"))
6098 })
6099 }
6100
6101 const TAKEN: &str = "20260101-000000-aaaa";
6102
6103 #[test]
6104 fn a_new_task_never_merges_into_a_record_with_its_candidate_id() {
6105 let dir = tempfile::tempdir().unwrap();
6106 let q = queue_with_ids(dir.path(), &[TAKEN, TAKEN]);
6107 let mut first = task("first");
6108 first.start("run-1".into());
6109 first.succeed();
6110 q.put(&mut first).unwrap();
6111 assert_eq!(first.id, TAKEN);
6112 let stored_before = q.get(&first.id).unwrap();
6113 assert!(!stored_before.runs.is_empty());
6114
6115 let mut second = task("second");
6116 q.put(&mut second).unwrap();
6117
6118 assert_ne!(second.id, first.id);
6119 assert!(second.runs.is_empty());
6120 assert!(second.passes.is_empty());
6121 assert!(second.chat_notices.is_empty());
6122 assert!(second.chat_notice.is_none());
6123 assert_eq!(second.chat_notice_due(), task("fresh").chat_notice_due());
6124 let a = q.get(&first.id).unwrap();
6125 let b = q.get(&second.id).unwrap();
6126 assert_eq!(a.title, "first");
6127 assert_eq!(a.runs, stored_before.runs);
6128 assert_eq!(a.passes, stored_before.passes);
6129 assert_eq!(a.chat_notices, stored_before.chat_notices);
6130 assert_eq!(a.chat_notice, stored_before.chat_notice);
6131 assert_eq!(a.chat_notice_due(), stored_before.chat_notice_due());
6132 assert_eq!(b.title, "second");
6133 assert!(b.runs.is_empty());
6134 assert!(b.passes.is_empty());
6135 assert!(b.chat_notice.is_none());
6136
6137 let mut again = q.get(&second.id).unwrap();
6139 again.title = "second, edited".into();
6140 q.put(&mut again).unwrap();
6141 assert_eq!(q.get(&second.id).unwrap().title, "second, edited");
6142 assert_eq!(q.list().len(), 2);
6143 }
6144
6145 #[test]
6146 fn a_new_task_skips_an_id_with_an_orphaned_claim_or_attachments() {
6147 let dir = tempfile::tempdir().unwrap();
6148 let q = queue_with_ids(dir.path(), &["orphan-lock", "orphan-dir", "free"]);
6149 std::fs::create_dir_all(q.root()).unwrap();
6150 std::fs::write(q.root().join("orphan-lock.lock"), "x").unwrap();
6151 std::fs::create_dir_all(q.attachments_dir("orphan-dir")).unwrap();
6152 let mut t = task("t");
6153 q.put(&mut t).unwrap();
6154 assert_eq!(t.id, "free");
6155 }
6156
6157 #[test]
6158 fn a_claimed_new_task_holds_the_claim_on_the_reserved_id() {
6159 let dir = tempfile::tempdir().unwrap();
6160 let q = queue_with_ids(dir.path(), &[TAKEN, TAKEN]);
6161 let mut first = task("first");
6162 q.put(&mut first).unwrap();
6163 let mut second = task("second");
6164 let claim = q.file_claimed(&mut second).unwrap();
6165 assert_ne!(second.id, first.id);
6166 assert!(q.claim(&second.id).is_err());
6167 assert!(q.claim(&first.id).is_ok(), "the first task was not claimed");
6168 drop(claim);
6169 }
6170
6171 #[test]
6172 fn an_attached_new_task_skips_a_taken_id() {
6173 let dir = tempfile::tempdir().unwrap();
6174 let q = queue_with_ids(dir.path(), &[TAKEN, TAKEN]);
6175 let mut first = task("first");
6176 q.put(&mut first).unwrap();
6177 let src = source_file(dir.path(), "shot.png", "x");
6178 let mut second = task("second");
6179 q.attach_and_put(&mut second, &[src]).unwrap();
6180 assert_ne!(second.id, first.id);
6181 assert!(!q.attachments_dir(&first.id).exists());
6182 assert_eq!(q.attachment_paths(&second).len(), 1);
6183 assert!(q.attachment_paths(&second)[0].is_file());
6184 }
6185
6186 #[test]
6187 fn attaching_before_putting_a_new_task_reserves_the_id() {
6188 let dir = tempfile::tempdir().unwrap();
6189 let q = queue_with_ids(dir.path(), &[TAKEN, TAKEN]);
6190 let mut first = task("first");
6191 q.put(&mut first).unwrap();
6192 let src = source_file(dir.path(), "shot.png", "x");
6193 let mut second = task("second");
6194 q.attach(&mut second, &[src]).unwrap();
6195 assert_ne!(second.id, first.id);
6196 assert!(q.attachments_dir(&second.id).is_dir());
6197 q.put(&mut second).unwrap();
6198 assert_eq!(q.get(&second.id).unwrap().attachments.len(), 1);
6199 assert_eq!(q.get(&first.id).unwrap().title, "first");
6200 assert!(q.get(&first.id).unwrap().attachments.is_empty());
6201 }
6202
6203 #[test]
6204 fn create_new_keeps_its_already_exists_meaning() {
6205 let dir = tempfile::tempdir().unwrap();
6206 let q = Queue::at(dir.path().join("queue"));
6207 let mut a = task("a");
6208 a.id = "fixed-id".into();
6209 assert!(q.create_new(&mut a).unwrap());
6210 let mut b = task("b");
6211 b.id = "fixed-id".into();
6212 assert!(!q.create_new(&mut b).unwrap());
6213 assert_eq!(q.get("fixed-id").unwrap().title, "a");
6214 }
6215
6216 #[test]
6217 fn filing_gives_up_when_no_id_is_ever_free() {
6218 let dir = tempfile::tempdir().unwrap();
6219 let q = Queue::at(dir.path().join("queue")).with_id_source(|| TAKEN.to_owned());
6220 let mut first = task("first");
6221 q.put(&mut first).unwrap();
6222 let mut second = task("second");
6223 assert!(q.put(&mut second).is_err());
6224 assert_eq!(q.get(&first.id).unwrap().title, "first");
6225 }
6226}