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 = 18;
130
131#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
133#[serde(rename_all = "lowercase")]
134pub enum HoldSource {
135 Manual,
137 Machine,
139}
140
141impl HoldSource {
142 pub fn label(self) -> &'static str {
144 match self {
145 Self::Manual => "manual",
146 Self::Machine => "machine",
147 }
148 }
149}
150
151#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(tag = "kind", rename_all = "lowercase")]
155pub enum Source {
156 Human,
158 Agent {
161 run: String,
163 node: String,
165 },
166 Issue {
168 number: u64,
170 repo: String,
172 },
173}
174
175pub const CHAT_NODE: &str = "chat";
178
179impl Source {
180 pub fn label(&self) -> String {
182 match self {
183 Self::Human => "human".to_owned(),
184 Self::Agent { run, node } => format!("{node}@{}", short(run)),
185 Self::Issue { number, .. } => format!("issue #{number}"),
186 }
187 }
188}
189
190#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
192#[serde(rename_all = "lowercase")]
193pub enum TaskStatus {
194 Queued,
196 Running,
198 Done,
200 Failed,
202 Held,
204 Blocked,
208 Parked,
213}
214
215impl TaskStatus {
216 pub fn runnable(self) -> bool {
218 matches!(self, Self::Queued | Self::Failed | Self::Parked)
219 }
220
221 pub fn as_str(self) -> &'static str {
223 match self {
224 Self::Queued => "queued",
225 Self::Running => "running",
226 Self::Done => "done",
227 Self::Failed => "failed",
228 Self::Held => "held",
229 Self::Blocked => "blocked",
230 Self::Parked => "parked",
231 }
232 }
233}
234
235#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
238pub struct TaskCounts {
239 pub queued: usize,
241 pub running: usize,
243 pub done: usize,
245 pub failed: usize,
247 pub held: usize,
249 pub blocked: usize,
251 pub parked: usize,
253}
254
255impl TaskCounts {
256 pub fn of(tasks: &[Task]) -> Self {
260 Self::of_statuses(tasks.iter().map(|task| task.status))
261 }
262
263 pub(crate) fn of_statuses(statuses: impl IntoIterator<Item = TaskStatus>) -> Self {
265 let mut counts = Self::default();
266 for status in statuses {
267 match status {
268 TaskStatus::Queued => counts.queued += 1,
269 TaskStatus::Running => counts.running += 1,
270 TaskStatus::Done => counts.done += 1,
271 TaskStatus::Failed => counts.failed += 1,
272 TaskStatus::Held => counts.held += 1,
273 TaskStatus::Blocked => counts.blocked += 1,
274 TaskStatus::Parked => counts.parked += 1,
275 }
276 }
277 counts
278 }
279}
280
281#[derive(Debug, Clone, Serialize, Deserialize)]
283#[serde(deny_unknown_fields)]
284pub struct Task {
285 pub schema: u32,
287 pub id: String,
289 pub title: String,
291 pub instruction: String,
293 pub repo: PathBuf,
295 pub source: Source,
297 #[serde(default)]
299 pub priority: i32,
300 #[serde(default)]
311 pub solo: bool,
312 pub status: TaskStatus,
314 #[serde(default)]
316 pub attempts: usize,
317 #[serde(default)]
319 pub attempt_limit: Option<NonZeroUsize>,
320 #[serde(default)]
323 pub runs: Vec<String>,
324 #[serde(default)]
326 pub last_error: Option<String>,
327 #[serde(default)]
340 pub hold_reason: Option<String>,
341 #[serde(default)]
344 pub hold_source: Option<HoldSource>,
345 #[serde(default)]
357 pub diagnostic: Option<String>,
358 #[serde(default)]
368 pub blocked_by: Vec<String>,
369 #[serde(default)]
372 pub block_reason: Option<String>,
373 #[serde(default)]
386 pub blocked_from: Option<TaskStatus>,
387 #[serde(default)]
389 pub blocked_at: Option<Timestamp>,
390 #[serde(default)]
401 pub answers: Vec<AnsweredQuestion>,
402 #[serde(default)]
408 pub triage_applied: Vec<String>,
409 #[serde(default)]
415 pub actions_applied: Vec<String>,
416 #[serde(default)]
421 pub resume_override: Option<OperatorResume>,
422 #[serde(default)]
425 pub resumption: Option<Resumption>,
426 #[serde(default)]
428 pub resumption_history: Vec<Resumption>,
429 #[serde(default)]
431 pub operator_decided_at: Option<Timestamp>,
432 #[serde(default)]
444 pub review_branch: Option<String>,
445 #[serde(default)]
447 pub review_instruction: Option<String>,
448 #[serde(default)]
450 pub pr_recovery: Option<crate::pr_recovery::Selection>,
451 #[serde(default)]
454 pub fresh_start: bool,
455 #[serde(default)]
457 pub replacement: Option<crate::merge_decision::Replacement>,
458 #[serde(default)]
469 pub interrupt: bool,
470 #[serde(default)]
493 pub urgent: bool,
494 #[serde(default)]
500 pub attachments: Vec<String>,
501 #[serde(default)]
505 pub followup: Option<FollowUp>,
506 #[serde(default)]
512 pub origin_chat: Option<String>,
513 #[serde(default)]
520 pub overrides: Option<RunOverrides>,
521 #[serde(default)]
528 pub review_of: Option<String>,
529 #[serde(default)]
534 pub held_at: Option<Timestamp>,
535 #[serde(default)]
539 pub park_reason: Option<String>,
540 #[serde(default)]
543 pub chat_notice: Option<ChatNotice>,
544 #[serde(default)]
546 pub chat_notices: Vec<String>,
547 #[serde(default)]
550 pub hold_notice_cause: Option<NoticeCause>,
551 #[serde(default)]
556 pub passes: Vec<TaskPass>,
557 #[serde(default)]
559 pub recovery_budget: Option<RecoveryBudget>,
560 #[serde(default)]
563 pub recovery_admission: Option<RecoveryAdmission>,
564 #[serde(default)]
567 pub recovery_generations: Vec<RecoveryBudget>,
568 pub created_at: Timestamp,
570 pub updated_at: Timestamp,
572}
573
574#[derive(Debug, Clone, Serialize, Deserialize)]
576pub struct RecoveryAdmission {
577 pub base: String,
579 pub inputs: String,
581}
582
583#[derive(Debug, Clone, Serialize, Deserialize)]
586pub struct RecoveryBudget {
587 pub instruction: String,
589 pub base: String,
591 pub inputs: String,
593 pub stage: String,
595 pub failures: usize,
597 #[serde(default)]
600 pub progress: Option<String>,
601}
602
603fn recovery_progress(state: &crate::run::RunState) -> Option<String> {
604 use crate::proc::Quiet as _;
605 use std::hash::{Hash, Hasher};
606 let mut patches = Vec::new();
607 for candidate in state.candidates.iter().filter(|c| c.commits > 0) {
608 let output = std::process::Command::new("git")
609 .current_dir(&state.repo)
610 .args(["diff", "--binary", &state.base_commit, &candidate.branch])
611 .quiet()
612 .output()
613 .ok()?;
614 if !output.status.success() {
615 return None;
616 }
617 patches.push(output.stdout);
618 }
619 patches.sort();
620 patches.dedup();
621 let mut hash = std::collections::hash_map::DefaultHasher::new();
622 patches.hash(&mut hash);
623 Some(format!("{:x}", hash.finish()))
624}
625
626pub(crate) fn recovery_inputs(seeds: &[crate::refs::Seed]) -> String {
627 serde_json::to_string(
628 &seeds
629 .iter()
630 .map(|s| (&s.token, &s.sha, s.kind))
631 .collect::<Vec<_>>(),
632 )
633 .unwrap_or_default()
634}
635
636#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
638pub struct NoticeCause {
639 pub stage: String,
641 #[serde(default, skip_serializing_if = "Option::is_none")]
644 pub target: Option<String>,
645 pub run: Option<String>,
647 pub pass: Option<usize>,
649}
650
651pub(crate) fn is_done_notice_key(key: &str) -> bool {
653 key == "done"
654 || key
655 .strip_prefix("done#")
656 .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
657}
658
659#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
661#[serde(rename_all = "snake_case")]
662pub enum PassCost {
663 Spent,
665 Refunded,
667}
668
669#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
671pub struct TaskPass {
672 pub run: String,
674 pub pass_index: usize,
676 pub cost: PassCost,
678 pub attempts_after: usize,
680 pub settled_at: Timestamp,
682 #[serde(default)]
684 pub parked: Option<crate::run::ParkCause>,
685}
686
687#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
689#[serde(rename_all = "lowercase")]
690pub enum ChatNoticeOutcome {
691 Posted,
693 Skipped,
695}
696
697#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
699pub struct ChatNotice {
700 pub key: String,
702 pub at: Timestamp,
704 pub outcome: ChatNoticeOutcome,
706}
707
708#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
710pub struct RunOverrides {
711 #[serde(default)]
714 pub inherit_daemon_options: bool,
715 #[serde(default)]
717 pub merge: Option<String>,
718 #[serde(default)]
720 pub candidates: Option<usize>,
721 #[serde(default)]
723 pub judges: Option<usize>,
724 #[serde(default)]
726 pub reviewers: Option<usize>,
727 #[serde(default)]
729 pub review_rounds: Option<usize>,
730 #[serde(default)]
732 pub seed: Option<u64>,
733 #[serde(default)]
735 pub config: Option<PathBuf>,
736}
737
738impl RunOverrides {
739 pub fn merge_over(&mut self, other: &RunOverrides) {
741 macro_rules! take {
742 ($($f:ident),*) => {$(
743 if other.$f.is_some() {
744 self.$f.clone_from(&other.$f);
745 }
746 )*};
747 }
748 take!(
749 merge,
750 candidates,
751 judges,
752 reviewers,
753 review_rounds,
754 seed,
755 config
756 );
757 }
758
759 pub fn apply(&self, config: &mut crate::config::Config) {
761 if let Some(n) = self.candidates {
762 config.graph.implementers = n;
763 }
764 if let Some(n) = self.judges {
765 config.graph.judges = n;
766 }
767 if let Some(n) = self.reviewers {
768 config.graph.reviewers = n;
769 }
770 if let Some(n) = self.review_rounds {
771 config.graph.review_rounds = n;
772 }
773 if let Some(m) = &self.merge
774 && let Ok(mode) = crate::daemon::merge_mode(m)
775 {
776 config.merge.mode = mode;
777 }
778 if let Some(s) = self.seed {
779 config.blind.seed = Some(s);
780 }
781 }
782}
783
784#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
786pub struct FollowUp {
787 pub run: String,
789 #[serde(default)]
791 pub origin_task: Option<String>,
792 pub pr: String,
794 pub findings: Vec<String>,
796 pub generation: u32,
799}
800
801#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
803pub struct Resumption {
804 pub generation: u64,
806 pub source: String,
808 pub question_id: Option<String>,
810 pub at: Timestamp,
812 pub instruction: String,
814 pub answers_through: usize,
816 pub runs_through: usize,
819 pub cleared_reason: Option<String>,
821 #[serde(default)]
823 pub blocking_evidence: Vec<String>,
824}
825
826#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
829pub struct OperatorResume {
830 pub question_id: String,
832 pub at: Timestamp,
834 #[serde(default)]
837 pub conductor_rehold: Option<String>,
838 #[serde(default)]
841 pub forced: bool,
842 #[serde(default)]
849 pub pinned_run: Option<String>,
850}
851
852#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
855pub struct AnsweredQuestion {
856 pub question: String,
858 pub answer: String,
860}
861
862impl Task {
863 pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
865 let now = Timestamp::now();
866 let origin_chat = match &source {
867 Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
868 _ => None,
869 };
870 Self {
871 schema: SCHEMA,
872 id: new_id(),
873 title,
874 instruction,
875 repo,
876 source,
877 priority: 0,
878 solo: false,
879 status: TaskStatus::Queued,
880 attempts: 0,
881 attempt_limit: None,
882 runs: Vec::new(),
883 last_error: None,
884 hold_reason: None,
885 hold_source: None,
886 diagnostic: None,
887 blocked_by: Vec::new(),
888 block_reason: None,
889 blocked_from: None,
890 blocked_at: None,
891 answers: Vec::new(),
892 triage_applied: Vec::new(),
893 actions_applied: Vec::new(),
894 resume_override: None,
895 resumption: None,
896 resumption_history: Vec::new(),
897 operator_decided_at: None,
898 review_branch: None,
899 review_instruction: None,
900 pr_recovery: None,
901 fresh_start: false,
902 replacement: None,
903 interrupt: false,
904 urgent: false,
905 attachments: Vec::new(),
906 followup: None,
907 origin_chat,
908 overrides: None,
909 review_of: None,
910 held_at: None,
911 park_reason: None,
912 chat_notice: None,
913 chat_notices: Vec::new(),
914 hold_notice_cause: None,
915 passes: Vec::new(),
916 recovery_budget: None,
917 recovery_admission: None,
918 recovery_generations: Vec::new(),
919 created_at: now,
920 updated_at: now,
921 }
922 }
923
924 pub fn chat_talk(&self) -> Option<&str> {
927 if let Some(id) = &self.origin_chat {
928 return Some(id);
929 }
930 match &self.source {
931 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
932 _ => None,
933 }
934 }
935
936 pub fn filed_by_chat(&self) -> Option<&str> {
940 match &self.source {
941 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
942 _ => None,
943 }
944 }
945
946 pub fn chat_notice_key(&self) -> Option<String> {
949 match self.status {
950 TaskStatus::Done => Some("done".to_owned()),
951 TaskStatus::Held => {
952 let cause = self
953 .hold_notice_cause
954 .clone()
955 .unwrap_or_else(|| self.unresolved_hold_cause());
956 Some(format!(
957 "cause:{}",
958 serde_json::to_string(&cause).expect("cause")
959 ))
960 }
961 TaskStatus::Blocked => {
962 let ids: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
963 let evidence = self
966 .hold_notice_cause
967 .as_ref()
968 .filter(|c| c.stage.starts_with("conductor-evidence:"));
969 Some(format!(
970 "dependencies:{}",
971 serde_json::to_string(&(ids, evidence)).expect("ids")
972 ))
973 }
974 _ => None,
975 }
976 }
977
978 fn unresolved_hold_cause(&self) -> NoticeCause {
979 NoticeCause {
980 stage: "hold".into(),
981 target: None,
982 run: self.runs.last().cloned(),
983 pass: None,
984 }
985 }
986
987 pub(crate) fn legacy_chat_notice_key(&self) -> Option<String> {
989 let key = match self.status {
990 TaskStatus::Held if self.hold_notice_cause.is_none() => Some(format!(
991 "held#{}",
992 self.held_at.map(|t| t.to_string()).unwrap_or_default()
993 )),
994 TaskStatus::Blocked => Some(format!(
995 "blocked#{}#{}",
996 self.blocked_by.join(","),
997 self.block_reason.as_deref().unwrap_or_default()
998 )),
999 _ => None,
1000 };
1001 key.filter(|key| self.legacy_notice_matches(key))
1002 }
1003
1004 pub(crate) fn legacy_notice_matches(&self, key: &str) -> bool {
1005 match self.status {
1006 TaskStatus::Held if self.hold_notice_cause.is_none() => key
1007 .strip_prefix("held#")
1008 .is_some_and(|time| time.is_empty() || time.parse::<Timestamp>().is_ok()),
1009 TaskStatus::Blocked
1010 if self
1011 .hold_notice_cause
1012 .as_ref()
1013 .is_none_or(|c| !c.stage.starts_with("conductor-evidence:")) =>
1014 {
1015 key.strip_prefix("blocked#").is_some_and(|tail| {
1016 let Some((ids, _)) = tail.split_once('#') else {
1017 return false;
1018 };
1019 let old: std::collections::BTreeSet<_> =
1020 ids.split(',').filter(|id| !id.is_empty()).collect();
1021 let current: std::collections::BTreeSet<_> =
1022 self.blocked_by.iter().map(String::as_str).collect();
1023 old == current
1024 })
1025 }
1026 _ => false,
1027 }
1028 }
1029
1030 fn import_chat_notice(&mut self) {
1031 if let Some(n) = &self.chat_notice {
1032 let key = if is_done_notice_key(&n.key) {
1033 "done".to_owned()
1034 } else if self.legacy_notice_matches(&n.key) {
1035 self.chat_notice_key().unwrap_or_else(|| n.key.clone())
1036 } else {
1037 n.key.clone()
1038 };
1039 if !self.chat_notices.contains(&key) {
1040 self.chat_notices.push(key);
1041 }
1042 }
1043 }
1044
1045 pub(crate) fn notice_cause(&mut self, stage: &str, run: Option<&str>, pass: Option<usize>) {
1048 self.import_chat_notice();
1049 self.hold_notice_cause = Some(NoticeCause {
1050 stage: stage.to_owned(),
1051 target: None,
1052 run: run.map(str::to_owned),
1053 pass,
1054 });
1055 }
1056
1057 pub(crate) fn startup_notice_cause(&mut self, stage: &str) {
1060 let recovery_only = matches!(
1061 stage,
1062 "startup-source" | "startup-candidates" | "startup-public-head"
1063 );
1064 match self.hold_notice_cause.as_ref() {
1065 Some(cause) if recovery_only && !cause.stage.starts_with("startup-") => return,
1066 None if recovery_only
1067 && self
1068 .chat_notice
1069 .as_ref()
1070 .is_some_and(|n| !is_done_notice_key(&n.key)) =>
1071 {
1072 return;
1073 }
1074 _ => {}
1075 }
1076 self.notice_cause(stage, None, None);
1077 }
1078
1079 pub(crate) fn fail_startup(
1082 &mut self,
1083 stage: &str,
1084 why: impl Into<String>,
1085 max_attempts: usize,
1086 ) {
1087 let unresolved = self.hold_notice_cause.clone();
1088 self.fail(why, max_attempts);
1089 self.hold_notice_cause = unresolved;
1090 self.startup_notice_cause(stage);
1091 }
1092
1093 pub(crate) fn fail_startup_target(
1095 &mut self,
1096 stage: &str,
1097 target: String,
1098 why: impl Into<String>,
1099 max_attempts: usize,
1100 ) {
1101 self.fail_startup(stage, why, max_attempts);
1102 if let Some(cause) = self.hold_notice_cause.as_mut() {
1103 cause.target = Some(target);
1104 }
1105 }
1106
1107 pub fn chat_notice_due(&self) -> Option<String> {
1109 self.filed_by_chat()?;
1110 if self.schema < 13 {
1111 return None;
1112 }
1113 let key = self.chat_notice_key()?;
1114 if self.chat_notices.contains(&key) {
1115 return None;
1116 }
1117 match &self.chat_notice {
1118 Some(n)
1119 if n.key == key
1120 || (key == "done" && is_done_notice_key(&n.key))
1121 || self.legacy_notice_matches(&n.key) =>
1122 {
1123 None
1124 }
1125 _ => Some(key),
1126 }
1127 }
1128
1129 pub fn short(&self) -> &str {
1131 short(&self.id)
1132 }
1133
1134 pub fn mark_triage_applied(&mut self, question_id: &str) {
1136 if !self.triage_applied(question_id) {
1137 self.triage_applied.push(question_id.to_owned());
1138 }
1139 }
1140
1141 pub fn action_applied(&self, question_id: &str) -> bool {
1144 self.actions_applied.iter().any(|id| id == question_id)
1145 }
1146
1147 pub fn mark_action_applied(&mut self, question_id: &str) {
1149 if !self.action_applied(question_id) {
1150 self.actions_applied.push(question_id.to_owned());
1151 }
1152 }
1153
1154 pub fn triage_applied(&self, question_id: &str) -> bool {
1156 self.triage_applied.iter().any(|id| id == question_id)
1157 }
1158
1159 pub fn start(&mut self, run: String) {
1172 self.stamp_schema();
1173 self.status = TaskStatus::Running;
1174 self.attempts += 1;
1175 self.runs.retain(|id| id != &run);
1176 self.runs.push(run);
1177 self.last_error = None;
1178 self.park_reason = None;
1179 self.fresh_start = false;
1180 self.interrupt = false;
1181 self.resume_override = None;
1183 }
1184
1185 pub fn link_run(&mut self, run: &str) -> bool {
1190 if self.runs.iter().any(|r| r == run) {
1191 return false;
1192 }
1193 self.runs.push(run.to_owned());
1194 true
1195 }
1196
1197 pub fn succeed(&mut self) {
1207 self.stamp_schema();
1208 self.status = TaskStatus::Done;
1209 self.resume_override = None;
1210 self.last_error = None;
1211 self.park_reason = None;
1212 self.hold_reason = None;
1213 self.hold_source = None;
1214 self.diagnostic = None;
1215 self.blocked_by.clear();
1216 self.block_reason = None;
1217 self.blocked_from = None;
1218 }
1219
1220 pub fn already_landed(&mut self, note: impl Into<String>) {
1228 self.succeed();
1229 self.attempts = self.attempts.saturating_sub(1);
1230 self.last_error = Some(note.into());
1231 }
1232
1233 pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
1263 if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
1264 return &[];
1265 }
1266 &self.runs[..self.runs.len() - 1]
1267 }
1268
1269 pub fn successor_of(&self, run: &str) -> Option<&String> {
1277 let pos = self.runs.iter().rposition(|r| r == run)?;
1278 self.runs.get(pos + 1)
1279 }
1280
1281 pub fn earlier_attempts(&self) -> &[String] {
1288 &self.runs
1289 }
1290
1291 fn stamp_schema(&mut self) {
1298 self.schema = self.schema.max(SCHEMA);
1299 }
1300
1301 fn note_held(&mut self) {
1302 self.stamp_schema();
1303 let continuing = self.status == TaskStatus::Held
1304 || (self.status == TaskStatus::Blocked
1305 && (self.blocked_from == Some(TaskStatus::Held)
1306 || (self.blocked_from.is_none()
1307 && (self.hold_reason.is_some() || self.hold_source.is_some()))));
1308 if !continuing || self.held_at.is_none() {
1309 self.held_at = Some(Timestamp::now());
1310 }
1311 }
1312
1313 pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
1329 let why = why.into();
1330 if self.status == TaskStatus::Running {
1331 let run = self.runs.last().cloned();
1332 self.notice_cause("run-failed", run.as_deref(), Some(self.attempts));
1333 }
1334 self.diagnostic = None;
1335 self.park_reason = None;
1336 self.status = if self.attempts >= self.effective_attempt_limit(max_attempts) {
1337 self.note_held();
1338 self.hold_source = Some(HoldSource::Machine);
1339 self.hold_reason = Some(why.clone());
1340 TaskStatus::Held
1341 } else {
1342 TaskStatus::Failed
1343 };
1344 self.last_error = Some(why);
1345 }
1346
1347 pub fn stall(&mut self, why: impl Into<String>) {
1357 self.last_error = Some(why.into());
1358 self.diagnostic = None;
1359 self.attempts = self.attempts.saturating_sub(1);
1360 self.status = TaskStatus::Failed;
1361 }
1362
1363 pub fn park(&mut self, why: impl Into<String>) {
1369 self.diagnostic = None;
1370 self.attempts = self.attempts.saturating_sub(1);
1371 self.status = TaskStatus::Parked;
1372 self.park_reason = Some(why.into());
1373 self.schema = self.schema.max(SCHEMA);
1376 }
1377
1378 pub fn record_pass(
1381 &mut self,
1382 run: &str,
1383 pass_index: usize,
1384 attempts_before: usize,
1385 parked: Option<crate::run::ParkCause>,
1386 ) {
1387 if self
1388 .passes
1389 .iter()
1390 .any(|p| p.run == run && p.pass_index == pass_index)
1391 {
1392 return;
1393 }
1394 let cost = if self.attempts < attempts_before {
1395 PassCost::Refunded
1396 } else {
1397 PassCost::Spent
1398 };
1399 self.passes.push(TaskPass {
1400 run: run.to_owned(),
1401 pass_index,
1402 cost,
1403 attempts_after: self.attempts,
1404 settled_at: Timestamp::now(),
1405 parked,
1406 });
1407 self.schema = self.schema.max(SCHEMA);
1408 }
1409
1410 pub fn operator_held(&self) -> bool {
1416 self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
1417 }
1418
1419 pub fn hold_manual(&mut self, reason: Option<String>) {
1429 self.operator_decided_at = Some(Timestamp::now());
1431 self.hold_unrequested_retry(reason);
1432 }
1433
1434 pub(crate) fn hold_unrequested_retry(&mut self, reason: Option<String>) {
1437 self.park_reason = None;
1438 self.note_held();
1439 self.status = TaskStatus::Held;
1440 if reason.is_some() {
1441 self.hold_reason = reason;
1442 }
1443 self.hold_source = Some(HoldSource::Manual);
1444 self.blocked_by.clear();
1445 self.block_reason = None;
1446 self.blocked_from = None;
1447 }
1448
1449 pub fn hold_machine(&mut self, reason: Option<String>) {
1454 self.park_reason = None;
1455 self.note_held();
1456 self.status = TaskStatus::Held;
1457 if reason.is_some() {
1458 self.hold_reason = reason;
1459 }
1460 self.hold_source = Some(HoldSource::Machine);
1461 self.blocked_by.clear();
1462 self.block_reason = None;
1463 self.blocked_from = None;
1464 }
1465
1466 pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
1476 self.stamp_schema();
1477 let before: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
1478 let after: std::collections::BTreeSet<_> = blocked_by.iter().collect();
1479 if self.status != TaskStatus::Blocked || before != after {
1480 self.blocked_at = Some(Timestamp::now());
1481 }
1482 if self.status != TaskStatus::Blocked {
1483 self.blocked_from = Some(self.status);
1484 }
1485 self.status = TaskStatus::Blocked;
1486 self.blocked_by = blocked_by;
1487 self.block_reason = reason;
1488 }
1489
1490 pub fn unblock(&mut self, resolved_id: &str) {
1513 if self.status != TaskStatus::Blocked {
1514 return;
1515 }
1516 self.blocked_by.retain(|id| id != resolved_id);
1517 self.restore_if_unblocked();
1518 }
1519
1520 pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
1531 if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
1532 return false;
1533 }
1534 self.blocked_by.retain(|id| id != deleted_id);
1535 self.restore_if_unblocked();
1536 true
1537 }
1538
1539 fn restore_if_unblocked(&mut self) {
1543 if self.blocked_by.is_empty() {
1544 self.status = match self.blocked_from {
1545 Some(TaskStatus::Running) => TaskStatus::Queued,
1546 Some(other) => other,
1547 None if self.hold_reason.is_some() || self.hold_source.is_some() => {
1548 TaskStatus::Held
1549 }
1550 None => TaskStatus::Queued,
1551 };
1552 self.block_reason = None;
1553 self.blocked_from = None;
1554 }
1555 }
1556
1557 pub fn record_answer(&mut self, question: String, answer: String) {
1562 self.answers.push(AnsweredQuestion { question, answer });
1563 }
1564
1565 pub fn request_review(&mut self, branch: String) {
1569 self.release();
1570 self.review_branch = Some(branch);
1571 self.review_instruction = Some(self.instruction.clone());
1572 }
1573
1574 pub fn requeue(&mut self) {
1577 self.release();
1578 self.review_branch = None;
1580 self.review_instruction = None;
1581 self.fresh_start = true;
1582 }
1583
1584 pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
1593 if branch.is_some() {
1594 self.review_branch = branch;
1595 self.review_instruction = Some(self.instruction.clone());
1596 }
1597 self.hold_machine(Some(reason));
1598 }
1599
1600 pub fn set_urgent(&mut self, urgent: bool) -> Result<()> {
1602 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1603 bail!(
1604 "task {} is {}; urgent can only change while queued or held; \
1605 wait for the active attempt to finish and requeue or hold the task first",
1606 self.short(),
1607 self.status.as_str()
1608 );
1609 }
1610 self.urgent = urgent;
1611 Ok(())
1612 }
1613
1614 pub fn set_priority(&mut self, priority: i32) -> Result<()> {
1623 if self.status == TaskStatus::Running {
1624 bail!(
1625 "task {} is running; its priority cannot be changed until \
1626 this attempt finishes",
1627 self.short()
1628 );
1629 }
1630 self.priority = priority;
1631 Ok(())
1632 }
1633
1634 pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
1649 if interrupt && !self.status.runnable() {
1650 bail!(
1651 "task {} is {}; only a queued or failed task can be marked \
1652 to interrupt",
1653 self.short(),
1654 self.status.as_str()
1655 );
1656 }
1657 self.interrupt = interrupt;
1658 Ok(())
1659 }
1660
1661 pub fn effective_attempt_limit(&self, daemon_limit: usize) -> usize {
1663 self.attempt_limit.map_or(daemon_limit, NonZeroUsize::get)
1664 }
1665
1666 pub fn change_attempt_limit(&mut self, limit: Option<NonZeroUsize>) -> Result<()> {
1669 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1670 bail!("only queued or held tasks can change the attempt limit");
1671 }
1672 self.stamp_schema();
1673 self.attempt_limit = limit;
1674 Ok(())
1675 }
1676
1677 pub fn implementers(&self) -> Option<usize> {
1679 if self.solo {
1680 Some(1)
1681 } else {
1682 self.overrides.as_ref().and_then(|o| o.candidates)
1683 }
1684 }
1685
1686 pub fn change_implementers(&mut self, count: u8) -> Result<()> {
1689 self.check_implementers_editable()?;
1690 let explicit = self.overrides.as_ref().and_then(|o| o.config.as_deref());
1691 let (mut config, _) = crate::config::Config::discover(&self.repo, explicit)?;
1692 if let Some(o) = &self.overrides {
1693 o.apply(&mut config);
1694 }
1695 self.set_implementers(count, &config)
1696 }
1697
1698 fn check_implementers_editable(&self) -> Result<()> {
1699 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1700 bail!(
1701 "task {} is {}; only queued or held tasks can change implementers",
1702 self.short(),
1703 self.status.as_str()
1704 );
1705 }
1706 Ok(())
1707 }
1708
1709 pub fn set_implementers(&mut self, count: u8, config: &crate::config::Config) -> Result<()> {
1711 self.check_implementers_editable()?;
1712 crate::talk::check_implementers(count, config).map_err(anyhow::Error::msg)?;
1713 self.stamp_schema();
1714 self.solo = count == 1;
1715 let mut overrides = self.overrides.clone().unwrap_or_else(|| RunOverrides {
1716 inherit_daemon_options: true,
1717 ..Default::default()
1718 });
1719 overrides.candidates = (count > 1).then_some(usize::from(count));
1720 let mut settings = overrides.clone();
1721 settings.inherit_daemon_options = false;
1722 self.overrides = if overrides.inherit_daemon_options && settings == RunOverrides::default()
1724 {
1725 None
1726 } else {
1727 Some(overrides)
1728 };
1729 Ok(())
1730 }
1731
1732 pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
1744 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1745 bail!(
1746 "task {} is {}; only a queued or held task's instruction can \
1747 be edited",
1748 self.short(),
1749 self.status.as_str()
1750 );
1751 }
1752 if self.review_branch.is_some() && self.review_instruction.is_none() {
1755 self.review_instruction = Some(self.instruction.clone());
1756 }
1757 if self.instruction != instruction {
1758 self.resume_override = None;
1761 }
1762 self.title = title;
1763 self.instruction = instruction;
1764 Ok(())
1765 }
1766
1767 pub fn handed_off(&mut self, why: impl Into<String>) {
1787 let why = why.into();
1788 self.diagnostic = None;
1789 self.note_held();
1790 self.status = TaskStatus::Held;
1791 self.hold_source = Some(HoldSource::Machine);
1792 self.hold_reason = Some(why.clone());
1793 self.last_error = Some(why);
1794 }
1795
1796 pub fn release_operator(&mut self, source: &str, question_id: Option<&str>) {
1798 self.recovery_budget = None;
1799 self.recovery_generations.clear();
1800 self.recovery_admission = None;
1801 if self
1802 .resumption
1803 .as_ref()
1804 .is_some_and(|r| r.instruction != self.instruction)
1805 {
1806 self.resume_override = None;
1809 }
1810 let approval = Resumption {
1811 generation: self.resumption.as_ref().map_or(1, |r| r.generation + 1),
1812 source: source.to_owned(),
1813 question_id: question_id.map(str::to_owned),
1814 at: Timestamp::now(),
1815 instruction: self.instruction.clone(),
1816 answers_through: self.answers.len(),
1817 runs_through: self.runs.len(),
1818 blocking_evidence: Vec::new(),
1819 cleared_reason: self
1820 .hold_reason
1821 .clone()
1822 .or_else(|| self.block_reason.clone()),
1823 };
1824 self.release();
1825 self.operator_decided_at = Some(approval.at);
1826 if let Some(previous) = self.resumption.replace(approval) {
1827 self.resumption_history.push(previous);
1828 }
1829 }
1830
1831 pub(crate) fn recovery_failed(&mut self, state: &crate::run::RunState) {
1833 let (base, inputs) = self.recovery_admission.as_ref().map_or_else(
1834 || (state.base_commit.clone(), recovery_inputs(&state.seeds)),
1835 |a| (a.base.clone(), a.inputs.clone()),
1836 );
1837 let progress = recovery_progress(state);
1838 let stage = if state.prep_failure.is_some() {
1839 "preparation"
1840 } else {
1841 "execution"
1842 };
1843 let matches = |b: &RecoveryBudget| {
1844 b.instruction == self.instruction
1845 && b.base == base
1846 && b.inputs == inputs
1847 && (progress.is_none() || b.progress == progress)
1848 };
1849 if !self.recovery_budget.as_ref().is_some_and(matches) {
1850 let previous = self
1851 .recovery_generations
1852 .iter()
1853 .position(matches)
1854 .map(|index| self.recovery_generations.remove(index));
1855 if let Some(old) = self.recovery_budget.take() {
1856 self.recovery_generations.push(old);
1857 }
1858 self.recovery_budget = previous;
1859 }
1860 if let Some(budget) = &mut self.recovery_budget {
1861 budget.failures = budget.failures.saturating_add(1);
1862 budget.stage = stage.to_owned();
1863 } else {
1864 self.recovery_budget = Some(RecoveryBudget {
1865 instruction: self.instruction.clone(),
1866 base,
1867 inputs,
1868 stage: stage.to_owned(),
1869 failures: 1,
1870 progress,
1871 });
1872 }
1873 }
1874
1875 pub(crate) fn recovery_exhausted(
1876 &self,
1877 base: &str,
1878 inputs: &str,
1879 limit: usize,
1880 ) -> Option<String> {
1881 let same_input = |b: &&RecoveryBudget| {
1882 b.instruction == self.instruction && b.base == base && b.inputs == inputs
1883 };
1884 let budget = self
1885 .recovery_budget
1886 .as_ref()
1887 .filter(same_input)
1888 .or_else(|| self.recovery_generations.iter().rev().find(same_input))?;
1889 (budget.instruction == self.instruction && budget.base == base && budget.inputs == inputs
1890 && budget.failures >= limit.max(1)).then(|| format!(
1891 "[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.",
1892 budget.failures, budget.stage))
1893 }
1894
1895 pub fn release(&mut self) {
1899 self.import_chat_notice();
1902 if self.hold_notice_cause.is_none()
1903 && matches!(self.status, TaskStatus::Held | TaskStatus::Blocked)
1904 {
1905 self.hold_notice_cause = Some(self.unresolved_hold_cause());
1906 }
1907 let refused_handover = self.status == TaskStatus::Held
1908 && self.hold_source == Some(HoldSource::Machine)
1909 && self.review_branch.is_some()
1910 && self
1911 .review_instruction
1912 .as_ref()
1913 .is_none_or(|i| *i == self.instruction);
1914 self.status = TaskStatus::Queued;
1915 self.held_at = None;
1916 self.attempts = 0;
1917 self.last_error = None;
1918 self.park_reason = None;
1919 self.hold_reason = None;
1922 self.hold_source = None;
1923 self.diagnostic = None;
1924 self.blocked_by.clear();
1929 self.block_reason = None;
1930 self.blocked_from = None;
1931 if !refused_handover {
1935 self.review_branch = None;
1936 self.review_instruction = None;
1937 }
1938 self.fresh_start = false;
1939 }
1940}
1941
1942fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
1946 use std::io::ErrorKind;
1947 let (stem, ext) = match name.rfind('.') {
1948 Some(i) if i > 0 => (&name[..i], &name[i..]),
1949 _ => (name, ""),
1950 };
1951 for n in 1u32.. {
1952 let candidate = if n == 1 {
1953 name.to_owned()
1954 } else {
1955 let suffix = format!("-{n}");
1956 let room = 64usize.saturating_sub(suffix.len() + ext.len());
1957 let stem: String = stem.chars().take(room).collect();
1958 format!("{stem}{suffix}{ext}")
1959 };
1960 if !crate::ask::valid_asset_name(&candidate) {
1961 bail!("no valid attachment name is left for `{name}`");
1962 }
1963 let path = dir.join(&candidate);
1964 match std::fs::OpenOptions::new()
1965 .write(true)
1966 .create_new(true)
1967 .open(&path)
1968 {
1969 Ok(mut out) => {
1970 let copied = std::fs::File::open(src)
1971 .and_then(|mut input| std::io::copy(&mut input, &mut out));
1972 if let Err(e) = copied {
1973 drop(out);
1974 let _ = std::fs::remove_file(&path);
1975 return Err(e).with_context(|| format!("copy {}", src.display()));
1976 }
1977 return Ok((candidate, path));
1978 }
1979 Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
1980 Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
1981 }
1982 }
1983 unreachable!("the counter never runs out")
1984}
1985
1986const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
1989
1990pub(crate) struct TaskLock {
1993 path: PathBuf,
1994 token: String,
1995}
1996
1997fn owner_token() -> String {
2000 static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
2001 let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2002 let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
2003 format!("{}-{:016x}", std::process::id(), r.next_u64())
2004}
2005
2006fn break_marker(path: &Path, token: &str) -> PathBuf {
2009 let mut h: u64 = 0xcbf29ce484222325;
2010 for b in token.bytes() {
2011 h ^= u64::from(b);
2012 h = h.wrapping_mul(0x100000001b3);
2013 }
2014 let mut name = path.as_os_str().to_owned();
2015 name.push(format!(".break-{h:016x}"));
2016 PathBuf::from(name)
2017}
2018
2019fn older_than_stale(path: &Path) -> bool {
2020 std::fs::metadata(path)
2021 .and_then(|m| m.modified())
2022 .ok()
2023 .and_then(|t| t.elapsed().ok())
2024 .is_some_and(|age| age > TASK_LOCK_STALE)
2025}
2026
2027fn writer_may_be_gone(token: &str) -> bool {
2028 let Some(pid) = token
2029 .split_once('-')
2030 .and_then(|(pid, _)| pid.parse::<u32>().ok())
2031 else {
2032 return true;
2035 };
2036 if pid == std::process::id() {
2037 return false;
2038 }
2039 crate::proc::pid_status(pid) == Some(false)
2041}
2042
2043const MAX_MARKER_DEPTH: u8 = 3;
2047
2048fn take_marker(
2054 path: &Path,
2055 token: &str,
2056 depth: u8,
2057 retry: crate::lock_file::Retry,
2058) -> Option<(PathBuf, String)> {
2059 use std::io::Write;
2060 let marker = break_marker(path, token);
2061 for _ in 0..2 {
2062 match retry.create(&marker) {
2063 Ok(mut f) => {
2064 let mine = owner_token();
2065 if f.write_all(mine.as_bytes()).is_err() {
2066 drop(f);
2067 let _ = std::fs::remove_file(&marker);
2068 return None;
2069 }
2070 return Some((marker, mine));
2071 }
2072 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2073 let judged = std::fs::read_to_string(&marker).ok();
2074 match judged.filter(|_| older_than_stale(&marker)) {
2075 Some(judged) if depth < MAX_MARKER_DEPTH => {
2076 remove_lock_if(&marker, &judged, true, depth + 1, retry)?;
2077 }
2078 _ => return None,
2079 }
2080 }
2081 Err(_) => return None,
2082 }
2083 }
2084 None
2085}
2086
2087fn remove_lock_if(
2092 path: &Path,
2093 token: &str,
2094 require_stale: bool,
2095 depth: u8,
2096 retry: crate::lock_file::Retry,
2097) -> Option<bool> {
2098 let (marker, mine) = take_marker(path, token, depth, retry)?;
2099 let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
2100 && (!require_stale || older_than_stale(path) && writer_may_be_gone(token));
2101 if still {
2102 let _ = std::fs::remove_file(path);
2103 }
2104 if depth >= MAX_MARKER_DEPTH {
2106 if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
2107 let _ = std::fs::remove_file(&marker);
2108 }
2109 } else {
2110 let _ = remove_lock_if(&marker, &mine, false, depth + 1, retry);
2111 }
2112 Some(still)
2113}
2114
2115#[cfg(test)]
2118fn break_stale(path: &Path, judged: &str) -> bool {
2119 break_stale_with(path, judged, crate::lock_file::Retry::new(None))
2120}
2121
2122fn break_stale_with(path: &Path, judged: &str, retry: crate::lock_file::Retry) -> bool {
2123 remove_lock_if(path, judged, true, 0, retry).unwrap_or(false)
2124}
2125
2126impl Drop for TaskLock {
2127 fn drop(&mut self) {
2128 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
2129 let retry = crate::lock_file::Retry::new(Some(deadline));
2130 loop {
2131 match remove_lock_if(&self.path, &self.token, false, 0, retry) {
2132 Some(_) => return,
2133 None if std::time::Instant::now() > deadline => return,
2134 None => std::thread::sleep(std::time::Duration::from_millis(5)),
2135 }
2136 }
2137 }
2138}
2139
2140#[derive(Debug, Clone)]
2142pub struct Queue {
2143 root: PathBuf,
2144}
2145
2146impl Queue {
2147 pub fn open() -> Self {
2149 Self::at(crate::run::home().join("queue"))
2150 }
2151
2152 pub fn at(root: PathBuf) -> Self {
2155 Self { root }
2156 }
2157
2158 pub fn root(&self) -> &Path {
2160 &self.root
2161 }
2162
2163 pub fn path_of(&self, id: &str) -> PathBuf {
2165 self.root.join(format!("{id}.json"))
2166 }
2167
2168 pub fn attachments_dir(&self, id: &str) -> PathBuf {
2171 self.root.join(format!("{id}.attachments"))
2172 }
2173
2174 pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2185 let mut wanted = Vec::new();
2186 for src in sources {
2187 let name = src
2188 .file_name()
2189 .and_then(|n| n.to_str())
2190 .with_context(|| format!("`{}` has no usable file name", src.display()))?;
2191 if !crate::ask::valid_asset_name(name) {
2192 bail!(
2193 "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
2194 with no `..`; rename the file and try again"
2195 );
2196 }
2197 if !src.is_file() {
2198 bail!("attachment `{}` is not a file", src.display());
2199 }
2200 wanted.push((src, name));
2201 }
2202 let dir = self.attachments_dir(&task.id);
2203 let existed = dir.is_dir();
2204 std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
2205 let mut created: Vec<PathBuf> = Vec::new();
2206 let mut names = Vec::new();
2207 let mut copy_all = || -> Result<()> {
2208 for (src, name) in &wanted {
2209 let (stored, path) = copy_new(&dir, src, name)?;
2210 created.push(path);
2211 names.push(stored);
2212 }
2213 Ok(())
2214 };
2215 if let Err(e) = copy_all() {
2216 for path in &created {
2217 let _ = std::fs::remove_file(path);
2218 }
2219 if !existed {
2220 let _ = std::fs::remove_dir(&dir);
2221 }
2222 return Err(e);
2223 }
2224 task.attachments.extend(names.iter().cloned());
2225 Ok(names)
2226 }
2227
2228 pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2239 let dir = self.attachments_dir(&task.id);
2240 let existed = dir.is_dir();
2241 let before = task.attachments.len();
2242 let names = self.attach(task, sources)?;
2243 if let Err(e) = self.put(task) {
2244 for name in &names {
2245 let _ = std::fs::remove_file(dir.join(name));
2246 }
2247 if !existed {
2248 let _ = std::fs::remove_dir(&dir);
2249 }
2250 task.attachments.truncate(before);
2251 return Err(e);
2252 }
2253 Ok(names)
2254 }
2255
2256 pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
2260 let dir = self.attachments_dir(&task.id);
2261 task.attachments
2262 .iter()
2263 .map(|n| {
2264 let p = dir.join(n);
2265 std::path::absolute(&p).unwrap_or(p)
2266 })
2267 .collect()
2268 }
2269
2270 pub fn put(&self, task: &mut Task) -> Result<()> {
2279 let _lock = self.lock_task(&task.id)?;
2280 self.put_unlocked(task)
2281 }
2282
2283 pub fn create_new(&self, task: &mut Task) -> Result<bool> {
2288 let _lock = self.lock_task(&task.id)?;
2289 if self.path_of(&task.id).exists() {
2290 return Ok(false);
2291 }
2292 self.put_unlocked(task)?;
2293 Ok(true)
2294 }
2295
2296 fn put_unlocked(&self, task: &mut Task) -> Result<()> {
2298 if let Ok(mut stored) = read_path(&self.path_of(&task.id)) {
2299 stored.import_chat_notice();
2300 for key in stored.chat_notices {
2301 if !task.chat_notices.contains(&key) {
2302 task.chat_notices.push(key);
2303 }
2304 }
2305 for run in stored.runs {
2306 if !task.runs.contains(&run) {
2307 task.runs.push(run);
2308 }
2309 }
2310 for p in stored.passes {
2311 if !task
2312 .passes
2313 .iter()
2314 .any(|q| q.run == p.run && q.pass_index == p.pass_index)
2315 {
2316 task.passes.push(p);
2317 }
2318 }
2319 if let Some(n) = stored.chat_notice
2322 && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
2323 {
2324 task.chat_notice = Some(n);
2325 }
2326 }
2327 task.import_chat_notice();
2328 task.updated_at = Timestamp::now();
2329 std::fs::create_dir_all(&self.root)
2330 .with_context(|| format!("create {}", self.root.display()))?;
2331 let body = serde_json::to_string_pretty(task).context("serialize task")?;
2332 let path = self.path_of(&task.id);
2333 let tmp = path.with_extension("json.tmp");
2334 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
2335 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
2336 if let (Some(notice), Some(home)) = (
2341 crate::notices::task_held(task),
2342 self.root.parent().filter(|p| !p.as_os_str().is_empty()),
2343 ) {
2344 crate::notices::raise_in(home, notice);
2345 }
2346 Ok(())
2347 }
2348
2349 pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
2354 let id = self.resolve_id(id)?;
2355 let _lock = self.lock_task(&id)?;
2356 let mut task = self.get(&id)?;
2357 if !f(&mut task) {
2358 return Ok(false);
2359 }
2360 self.put_unlocked(&mut task)?;
2361 Ok(true)
2362 }
2363
2364 pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
2369 let id = self.resolve_id(id)?;
2370 let _lock = self.lock_task(&id)?;
2373 let mut task = self.get(&id)?;
2374 if task.link_run(run) {
2375 self.put_unlocked(&mut task)?;
2376 }
2377 Ok(task)
2378 }
2379
2380 pub(crate) fn lock_task(&self, id: &str) -> Result<TaskLock> {
2388 std::fs::create_dir_all(&self.root)
2389 .with_context(|| format!("create {}", self.root.display()))?;
2390 let path = self.root.join(format!("{id}.write-lock"));
2391 let started = std::time::Instant::now();
2392 let retry = crate::lock_file::Retry::task_write(started + TASK_LOCK_STALE);
2393 let marker_retry = crate::lock_file::Retry::new(Some(started + TASK_LOCK_STALE));
2394 loop {
2395 match retry.create(&path) {
2396 Ok(mut file) => {
2397 use std::io::Write;
2398 let token = owner_token();
2399 if let Err(e) = file.write_all(token.as_bytes()) {
2400 drop(file);
2401 let _ = std::fs::remove_file(&path);
2402 return Err(e).with_context(|| format!("lock {}", path.display()));
2403 }
2404 return Ok(TaskLock { path, token });
2405 }
2406 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2407 let judged = std::fs::read_to_string(&path).ok();
2410 let broken = judged
2411 .filter(|_| older_than_stale(&path))
2412 .is_some_and(|judged| break_stale_with(&path, &judged, marker_retry));
2413 if !broken {
2417 if started.elapsed() > TASK_LOCK_STALE {
2418 bail!("could not lock task {id}");
2419 }
2420 std::thread::sleep(std::time::Duration::from_millis(15));
2421 }
2422 }
2423 Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
2424 }
2425 }
2426 }
2427
2428 pub fn get(&self, id: &str) -> Result<Task> {
2430 let resolved = self.resolve_id(id)?;
2431 read_path(&self.path_of(&resolved))
2432 }
2433
2434 pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
2466 let resolved = self.resolve_id(id)?;
2467 if in_flight {
2468 bail!("task {resolved} is being run by a live daemon right now");
2469 }
2470 self.write_tombstone(&resolved)?;
2471 if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
2472 {
2473 if self.path_of(&resolved).exists() {
2477 let _ = std::fs::remove_file(self.tombstone_path(&resolved));
2478 }
2479 return Err(e);
2480 }
2481 let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
2482 Ok(Removal {
2483 id: resolved,
2484 released,
2485 still_blocked,
2486 })
2487 }
2488
2489 fn tombstone_path(&self, id: &str) -> PathBuf {
2492 self.root.join(format!("{id}.removed"))
2493 }
2494
2495 fn write_tombstone(&self, id: &str) -> Result<()> {
2496 let path = self.tombstone_path(id);
2497 let tmp = self.root.join(format!("{id}.removed.tmp"));
2498 std::fs::write(&tmp, b"")
2499 .and_then(|()| std::fs::rename(&tmp, &path))
2500 .with_context(|| format!("write {}", path.display()))
2501 }
2502
2503 fn was_deleted(&self, id: &str) -> bool {
2507 self.tombstone_path(id).is_file() && !self.path_of(id).exists()
2508 }
2509
2510 pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
2514 let deleted = deleted_blockers(self, questions, &task.blocked_by);
2515 deleted
2516 .into_iter()
2517 .filter(|id| task.dependency_deleted(id))
2518 .collect()
2519 }
2520
2521 pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
2525 let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
2526 return;
2527 };
2528 let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
2529 let waiting = dependent.status == TaskStatus::Blocked;
2530 let message = match (ja, waiting) {
2531 (true, false) => format!(
2532 "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
2533 dependent.short(),
2534 short(deleted)
2535 ),
2536 (true, true) => format!(
2537 "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
2538 dependent.short(),
2539 short(deleted)
2540 ),
2541 (false, false) => format!(
2542 "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
2543 dependent.short(),
2544 short(deleted)
2545 ),
2546 (false, true) => format!(
2547 "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
2548 dependent.short(),
2549 short(deleted)
2550 ),
2551 };
2552 crate::notices::raise_in(
2553 home,
2554 crate::notices::Notice::info(
2555 &format!("unblocked:{}:{}", dependent.id, deleted),
2556 message,
2557 )
2558 .link(crate::notices::Link::Task {
2559 id: dependent.id.clone(),
2560 }),
2561 );
2562 }
2563
2564 fn release_dependents_of(
2569 &self,
2570 dependency: &str,
2571 questions: &Questions,
2572 ) -> (Vec<String>, Vec<String>) {
2573 let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
2574 for listed in self.list() {
2575 if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
2576 matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
2577 }) {
2578 continue;
2579 }
2580 let Ok(_claim) = self.claim(&listed.id) else {
2581 continue;
2582 };
2583 let Ok(mut task) = self.get(&listed.id) else {
2584 continue;
2585 };
2586 let references: Vec<String> = task.blocked_by.iter().filter(|id| {
2587 matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
2588 }).cloned().collect();
2589 if references.is_empty() {
2590 continue;
2591 }
2592 for reference in references {
2593 task.dependency_deleted(&reference);
2594 }
2595 if self.put(&mut task).is_err() {
2596 continue;
2597 }
2598 self.note_dependency_deleted(&task, dependency);
2599 if task.status == TaskStatus::Blocked {
2600 still_blocked.push(task.id.clone());
2601 } else {
2602 released.push(task.id.clone());
2603 }
2604 }
2605 (released, still_blocked)
2606 }
2607
2608 fn remove_record_with_attachments(
2612 &self,
2613 resolved: &str,
2614 remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
2615 ) -> Result<()> {
2616 self.sweep_removed_attachments();
2623 let attachments = self.attachments_dir(resolved);
2624 let aside = self.root.join(format!("{resolved}.attachments.removing"));
2625 let moved = match std::fs::rename(&attachments, &aside) {
2626 Ok(()) => true,
2627 Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
2628 Err(e) => {
2629 return Err(e).with_context(|| format!("remove {}", attachments.display()));
2630 }
2631 };
2632 let path = self.path_of(resolved);
2633 if let Err(e) = remove_record(&path) {
2634 if moved {
2635 let _ = std::fs::rename(&aside, &attachments);
2636 }
2637 return Err(e).with_context(|| format!("remove {}", path.display()));
2638 }
2639 if moved {
2640 if let Err(e) = std::fs::remove_dir_all(&aside) {
2641 tracing::warn!("leftover attachments {}: {e}", aside.display());
2642 }
2643 }
2644 let lock = self.lock_path(resolved);
2645 if let Err(e) = std::fs::remove_file(&lock) {
2646 if e.kind() != std::io::ErrorKind::NotFound {
2647 return Err(e).with_context(|| format!("remove {}", lock.display()));
2648 }
2649 }
2650 Ok(())
2651 }
2652
2653 fn sweep_removed_attachments(&self) {
2659 let Ok(entries) = std::fs::read_dir(&self.root) else {
2660 return;
2661 };
2662 for entry in entries.flatten() {
2663 let name = entry.file_name();
2664 let name = name.to_string_lossy();
2665 let Some(id) = name.strip_suffix(".attachments.removing") else {
2666 continue;
2667 };
2668 if !self.path_of(id).exists() {
2671 if let Err(e) = std::fs::remove_dir_all(entry.path()) {
2672 tracing::warn!("leftover attachments {}: {e}", entry.path().display());
2673 }
2674 }
2675 }
2676 self.sweep_tombstones();
2677 }
2678
2679 fn sweep_tombstones(&self) {
2688 let Ok(entries) = std::fs::read_dir(&self.root) else {
2689 return;
2690 };
2691 let tasks = self.list();
2692 for entry in entries.flatten() {
2693 let name = entry.file_name();
2694 let name = name.to_string_lossy();
2695 let Some(id) = name.strip_suffix(".removed") else {
2696 continue;
2697 };
2698 if self.path_of(id).exists()
2699 || tasks.iter().any(|t| {
2700 t.blocked_by
2701 .iter()
2702 .any(|b| id.starts_with(b) || id.ends_with(b))
2703 })
2704 {
2705 continue;
2706 }
2707 let old = entry
2708 .metadata()
2709 .and_then(|m| m.modified())
2710 .ok()
2711 .and_then(|m| m.elapsed().ok())
2712 .is_some_and(|age| age >= TOMBSTONE_GRACE);
2713 if old {
2714 let _ = std::fs::remove_file(entry.path());
2715 }
2716 }
2717 }
2718
2719 fn lock_path(&self, id: &str) -> PathBuf {
2722 self.root.join(format!("{id}.lock"))
2723 }
2724
2725 pub fn list(&self) -> Vec<Task> {
2738 let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
2739 .into_iter()
2740 .flatten()
2741 .flatten()
2742 .map(|e| e.path())
2743 .filter(|p| p.extension().is_some_and(|x| x == "json"))
2744 .filter_map(|p| read_path(&p).ok())
2745 .collect();
2746 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
2747 tasks
2748 }
2749
2750 pub fn superseded(&self) -> HashMap<String, String> {
2763 let mut by = HashMap::new();
2764 for task in self.list() {
2765 for earlier in &task.runs {
2766 if let Some(later) = task.successor_of(earlier) {
2767 by.insert(earlier.clone(), later.clone());
2768 }
2769 }
2770 }
2771 by
2772 }
2773
2774 pub fn superseded_by(&self, run: &str) -> Option<String> {
2782 for task in self.list() {
2783 if task.runs.iter().any(|r| r == run) {
2784 return task.successor_of(run).cloned();
2785 }
2786 }
2787 None
2788 }
2789
2790 pub fn latest_attempt(&self, run: &str) -> Option<String> {
2801 for task in self.list() {
2802 if task.runs.iter().any(|r| r == run) {
2803 return task.runs.last().filter(|last| **last != run).cloned();
2804 }
2805 }
2806 None
2807 }
2808
2809 pub fn next_runnable(&self) -> Option<Task> {
2814 let mut runnable: Vec<Task> = self
2815 .list()
2816 .into_iter()
2817 .filter(|t| t.status.runnable())
2818 .collect();
2819 runnable.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
2820 runnable.into_iter().next()
2821 }
2822
2823 pub fn claim(&self, id: &str) -> Result<Claim> {
2830 std::fs::create_dir_all(&self.root)
2831 .with_context(|| format!("create {}", self.root.display()))?;
2832 let path = self.lock_path(id);
2833 let retry = crate::lock_file::Retry::new(None);
2834 match retry.create(&path) {
2835 Ok(mut f) => {
2836 use std::io::Write as _;
2837 let _ = writeln!(f, "{}", std::process::id());
2839 Ok(Claim { path })
2840 }
2841 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2842 bail!("task {id} is already claimed ({} exists)", path.display())
2843 }
2844 Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
2845 }
2846 }
2847
2848 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
2850 if !prefix.is_empty()
2851 && prefix != "."
2852 && prefix != ".."
2853 && !prefix.contains(['/', '\\'])
2854 && self.path_of(prefix).is_file()
2855 {
2856 return Ok(prefix.to_owned());
2857 }
2858 let ids = record_ids(&self.root, "json")?;
2859 match resolve_reference(prefix, ids.iter().map(String::as_str)) {
2860 Reference::Found(id) => Ok(id),
2861 Reference::Missing => bail!("no task matches `{prefix}`"),
2862 Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
2863 }
2864 }
2865
2866 pub fn revision(&self) -> u64 {
2873 self.revision_excluding(&std::collections::BTreeSet::new())
2874 }
2875
2876 pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
2879 use std::hash::{Hash as _, Hasher as _};
2880
2881 let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
2882 .into_iter()
2883 .flatten()
2884 .flatten()
2885 .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
2886 .filter(|e| {
2887 let path = e.path();
2888 !path
2889 .file_stem()
2890 .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
2891 })
2892 .filter_map(|e| {
2893 let name = e.file_name().to_string_lossy().into_owned();
2894 let mtime = e
2895 .metadata()
2896 .ok()?
2897 .modified()
2898 .ok()?
2899 .duration_since(std::time::UNIX_EPOCH)
2900 .ok()?
2901 .as_millis() as u64;
2902 Some((name, mtime))
2903 })
2904 .collect();
2905
2906 if entries.is_empty() {
2907 return 0;
2908 }
2909
2910 entries.sort_unstable();
2911 let mut hasher = std::hash::DefaultHasher::new();
2912 for (name, mtime) in &entries {
2913 name.hash(&mut hasher);
2914 mtime.hash(&mut hasher);
2915 }
2916 let h = hasher.finish();
2917 if h == 0 { 1 } else { h }
2918 }
2919}
2920
2921const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
2924
2925#[derive(Debug, Clone)]
2927pub struct Removal {
2928 pub id: String,
2930 pub released: Vec<String>,
2934 pub still_blocked: Vec<String>,
2937}
2938
2939#[derive(Debug)]
2941pub struct Claim {
2942 path: PathBuf,
2943}
2944
2945impl Drop for Claim {
2946 fn drop(&mut self) {
2947 let _ = std::fs::remove_file(&self.path);
2948 }
2949}
2950
2951pub fn title_from(instruction: &str, max: usize) -> String {
2954 let Some(line) = first_line(instruction) else {
2955 return "(empty task)".to_owned();
2956 };
2957 if line.chars().count() <= max {
2958 return line.to_owned();
2959 }
2960 let head: String = line.chars().take(max.saturating_sub(1)).collect();
2961 format!("{head}…")
2962}
2963
2964pub(crate) fn first_line(instruction: &str) -> Option<&str> {
2970 let line = instruction
2971 .lines()
2972 .map(str::trim)
2973 .find(|l| !l.is_empty())?
2974 .trim_start_matches(['#', '-', '*', '>', ' '])
2975 .trim();
2976 (!line.is_empty()).then_some(line)
2977}
2978
2979fn read_path(path: &Path) -> Result<Task> {
2980 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
2981 let task: Task =
2982 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2983 if task.schema > SCHEMA {
2989 bail!(
2990 "task {} was written by a different magi (schema {}, this build \
2991 speaks {SCHEMA})",
2992 task.id,
2993 task.schema
2994 );
2995 }
2996 Ok(task)
2997}
2998
2999#[derive(Debug, PartialEq, Eq)]
3002pub(crate) enum Reference {
3003 Found(String),
3004 Missing,
3005 Ambiguous,
3006}
3007
3008pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
3009 let mut hits = Vec::new();
3010 for candidate in ids {
3011 if candidate == id {
3012 return Reference::Found(candidate.to_owned());
3013 }
3014 if candidate.starts_with(id) || candidate.ends_with(id) {
3015 hits.push(candidate);
3016 }
3017 }
3018 match hits.as_slice() {
3019 [] => Reference::Missing,
3020 [one] => Reference::Found((*one).to_owned()),
3021 _ => Reference::Ambiguous,
3022 }
3023}
3024
3025pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
3028 let entries = match std::fs::read_dir(root) {
3029 Ok(entries) => entries,
3030 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
3031 Err(e) => return Err(e.into()),
3032 };
3033 let mut ids = Vec::new();
3034 for entry in entries {
3035 let path = entry?.path();
3036 if path.extension().is_some_and(|e| e == extension)
3037 && let Some(id) = path.file_stem().and_then(|s| s.to_str())
3038 {
3039 ids.push(id.to_owned());
3040 }
3041 }
3042 Ok(ids)
3043}
3044
3045#[derive(Debug, PartialEq, Eq)]
3048pub enum Dependency {
3049 Task(String),
3051 Question(String),
3053 Deleted(String),
3055 Missing,
3057 Ambiguous,
3059 Unknown,
3061}
3062
3063pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
3065 let (Ok(tasks), Ok(qs), Ok(removed)) = (
3066 record_ids(queue.root(), "json"),
3067 record_ids(questions.root(), "json"),
3068 record_ids(queue.root(), "removed"),
3069 ) else {
3070 return Dependency::Unknown;
3071 };
3072 let mut ids = tasks
3075 .iter()
3076 .chain(qs.iter())
3077 .chain(removed.iter())
3078 .cloned()
3079 .collect::<Vec<_>>();
3080 ids.sort();
3081 ids.dedup();
3082 match resolve_reference(id, ids.iter().map(String::as_str)) {
3083 Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
3084 Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
3085 Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
3086 Reference::Found(_) => Dependency::Unknown,
3087 Reference::Missing => Dependency::Missing,
3088 Reference::Ambiguous => Dependency::Ambiguous,
3089 }
3090}
3091
3092pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
3094 for id in ids {
3095 if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
3096 dependency(queue, questions, id)
3097 {
3098 *id = full;
3099 }
3100 }
3101}
3102
3103pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
3106 let inventory = crate::blockers::Inventory::load(queue, questions);
3107 let mut text = String::from(
3108 "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
3109 );
3110 match record_ids(queue.root(), "json") {
3111 Ok(ids) => {
3112 for id in ids {
3113 match queue.get(&id) {
3114 Ok(task) => text.push_str(&format!(
3115 "{}: {}; blocked_by={:?}; chain={:?}\n",
3116 task.id,
3117 task.status.as_str(),
3118 task.blocked_by,
3119 inventory.waits_on(&task)
3120 )),
3121 Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
3122 }
3123 }
3124 }
3125 Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
3126 }
3127 text
3128}
3129
3130pub fn missing_blockers(
3132 queue: &Queue,
3133 questions: &Questions,
3134 blocked_by: &[String],
3135) -> Vec<String> {
3136 blocked_by
3137 .iter()
3138 .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
3139 .cloned()
3140 .collect()
3141}
3142
3143pub fn deleted_blockers(
3146 queue: &Queue,
3147 questions: &Questions,
3148 blocked_by: &[String],
3149) -> Vec<String> {
3150 blocked_by
3151 .iter()
3152 .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
3153 .cloned()
3154 .collect()
3155}
3156
3157pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
3169 missing_blocker_hold_reason_in(blocked_by, missing, "en")
3170}
3171
3172pub fn missing_blocker_hold_reason_in(
3174 blocked_by: &[String],
3175 missing: &[String],
3176 language: &str,
3177) -> String {
3178 if crate::lang::is_japanese(language) {
3179 format!(
3180 "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
3181 blocked_by.join(", "),
3182 missing.join(", "),
3183 )
3184 } else {
3185 format!(
3186 "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
3187 blocked_by.join(", "),
3188 missing.join(", "),
3189 )
3190 }
3191}
3192
3193pub fn short(id: &str) -> &str {
3195 id.split('-').next_back().unwrap_or(id)
3196}
3197
3198fn new_id() -> String {
3199 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
3200 let seed = crate::rng::entropy();
3201 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
3202}
3203
3204#[cfg(test)]
3205mod tests {
3206 #[test]
3207 fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
3208 let t = Task::new(
3209 "retry".into(),
3210 "do it".into(),
3211 PathBuf::from("."),
3212 Source::Human,
3213 );
3214 let mut value = serde_json::to_value(&t).unwrap();
3215 for field in ["resumption", "resumption_history", "operator_decided_at"] {
3216 value.as_object_mut().unwrap().remove(field);
3217 }
3218 value["schema"] = serde_json::json!(15);
3219 let mut legacy: Task = serde_json::from_value(value).unwrap();
3220 legacy.hold_machine(Some("automatic retry".into()));
3221 legacy.release();
3222 assert!(legacy.resumption.is_none());
3223 assert!(legacy.resumption_history.is_empty());
3224 assert!(legacy.operator_decided_at.is_none());
3225 legacy.release_operator("cli", None);
3226 legacy.start("run".into());
3227 let reloaded: Task =
3228 serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
3229 assert!(reloaded.resume_override.is_none());
3230 assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
3231 }
3232
3233 #[test]
3234 fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
3235 let mut base = RunOverrides {
3236 merge: Some("pr".to_owned()),
3237 candidates: Some(3),
3238 ..RunOverrides::default()
3239 };
3240 base.merge_over(&RunOverrides {
3241 merge: Some("none".to_owned()),
3242 seed: Some(7),
3243 ..RunOverrides::default()
3244 });
3245 assert_eq!(base.merge.as_deref(), Some("none"));
3246 assert_eq!(base.candidates, Some(3));
3247 assert_eq!(base.seed, Some(7));
3248 }
3249
3250 #[test]
3251 fn an_already_landed_task_is_done_with_its_attempt_refunded() {
3252 let mut t = task("relanded");
3253 t.attempts = 1;
3254 t.status = TaskStatus::Running;
3255 t.already_landed("already in main as 0e368de");
3256 assert_eq!(t.status, TaskStatus::Done);
3257 assert_eq!(t.attempts, 0);
3258 assert!(t.hold_reason.is_none());
3259 assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
3260 }
3261
3262 #[test]
3263 fn missing_blocker_reason_follows_the_language() {
3264 let b = vec!["a".to_owned()];
3265 let en = missing_blocker_hold_reason_in(&b, &b, "en");
3266 assert_eq!(en, missing_blocker_hold_reason(&b, &b));
3267 assert!(en.starts_with("blocked on a"));
3268 assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
3269 assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
3270 }
3271
3272 use super::*;
3273
3274 #[test]
3275 fn triage_applied_survives_release_and_old_records_read_as_empty() {
3276 let mut t = Task::new(
3277 "t".to_owned(),
3278 "i".to_owned(),
3279 PathBuf::from("r"),
3280 Source::Human,
3281 );
3282 t.mark_triage_applied("q1");
3283 t.mark_triage_applied("q1");
3284 t.hold_machine(Some("x".to_owned()));
3285 t.release();
3286 assert_eq!(t.triage_applied, ["q1"]);
3287 assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
3288
3289 let mut v = serde_json::to_value(&t).unwrap();
3290 v.as_object_mut().unwrap().remove("triage_applied");
3291 let old: Task = serde_json::from_value(v).unwrap();
3292 assert!(old.triage_applied.is_empty());
3293 }
3294
3295 #[test]
3296 fn task_counts_of_empty_is_all_zero() {
3297 assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
3298 }
3299
3300 #[test]
3301 fn task_counts_of_tallies_every_status() {
3302 let mut queued = Task::new(
3303 "q".to_owned(),
3304 "i".to_owned(),
3305 PathBuf::from("."),
3306 Source::Human,
3307 );
3308 queued.status = TaskStatus::Queued;
3309 let mut running = queued.clone();
3310 running.status = TaskStatus::Running;
3311 let mut done = queued.clone();
3312 done.status = TaskStatus::Done;
3313 let mut failed = queued.clone();
3314 failed.status = TaskStatus::Failed;
3315 let mut held = queued.clone();
3316 held.status = TaskStatus::Held;
3317 let mut blocked = queued.clone();
3318 blocked.status = TaskStatus::Blocked;
3319
3320 let mut parked = queued.clone();
3321 parked.status = TaskStatus::Parked;
3322
3323 let counts = TaskCounts::of(&[
3324 queued,
3325 running,
3326 done.clone(),
3327 done,
3328 failed,
3329 held,
3330 blocked,
3331 parked,
3332 ]);
3333 assert_eq!(
3334 counts,
3335 TaskCounts {
3336 queued: 1,
3337 running: 1,
3338 done: 2,
3339 failed: 1,
3340 held: 1,
3341 blocked: 1,
3342 parked: 1,
3343 }
3344 );
3345 }
3346
3347 #[test]
3348 fn task_implementers_persist_normalized_without_losing_history() {
3349 let (_dir, q) = queue();
3350 let mut t = task("keep me");
3351 t.hold_manual(Some("waiting".into()));
3352 t.runs.push("earlier-run".into());
3353 t.attachments.push("reference.txt".into());
3354 let cfg: crate::config::Config =
3355 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
3356 .unwrap();
3357 q.put(&mut t).unwrap();
3358 let original = serde_json::to_value(&t).unwrap();
3359 for n in [2, 1, 2] {
3360 t.set_implementers(n, &cfg).unwrap();
3361 q.put(&mut t).unwrap();
3362 t = q.get(&t.id).unwrap();
3363 assert_eq!(t.implementers(), Some(usize::from(n)));
3364 assert_eq!(t.solo, n == 1);
3365 if n == 1 {
3366 assert!(t.overrides.is_none());
3367 }
3368 let mut actual = serde_json::to_value(&t).unwrap();
3369 for field in ["solo", "overrides", "updated_at"] {
3370 actual[field] = original[field].clone();
3371 }
3372 assert_eq!(actual, original);
3373 }
3374 t.overrides.as_mut().unwrap().judges = Some(2);
3375 t.set_implementers(1, &cfg).unwrap();
3376 assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
3377 assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
3378 }
3379
3380 #[test]
3381 fn task_implementers_old_records_keep_inheritance_until_changed() {
3382 let mut value = serde_json::to_value(task("old task")).unwrap();
3383 value.as_object_mut().unwrap().remove("solo");
3384 value.as_object_mut().unwrap().remove("overrides");
3385 let mut old: Task = serde_json::from_value(value).unwrap();
3386 assert_eq!(old.implementers(), None);
3387 old.set_implementers(1, &crate::config::Config::default())
3388 .unwrap();
3389 assert!(old.solo);
3390 assert!(old.overrides.is_none());
3391 }
3392
3393 #[test]
3394 fn task_implementers_reject_invalid_counts_and_states() {
3395 let cfg = crate::config::Config::default();
3396 let mut t = task("unchanged");
3397 for n in [0, 4, 255] {
3398 assert!(t.set_implementers(n, &cfg).is_err());
3399 }
3400 for status in [
3401 TaskStatus::Running,
3402 TaskStatus::Done,
3403 TaskStatus::Failed,
3404 TaskStatus::Blocked,
3405 TaskStatus::Parked,
3406 ] {
3407 t.status = status;
3408 let before = serde_json::to_value(&t).unwrap();
3409 assert!(t.set_implementers(1, &cfg).is_err());
3410 assert_eq!(serde_json::to_value(&t).unwrap(), before);
3411 }
3412 }
3413
3414 fn queue() -> (tempfile::TempDir, Queue) {
3417 let dir = tempfile::tempdir().unwrap();
3418 let q = Queue::at(dir.path().join("queue"));
3419 (dir, q)
3420 }
3421
3422 #[test]
3423 fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
3424 let dir = tempfile::tempdir().unwrap();
3425 let q = Queue::at(dir.path().join("queue"));
3426 let mut t = task("held");
3427 q.put(&mut t).unwrap();
3428 assert_eq!(
3429 crate::notices::Notices::at(dir.path().join("notifications"))
3430 .list()
3431 .len(),
3432 0
3433 );
3434 t.hold_machine(Some("out of attempts".to_owned()));
3435 q.put(&mut t).unwrap();
3436 let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
3437 assert_eq!(listed.len(), 1);
3438 assert!(listed[0].message.contains("out of attempts"));
3439 }
3440
3441 fn task(title: &str) -> Task {
3442 Task::new(
3443 title.to_owned(),
3444 format!("do {title}"),
3445 PathBuf::from("."),
3446 Source::Human,
3447 )
3448 }
3449
3450 #[test]
3451 fn attempt_limit_inherits_old_records_and_controls_failure() {
3452 let mut t = task("budget");
3453 let mut old = serde_json::to_value(&t).unwrap();
3454 old.as_object_mut().unwrap().remove("attempt_limit");
3455 old["schema"] = serde_json::json!(15);
3456 let legacy: Task = serde_json::from_value(old).unwrap();
3457 assert_eq!(legacy.attempt_limit, None);
3458 assert_eq!(legacy.effective_attempt_limit(2), 2);
3459 t.attempts = 2;
3460 t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
3461 t.fail("retry", 1);
3462 assert_eq!(t.status, TaskStatus::Failed);
3463 t.attempts = 3;
3464 t.fail("stop", 10);
3465 assert_eq!(t.status, TaskStatus::Held);
3466 t.change_attempt_limit(None).unwrap();
3467 assert_eq!(t.status, TaskStatus::Held);
3468 assert_eq!(t.attempts, 3);
3469 t.fail("inherit", 4);
3470 assert_eq!(t.status, TaskStatus::Failed);
3471 for status in [
3472 TaskStatus::Running,
3473 TaskStatus::Done,
3474 TaskStatus::Failed,
3475 TaskStatus::Blocked,
3476 TaskStatus::Parked,
3477 ] {
3478 t.status = status;
3479 assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
3480 assert_eq!(t.attempt_limit, None);
3481 }
3482 }
3483
3484 #[test]
3485 fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
3486 let mut t = task("retried");
3487 assert!(
3488 t.earlier_attempts().is_empty(),
3489 "a first attempt takes nothing over"
3490 );
3491 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3492 assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
3493 assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
3495 assert_eq!(t.successor_of("bbbb"), None);
3496 assert_eq!(t.successor_of("zzzz"), None);
3497 }
3498
3499 #[test]
3500 fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
3501 let (_dir, q) = queue();
3502 let mut t = task("resumed");
3503 for (runs, successor) in [
3504 (vec!["aaaa", "aaaa"], None),
3505 (vec!["aaaa", "bbbb", "aaaa"], None),
3506 (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
3507 (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
3508 ] {
3509 t.runs = runs.into_iter().map(str::to_owned).collect();
3510 q.put(&mut t).unwrap();
3511
3512 assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
3513 assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
3514 assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
3515 }
3516 }
3517
3518 #[test]
3519 fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
3520 let (_dir, q) = queue();
3521 let mut t = task("retried");
3522 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3523 q.put(&mut t).unwrap();
3524
3525 assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
3526 assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
3527 assert_eq!(
3528 q.superseded_by("cccc"),
3529 None,
3530 "the latest attempt replaces nothing"
3531 );
3532 assert_eq!(
3533 q.superseded_by("never-heard-of-it"),
3534 None,
3535 "a run belonging to no task on this queue is not superseded"
3536 );
3537
3538 let mut by = HashMap::new();
3539 by.insert("aaaa".to_owned(), "bbbb".to_owned());
3540 by.insert("bbbb".to_owned(), "cccc".to_owned());
3541 assert_eq!(
3542 q.superseded(),
3543 by,
3544 "the whole-map and single-run forms must agree"
3545 );
3546 }
3547
3548 #[test]
3549 fn superseded_attempts_is_empty_until_the_task_is_done() {
3550 let mut t = task("retried");
3551 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3552 t.status = TaskStatus::Failed;
3553 assert_eq!(
3554 t.superseded_attempts(true),
3555 &[] as &[String],
3556 "a task still retrying has no attempt yet that a later one made moot"
3557 );
3558
3559 t.status = TaskStatus::Running;
3560 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3561 }
3562
3563 #[test]
3564 fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
3565 let mut t = task("retried");
3566 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3567 t.status = TaskStatus::Done;
3568 assert_eq!(
3569 t.superseded_attempts(true),
3570 &["aaaa".to_owned(), "bbbb".to_owned()],
3571 "cccc is the attempt whose success made the task done, and stays out"
3572 );
3573 }
3574
3575 #[test]
3576 fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
3577 let mut t = task("first try landed");
3578 t.runs = vec!["aaaa".to_owned()];
3579 t.status = TaskStatus::Done;
3580 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3581 }
3582
3583 #[test]
3584 fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
3585 let mut t = task("closed by hand after a manual merge");
3592 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3593 t.status = TaskStatus::Done;
3594 assert_eq!(
3595 t.superseded_attempts(false),
3596 &[] as &[String],
3597 "nothing here is provably why the task is done, so nothing is superseded"
3598 );
3599 }
3600
3601 #[test]
3602 fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
3603 let (_dir, q) = queue();
3604 let mut t = task("retried twice");
3605 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3606 q.put(&mut t).unwrap();
3607
3608 assert_eq!(
3609 q.latest_attempt("aaaa"),
3610 Some("cccc".to_owned()),
3611 "an old attempt points straight at the chain's current head, not the \
3612 next attempt in the middle of it"
3613 );
3614 assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
3615 assert_eq!(
3616 q.latest_attempt("cccc"),
3617 None,
3618 "the latest attempt is not superseded by anything"
3619 );
3620 assert_eq!(
3621 q.latest_attempt("never-heard-of-it"),
3622 None,
3623 "a run belonging to no task on this queue is not superseded"
3624 );
3625 }
3626
3627 #[test]
3628 fn a_markdown_heading_is_the_title_not_decoration() {
3629 assert_eq!(
3634 title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
3635 "Rework the config loader"
3636 );
3637 assert_eq!(title_from("- fix the thing", 40), "fix the thing");
3638 assert_eq!(title_from("> quoted task", 40), "quoted task");
3639 assert_eq!(title_from(" \n\n", 40), "(empty task)");
3641 assert_eq!(title_from("###\n", 40), "(empty task)");
3642 }
3643
3644 #[test]
3645 fn a_long_title_is_elided_by_characters_not_bytes() {
3646 let long = "課題".repeat(30);
3648 let title = title_from(&long, 10);
3649 assert_eq!(title.chars().count(), 10);
3650 assert!(title.ends_with('…'));
3651 }
3652
3653 #[test]
3654 fn priority_wins_and_ties_break_oldest_first() {
3655 let (_dir, q) = queue();
3656 let mut a = task("first");
3657 let mut b = task("second");
3658 let mut c = task("urgent");
3659 a.id = "20260101-000001-aaaa".to_owned();
3661 b.id = "20260101-000002-bbbb".to_owned();
3662 c.id = "20260101-000003-cccc".to_owned();
3663 c.priority = 5;
3664 for t in [&mut a, &mut b, &mut c] {
3665 q.put(t).unwrap();
3666 }
3667
3668 assert_eq!(q.next_runnable().unwrap().id, c.id);
3670 c.hold_machine(None);
3671 q.put(&mut c).unwrap();
3672 assert_eq!(q.next_runnable().unwrap().id, a.id);
3674 assert_eq!(q.list().len(), 3, "b is still waiting its turn");
3675 }
3676
3677 #[test]
3678 fn a_blocked_task_never_starves_another_runnable_one() {
3679 let (_dir, q) = queue();
3680 let mut blocked = task("blocked");
3681 blocked.block(vec!["something".to_owned()], None);
3682 q.put(&mut blocked).unwrap();
3683
3684 let mut runnable = task("free to go");
3685 q.put(&mut runnable).unwrap();
3686
3687 let next = q.next_runnable().expect("a runnable task is still offered");
3688 assert_eq!(next.id, runnable.id);
3689 }
3690
3691 #[test]
3692 fn a_held_task_is_never_offered_to_the_loop() {
3693 let (_dir, q) = queue();
3694 let mut t = task("held");
3695 q.put(&mut t).unwrap();
3696 assert!(q.next_runnable().is_some());
3697
3698 t.hold_machine(None);
3699 q.put(&mut t).unwrap();
3700 assert!(
3701 q.next_runnable().is_none(),
3702 "a held task must wait for a human"
3703 );
3704
3705 t.status = TaskStatus::Failed;
3707 q.put(&mut t).unwrap();
3708 assert!(q.next_runnable().is_some());
3709 }
3710
3711 #[test]
3712 fn attempts_are_capped_and_then_the_task_is_held() {
3713 let mut t = task("doomed");
3714
3715 t.start("run-1".to_owned());
3716 t.fail("gate red", 2);
3717 assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
3718
3719 t.start("run-2".to_owned());
3720 t.fail("gate red", 2);
3721 assert_eq!(
3722 t.status,
3723 TaskStatus::Held,
3724 "out of attempts: stop spending money on it"
3725 );
3726 assert_eq!(t.runs, ["run-1", "run-2"]);
3727 assert_eq!(t.last_error.as_deref(), Some("gate red"));
3728 assert_eq!(
3729 t.hold_reason.as_deref(),
3730 Some("gate red"),
3731 "the hold must say why, not leave hold_reason null next to a \
3732 populated last_error"
3733 );
3734 }
3735
3736 #[test]
3737 fn handing_off_a_task_records_a_hold_reason_too() {
3738 let mut t = task("left a pull request");
3739 t.start("run-1".to_owned());
3740 t.handed_off("run ended with a pull request open [run run-1]");
3741 assert_eq!(t.status, TaskStatus::Held);
3742 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3743 assert_eq!(
3744 t.hold_reason.as_deref(),
3745 Some("run ended with a pull request open [run run-1]")
3746 );
3747 assert_eq!(t.hold_reason, t.last_error);
3748 }
3749
3750 #[test]
3751 fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
3752 let mut t = task("stalled by quota");
3753
3754 t.start("run-1".to_owned());
3755 assert_eq!(t.attempts, 1);
3756 t.stall("judge-1, judge-2 out of quota");
3757 assert_eq!(
3758 t.attempts, 0,
3759 "a closed quota window must not spend the task's retry budget"
3760 );
3761 assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
3762 assert_eq!(
3763 t.last_error.as_deref(),
3764 Some("judge-1, judge-2 out of quota")
3765 );
3766
3767 for _ in 0..20 {
3770 t.start("run-n".to_owned());
3771 t.stall("still out of quota");
3772 }
3773 t.start("run-real".to_owned());
3774 t.fail("gate red", 2);
3775 assert_eq!(
3776 t.status,
3777 TaskStatus::Failed,
3778 "the first attempt that was really judged is attempt one"
3779 );
3780 }
3781
3782 #[test]
3783 fn releasing_a_held_task_gives_it_a_real_second_chance() {
3784 let mut t = task("retry me");
3785 t.start("run-1".to_owned());
3786 t.fail("gate red", 1);
3787 assert_eq!(t.status, TaskStatus::Held);
3788
3789 t.release();
3790 assert_eq!(t.status, TaskStatus::Queued);
3791 assert_eq!(t.attempts, 0);
3794 assert!(t.last_error.is_none());
3795 assert_eq!(
3796 t.runs.len(),
3797 1,
3798 "history is kept: attempts reset, evidence does not"
3799 );
3800 }
3801
3802 #[test]
3803 fn a_hold_reason_survives_and_a_release_clears_it() {
3804 let mut t = task("waiting on something else");
3805 t.hold_manual(Some(
3806 "waiting for 20260101-000000-aaaa to land first".to_owned(),
3807 ));
3808 assert_eq!(t.status, TaskStatus::Held);
3809 assert_eq!(
3810 t.hold_reason.as_deref(),
3811 Some("waiting for 20260101-000000-aaaa to land first")
3812 );
3813
3814 t.hold_manual(None);
3816 assert_eq!(
3817 t.hold_reason.as_deref(),
3818 Some("waiting for 20260101-000000-aaaa to land first"),
3819 "a bare re-hold keeps whatever a human already wrote down"
3820 );
3821
3822 let mut plain = task("no reason given");
3824 plain.hold_manual(None);
3825 assert_eq!(plain.status, TaskStatus::Held);
3826 assert!(plain.hold_reason.is_none());
3827
3828 t.release();
3829 assert_eq!(t.status, TaskStatus::Queued);
3830 assert!(
3831 t.hold_reason.is_none(),
3832 "a stale reason must not greet the next person who holds this task"
3833 );
3834 }
3835
3836 #[test]
3837 fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
3838 let mut t = task("landed by hand while held");
3843 t.hold_manual(Some("waiting on 3ed9".to_owned()));
3844 assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
3845
3846 t.succeed();
3847 assert_eq!(t.status, TaskStatus::Done);
3848 assert!(
3849 t.hold_reason.is_none(),
3850 "a done task cannot still be waiting on something"
3851 );
3852 }
3853
3854 #[test]
3855 fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
3856 let mut held = task("held straight out of blocked");
3863 held.block(
3864 vec!["20260101-000000-dead".to_owned()],
3865 Some("waiting on the migration script".to_owned()),
3866 );
3867 assert_eq!(held.status, TaskStatus::Blocked);
3868
3869 held.hold_manual(None);
3870 assert_eq!(held.status, TaskStatus::Held);
3871 assert!(
3872 held.blocked_by.is_empty(),
3873 "hold overrides the wait, same as release"
3874 );
3875 assert!(held.block_reason.is_none());
3876
3877 let mut done = task("closed straight out of blocked");
3878 done.block(
3879 vec!["20260101-000000-dead".to_owned()],
3880 Some("waiting on the migration script".to_owned()),
3881 );
3882 done.succeed();
3883 assert_eq!(done.status, TaskStatus::Done);
3884 assert!(
3885 done.blocked_by.is_empty(),
3886 "a done task cannot still be waiting on a dependency"
3887 );
3888 assert!(done.block_reason.is_none());
3889 }
3890
3891 #[test]
3892 fn a_blocked_task_is_never_offered_to_the_loop() {
3893 let mut t = task("blocked");
3894 assert!(t.status.runnable());
3895 t.block(
3896 vec!["dep-id".to_owned()],
3897 Some("waits on dep-id".to_owned()),
3898 );
3899 assert_eq!(t.status, TaskStatus::Blocked);
3900 assert!(!t.status.runnable());
3901 assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
3902 }
3903
3904 #[test]
3905 fn unblocking_the_last_dependency_returns_the_task_to_queued() {
3906 let mut t = task("blocked on two");
3907 t.block(
3908 vec!["a".to_owned(), "b".to_owned()],
3909 Some("waits on a and b".to_owned()),
3910 );
3911
3912 t.unblock("a");
3913 assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
3914 assert_eq!(t.blocked_by, ["b"]);
3915
3916 t.unblock("b");
3917 assert_eq!(t.status, TaskStatus::Queued);
3918 assert!(t.blocked_by.is_empty());
3919 assert!(t.block_reason.is_none());
3920 }
3921
3922 #[test]
3923 fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
3924 let mut t = task("never blocked");
3925 t.unblock("whatever");
3926 assert_eq!(t.status, TaskStatus::Queued);
3927 }
3928
3929 #[test]
3930 fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
3931 let mut t = task("held, then asked about");
3936 t.hold_machine(Some("out of attempts".to_owned()));
3937 assert_eq!(t.status, TaskStatus::Held);
3938
3939 t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
3940 assert_eq!(t.status, TaskStatus::Blocked);
3941
3942 t.record_answer("what now?".to_owned(), "leave it held".to_owned());
3943 t.unblock("q1");
3944 assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
3945 assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
3946 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3947 assert!(t.blocked_from.is_none(), "consumed once restored");
3948 }
3949
3950 #[test]
3951 fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
3952 let mut t = task("manually held, then asked about");
3953 t.hold_manual(Some("waiting on a dependency".to_owned()));
3954
3955 t.block(vec!["q1".to_owned()], None);
3956 t.unblock("q1");
3957
3958 assert_eq!(t.status, TaskStatus::Held);
3959 assert_eq!(t.hold_source, Some(HoldSource::Manual));
3960 }
3961
3962 #[test]
3963 fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
3964 let mut t = task("held, blocked twice");
3968 t.hold_machine(None);
3969 t.block(vec!["q1".to_owned()], Some("first".to_owned()));
3970 t.block(
3971 vec!["q1".to_owned(), "q2".to_owned()],
3972 Some("second".to_owned()),
3973 );
3974
3975 t.unblock("q1");
3976 assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
3977 t.unblock("q2");
3978 assert_eq!(t.status, TaskStatus::Held);
3979 }
3980
3981 #[test]
3982 fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
3983 let mut t = task("blocked mid-run");
3987 t.start("run-1".to_owned());
3988 assert_eq!(t.status, TaskStatus::Running);
3989
3990 t.block(vec!["q1".to_owned()], None);
3991 t.unblock("q1");
3992 assert_eq!(t.status, TaskStatus::Queued);
3993 }
3994
3995 #[test]
3996 fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
3997 let mut t = task("legacy record, held before it was blocked");
4003 t.hold_source = Some(HoldSource::Machine);
4004 t.hold_reason = Some("legacy hold reason".to_owned());
4005 t.status = TaskStatus::Blocked;
4006 t.blocked_by = vec!["q1".to_owned()];
4007 t.blocked_from = None;
4008
4009 t.unblock("q1");
4010 assert_eq!(t.status, TaskStatus::Held);
4011 }
4012
4013 #[test]
4014 fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
4015 let mut t = task("legacy record, ordinary dependency block");
4016 t.status = TaskStatus::Blocked;
4017 t.blocked_by = vec!["dep".to_owned()];
4018 t.blocked_from = None;
4019
4020 t.unblock("dep");
4021 assert_eq!(t.status, TaskStatus::Queued);
4022 }
4023
4024 #[test]
4025 fn answering_a_question_is_recorded_and_survives_a_release() {
4026 let mut t = task("asked something");
4027 t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
4028 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
4029 t.unblock("q1");
4030 assert_eq!(t.status, TaskStatus::Queued);
4031 assert_eq!(t.answers.len(), 1);
4032 assert_eq!(t.answers[0].answer, "SQLite");
4033
4034 t.release();
4038 assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
4039 }
4040
4041 #[test]
4042 fn resumption_review_branch_requires_unchanged_authored_criteria() {
4043 for source in ["cli", "web"] {
4044 let mut t = task("review old candidate");
4045 t.request_review("candidate-A".into());
4046 t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
4047 let mut unchanged = t.clone();
4048 unchanged.release_operator(source, None);
4049 assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
4050 t.edit(
4051 "review approved head".into(),
4052 "Reuse measured PR head B; fix Clippy; review".into(),
4053 )
4054 .unwrap();
4055 t.release_operator(source, None);
4056 assert!(t.review_branch.is_none());
4057 assert!(t.review_instruction.is_none());
4058 let mut legacy = unchanged;
4060 legacy.review_instruction = None;
4061 legacy.hold_for_handover(None, "still checked out".into());
4062 legacy
4063 .edit("new plan".into(), "Reuse measured PR head B".into())
4064 .unwrap();
4065 legacy.release_operator(source, None);
4066 assert!(legacy.review_branch.is_none());
4067 }
4068 }
4069
4070 #[test]
4071 fn a_refused_handover_keeps_the_review_branch_across_release() {
4072 let mut t = task("refused takeover");
4073 t.start("run-1".to_owned());
4074 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
4075 assert_eq!(t.status, TaskStatus::Held);
4076 assert_eq!(t.attempts, 1);
4077 t.release();
4078 assert_eq!(t.status, TaskStatus::Queued);
4079 assert_eq!(t.attempts, 0);
4080 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4081
4082 t.hold_for_handover(None, "again".to_owned());
4084 t.requeue();
4085 assert!(t.review_branch.is_none());
4086
4087 let mut m = task("manual");
4089 m.review_branch = Some("magi/x/A".to_owned());
4090 m.hold_manual(None);
4091 m.release();
4092 assert!(m.review_branch.is_none());
4093 }
4094
4095 #[test]
4096 fn requesting_review_requeues_the_task_and_remembers_the_branch() {
4097 let mut t = task("blocked run with a surviving branch");
4098 t.start("run-1".to_owned());
4099 t.fail("blocked with major findings", 5);
4100 assert_eq!(t.status, TaskStatus::Failed);
4101
4102 t.request_review("magi/eba2/A".to_owned());
4103 assert_eq!(t.status, TaskStatus::Queued);
4104 assert_eq!(t.attempts, 0);
4105 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4106
4107 t.release();
4109 assert!(t.review_branch.is_none());
4110 }
4111
4112 #[test]
4113 fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
4114 let mut t = task("retry");
4115 t.start("run-1".to_owned());
4116 t.requeue();
4117 assert!(t.fresh_start);
4118
4119 t.release();
4120 assert!(!t.fresh_start);
4121 }
4122
4123 #[test]
4124 fn urgent_only_changes_queued_or_held_and_preserves_settings() {
4125 for status in [
4126 TaskStatus::Queued,
4127 TaskStatus::Held,
4128 TaskStatus::Running,
4129 TaskStatus::Done,
4130 TaskStatus::Failed,
4131 TaskStatus::Blocked,
4132 TaskStatus::Parked,
4133 ] {
4134 let mut t = task("urgent settings");
4135 t.status = status;
4136 t.attempts = 2;
4137 t.hold_reason = Some("waiting".into());
4138 t.priority = 7;
4139 t.interrupt = true;
4140 let before = serde_json::to_value(&t).unwrap();
4141 for urgent in [true, true, false, false] {
4142 let result = t.set_urgent(urgent);
4143 let mut expected = before.clone();
4144 if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
4145 result.unwrap();
4146 expected["urgent"] = urgent.into();
4147 } else {
4148 let error = result.unwrap_err().to_string();
4149 assert!(error.contains(status.as_str()), "{error}");
4150 assert!(error.contains("queued or held"), "{error}");
4151 }
4152 assert_eq!(serde_json::to_value(&t).unwrap(), expected);
4153 }
4154 }
4155 }
4156
4157 #[test]
4158 fn priority_can_be_changed_while_queued_but_not_while_running() {
4159 let mut t = task("reprioritise me");
4160 t.set_priority(5).unwrap();
4161 assert_eq!(t.priority, 5);
4162
4163 t.start("run-1".to_owned());
4164 let err = t.set_priority(9).unwrap_err().to_string();
4165 assert!(err.contains("running"), "{err}");
4166 assert_eq!(t.priority, 5, "the rejected write must not partially apply");
4167 }
4168
4169 #[test]
4170 fn interrupt_can_be_marked_while_queued_but_not_while_running() {
4171 let mut t = task("interrupt me");
4172 assert!(!t.interrupt, "off unless asked, same as any other task");
4173
4174 t.set_interrupt(true).unwrap();
4175 assert!(t.interrupt);
4176
4177 t.start("run-1".to_owned());
4178 assert!(
4179 !t.interrupt,
4180 "the mark is one-shot: dispatching the task fulfils it, \
4181 whatever the run that follows ends up doing"
4182 );
4183 let err = t.set_interrupt(true).unwrap_err().to_string();
4184 assert!(err.contains("running"), "{err}");
4185 t.set_interrupt(false).unwrap();
4188 assert!(!t.interrupt);
4189 }
4190
4191 #[test]
4195 fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
4196 let mut t = task("interrupt me");
4197 t.set_interrupt(true).unwrap();
4198 t.start("run-1".to_owned());
4199 t.fail("mock failure", 5);
4200 assert_eq!(t.status, TaskStatus::Failed);
4201 assert!(
4202 !t.interrupt,
4203 "one attempt already spent the mark; a retry is an ordinary \
4204 requeue, not a fresh interrupt request"
4205 );
4206 }
4207
4208 #[test]
4209 fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
4210 let (_dir, q) = queue();
4211 let mut a = task("first filed");
4212 let mut b = task("second filed");
4213 a.id = "20260101-000001-aaaa".to_owned();
4214 b.id = "20260101-000002-bbbb".to_owned();
4215 q.put(&mut a).unwrap();
4216 q.put(&mut b).unwrap();
4217
4218 assert_eq!(
4219 q.next_runnable().unwrap().id,
4220 a.id,
4221 "with equal priority the older task goes first, so a burst of \
4222 new work cannot starve it"
4223 );
4224 assert_eq!(
4225 q.list()[0].id,
4226 b.id,
4227 "but the list an operator reads is newest first, the same as \
4228 before priority existed - a's turn to run does not make it the \
4229 newest task"
4230 );
4231
4232 let mut a = q.get(&a.id).unwrap();
4233 a.set_priority(10).unwrap();
4234 q.put(&mut a).unwrap();
4235
4236 assert_eq!(
4237 q.next_runnable().unwrap().id,
4238 a.id,
4239 "a raised priority must be reflected the moment it is saved"
4240 );
4241 assert_eq!(
4245 q.list()[0].id,
4246 a.id,
4247 "the raised task must sort first in the list an operator reads, \
4248 not only in next_runnable's own ordering"
4249 );
4250 }
4251
4252 #[test]
4253 fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
4254 let mut t = Task::new(
4255 "old title".to_owned(),
4256 "old instruction".to_owned(),
4257 PathBuf::from("/repo"),
4258 Source::Agent {
4259 run: "20260101-000000-beef".to_owned(),
4260 node: "implement".to_owned(),
4261 },
4262 );
4263 let id = t.id.clone();
4264 let created_at = t.created_at;
4265 t.runs.push("20260101-000000-beef".to_owned());
4266
4267 t.edit("new title".to_owned(), "new instruction".to_owned())
4268 .unwrap();
4269
4270 assert_eq!(t.title, "new title");
4271 assert_eq!(t.instruction, "new instruction");
4272 assert_eq!(t.id, id, "editing must not mint a new id");
4273 assert_eq!(t.created_at, created_at);
4274 assert_eq!(
4275 t.source,
4276 Source::Agent {
4277 run: "20260101-000000-beef".to_owned(),
4278 node: "implement".to_owned(),
4279 },
4280 "editing must not turn agent attribution into human"
4281 );
4282 assert_eq!(t.runs, ["20260101-000000-beef"]);
4283 }
4284
4285 #[test]
4286 fn editing_is_refused_once_a_task_is_running_or_finished() {
4287 let mut running = task("in flight");
4288 running.start("run-1".to_owned());
4289 let err = running
4290 .edit("x".to_owned(), "y".to_owned())
4291 .unwrap_err()
4292 .to_string();
4293 assert!(err.contains("running"), "{err}");
4294
4295 let mut done = task("finished");
4296 done.succeed();
4297 let err = done
4298 .edit("x".to_owned(), "y".to_owned())
4299 .unwrap_err()
4300 .to_string();
4301 assert!(err.contains("done"), "{err}");
4302
4303 let mut queued = task("waiting");
4305 queued.edit("x".to_owned(), "y".to_owned()).unwrap();
4306 let mut held = task("parked");
4307 held.hold_machine(None);
4308 held.edit("x".to_owned(), "y".to_owned()).unwrap();
4309 }
4310
4311 #[test]
4312 fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
4313 let (_dir, q) = queue();
4314 let path = q.path_of("20260101-000000-aaaa");
4315 std::fs::create_dir_all(q.root()).unwrap();
4316 std::fs::write(
4317 &path,
4318 serde_json::json!({
4319 "schema": SCHEMA,
4320 "id": "20260101-000000-aaaa",
4321 "title": "from before hold reasons existed",
4322 "instruction": "from before hold reasons existed",
4323 "repo": ".",
4324 "source": { "kind": "human" },
4325 "status": "held",
4326 "created_at": Timestamp::now().to_string(),
4327 "updated_at": Timestamp::now().to_string(),
4328 })
4329 .to_string(),
4330 )
4331 .unwrap();
4332
4333 let task = q.get("20260101-000000-aaaa").expect("must still read");
4334 assert!(task.hold_reason.is_none());
4335 assert!(task.operator_held());
4336 }
4337
4338 #[test]
4339 fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
4340 let (_dir, q) = queue();
4341 let path = q.path_of("20260101-000000-bbbb");
4342 std::fs::create_dir_all(q.root()).unwrap();
4343 std::fs::write(
4344 &path,
4345 serde_json::json!({
4346 "schema": 2,
4347 "id": "20260101-000000-bbbb",
4348 "title": "old manual recovery",
4349 "instruction": "old manual recovery",
4350 "repo": ".",
4351 "source": { "kind": "human" },
4352 "status": "held",
4353 "hold_reason": "active manual recovery run20260912-224242-daf5",
4354 "created_at": Timestamp::now().to_string(),
4355 "updated_at": Timestamp::now().to_string(),
4356 })
4357 .to_string(),
4358 )
4359 .unwrap();
4360
4361 let task = q.get("20260101-000000-bbbb").expect("must still read");
4362 assert_eq!(task.hold_source, None);
4363 assert!(task.operator_held());
4364 }
4365
4366 #[test]
4367 fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
4368 let (_dir, q) = queue();
4369 let path = q.path_of("20260101-000000-aaaa");
4370 std::fs::create_dir_all(q.root()).unwrap();
4371 std::fs::write(
4372 &path,
4373 serde_json::json!({
4374 "schema": SCHEMA,
4375 "id": "20260101-000000-aaaa",
4376 "title": "from before diagnostics existed",
4377 "instruction": "from before diagnostics existed",
4378 "repo": ".",
4379 "source": { "kind": "human" },
4380 "status": "held",
4381 "created_at": Timestamp::now().to_string(),
4382 "updated_at": Timestamp::now().to_string(),
4383 })
4384 .to_string(),
4385 )
4386 .unwrap();
4387
4388 let task = q.get("20260101-000000-aaaa").expect("must still read");
4389 assert!(task.diagnostic.is_none());
4390 }
4391
4392 #[test]
4393 fn a_schema_1_task_with_no_blocking_fields_still_reads() {
4394 let (_dir, q) = queue();
4398 let path = q.path_of("20260101-000000-aaaa");
4399 std::fs::create_dir_all(q.root()).unwrap();
4400 std::fs::write(
4401 &path,
4402 serde_json::json!({
4403 "schema": 1,
4404 "id": "20260101-000000-aaaa",
4405 "title": "from before blocking existed",
4406 "instruction": "from before blocking existed",
4407 "repo": ".",
4408 "source": { "kind": "human" },
4409 "status": "queued",
4410 "created_at": Timestamp::now().to_string(),
4411 "updated_at": Timestamp::now().to_string(),
4412 })
4413 .to_string(),
4414 )
4415 .unwrap();
4416
4417 let task = q.get("20260101-000000-aaaa").expect("must still read");
4418 assert!(task.blocked_by.is_empty());
4419 assert!(task.block_reason.is_none());
4420 assert!(task.answers.is_empty());
4421 assert!(task.review_branch.is_none());
4422 }
4423
4424 #[test]
4425 fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
4426 let mut held = task("diagnosed");
4431 held.start("run-1".to_owned());
4432 held.fail("gate red", 1);
4433 held.diagnostic = Some("cargo test failed: ...".to_owned());
4434 assert_eq!(held.status, TaskStatus::Held);
4435
4436 held.release();
4437 assert!(held.diagnostic.is_none());
4438
4439 held.diagnostic = Some("cargo test failed: ...".to_owned());
4440 held.succeed();
4441 assert!(held.diagnostic.is_none());
4442 }
4443
4444 #[test]
4445 fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
4446 let mut t = task("retried");
4447 t.start("run-1".to_owned());
4448 t.diagnostic = Some("stale evidence from a previous hold".to_owned());
4449 t.fail("unrelated config error", 5);
4450 assert_eq!(t.status, TaskStatus::Failed);
4451 assert!(
4452 t.diagnostic.is_none(),
4453 "fail() must not let an old diagnostic outlive the run that produced it"
4454 );
4455 }
4456
4457 #[test]
4458 fn a_claim_is_exclusive_and_releases_on_drop() {
4459 let (_dir, q) = queue();
4460 let mut t = task("contended");
4461 q.put(&mut t).unwrap();
4462
4463 let held = q.claim(&t.id).unwrap();
4464 assert!(
4465 q.claim(&t.id).is_err(),
4466 "two daemons must not drive one task into two runs"
4467 );
4468 drop(held);
4469 assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
4470 }
4471
4472 #[test]
4473 fn a_round_trip_survives_disk() {
4474 let (_dir, q) = queue();
4475 let mut t = Task::new(
4476 "titled".to_owned(),
4477 "body".to_owned(),
4478 PathBuf::from("/repo"),
4479 Source::Agent {
4480 run: "20260101-000000-beef".to_owned(),
4481 node: "implement".to_owned(),
4482 },
4483 );
4484 t.priority = 3;
4485 q.put(&mut t).unwrap();
4486
4487 let back = q.get(&t.id).unwrap();
4488 assert_eq!(back.id, t.id);
4489 assert_eq!(back.priority, 3);
4490 assert_eq!(back.source.label(), "implement@beef");
4491 assert_eq!(q.get(t.short()).unwrap().id, t.id);
4493 }
4494
4495 #[test]
4496 fn an_unreadable_task_does_not_take_the_queue_down() {
4497 let (_dir, q) = queue();
4498 let mut t = task("fine");
4499 q.put(&mut t).unwrap();
4500 std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
4501
4502 let listed = q.list();
4503 assert_eq!(listed.len(), 1, "the readable task still lists");
4504 assert_eq!(listed[0].id, t.id);
4505 }
4506
4507 #[test]
4508 fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
4509 let (_dir, q) = queue();
4510 let path = q.path_of("20260101-000000-aaaa");
4511 std::fs::create_dir_all(q.root()).unwrap();
4512 std::fs::write(
4513 &path,
4514 serde_json::json!({
4515 "schema": SCHEMA,
4516 "id": "20260101-000000-aaaa",
4517 "title": "from before solo existed",
4518 "instruction": "from before solo existed",
4519 "repo": ".",
4520 "source": { "kind": "human" },
4521 "status": "queued",
4522 "created_at": Timestamp::now().to_string(),
4523 "updated_at": Timestamp::now().to_string(),
4524 })
4525 .to_string(),
4526 )
4527 .unwrap();
4528
4529 let task = q.get("20260101-000000-aaaa").expect("must still read");
4530 assert!(!task.solo, "a queue file with no `solo` field means false");
4531 }
4532
4533 #[test]
4534 fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
4535 let (_dir, q) = queue();
4536 let path = q.path_of("20260101-000000-bbbb");
4537 std::fs::create_dir_all(q.root()).unwrap();
4538 std::fs::write(
4539 &path,
4540 serde_json::json!({
4541 "schema": SCHEMA,
4542 "id": "20260101-000000-bbbb",
4543 "title": "from before urgent existed",
4544 "instruction": "from before urgent existed",
4545 "repo": ".",
4546 "source": { "kind": "human" },
4547 "status": "queued",
4548 "created_at": Timestamp::now().to_string(),
4549 "updated_at": Timestamp::now().to_string(),
4550 })
4551 .to_string(),
4552 )
4553 .unwrap();
4554
4555 let task = q.get("20260101-000000-bbbb").expect("must still read");
4556 assert!(
4557 !task.urgent,
4558 "a queue file with no `urgent` field means false, same as `solo`"
4559 );
4560 }
4561
4562 #[test]
4563 fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
4564 let (_dir, q) = queue();
4565 let mut t = task("from the future");
4566 q.put(&mut t).unwrap();
4567 let path = q.path_of(&t.id);
4568 let body = std::fs::read_to_string(&path)
4569 .unwrap()
4570 .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
4571 std::fs::write(&path, body).unwrap();
4572
4573 let err = q.get(&t.id).unwrap_err().to_string();
4574 assert!(err.contains("schema 99"), "{err}");
4575 }
4576
4577 #[test]
4578 fn revision_moves_when_the_queue_changes() {
4579 let (_dir, q) = queue();
4580 assert_eq!(q.revision(), 0, "an empty queue has no revision");
4581 let mut t = task("first");
4582 q.put(&mut t).unwrap();
4583 assert!(q.revision() > 0, "a written task moves the revision");
4584 }
4585
4586 #[test]
4587 fn revision_moves_when_deleting_an_older_task() {
4588 let (dir, q) = queue();
4589 let questions = Questions::at(dir.path().join("questions"));
4590 let mut t1 = task("older");
4591 q.put(&mut t1).unwrap();
4592 std::thread::sleep(std::time::Duration::from_millis(10));
4594 let mut t2 = task("newer");
4595 q.put(&mut t2).unwrap();
4596
4597 let rev_before = q.revision();
4598 q.remove(&t1.id, false, &questions).unwrap();
4599 let rev_after = q.revision();
4600
4601 assert_ne!(
4602 rev_before, rev_after,
4603 "deleting an older task must change the revision so other clients see the deletion"
4604 );
4605 }
4606
4607 #[test]
4608 fn removing_a_task_takes_it_out_of_the_listing() {
4609 let (dir, q) = queue();
4610 let questions = Questions::at(dir.path().join("questions"));
4611 let mut t = task("delete me");
4612 q.put(&mut t).unwrap();
4613 let removed = q.remove(t.short(), false, &questions).unwrap();
4614 assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
4615 assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
4616 assert!(q.list().is_empty());
4617 assert!(
4618 q.remove(&t.id, false, &questions).is_err(),
4619 "removing twice is an error"
4620 );
4621 }
4622
4623 #[test]
4624 fn removing_a_task_takes_its_stale_lock_with_it() {
4625 let (dir, q) = queue();
4626 let questions = Questions::at(dir.path().join("questions"));
4627 let mut t = task("interrupted");
4628 q.put(&mut t).unwrap();
4629
4630 let claim = q.claim(&t.id).unwrap();
4633 std::mem::forget(claim);
4634 assert!(
4635 q.claim(&t.id).is_err(),
4636 "the orphaned lock is what makes the task look claimed"
4637 );
4638
4639 let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
4641 assert!(err.contains("live daemon"), "{err}");
4642 assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
4643
4644 q.remove(&t.id, false, &questions).unwrap();
4646 assert!(q.list().is_empty());
4647 let mut again = task("interrupted");
4648 again.id = t.id.clone();
4649 q.put(&mut again).unwrap();
4650 assert!(
4651 q.claim(&t.id).is_ok(),
4652 "a task that comes back must be claimable, which a left-behind lock would prevent"
4653 );
4654 }
4655
4656 fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
4657 crate::notices::Notices::at(dir.join("notifications")).list()
4658 }
4659
4660 #[test]
4661 fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
4662 let (dir, q) = queue();
4663 let questions = Questions::at(dir.path().join("questions"));
4664 let mut dep = task("dependency");
4665 q.put(&mut dep).unwrap();
4666 let mut blocked = task("waiting");
4667 blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
4668 q.put(&mut blocked).unwrap();
4669
4670 let removed = q.remove(&dep.id, false, &questions).unwrap();
4671 assert_eq!(removed.released, [blocked.id.clone()]);
4672 assert!(removed.still_blocked.is_empty());
4673
4674 let after = q.get(&blocked.id).unwrap();
4675 assert_eq!(after.status, TaskStatus::Queued);
4676 assert!(after.blocked_by.is_empty());
4677 assert!(after.block_reason.is_none());
4678 let notes = notices_of(dir.path());
4679 assert_eq!(notes.len(), 1, "{notes:?}");
4680 assert_eq!(notes[0].severity, crate::notices::Severity::Info);
4681 assert_eq!(
4682 notes[0].link,
4683 Some(crate::notices::Link::Task {
4684 id: blocked.id.clone(),
4685 })
4686 );
4687 }
4688
4689 #[test]
4690 fn removing_one_of_two_dependencies_keeps_the_other() {
4691 let (dir, q) = queue();
4692 let questions = Questions::at(dir.path().join("questions"));
4693 let mut dep = task("dependency");
4694 q.put(&mut dep).unwrap();
4695 let mut other = task("other");
4696 q.put(&mut other).unwrap();
4697 let mut blocked = task("waiting");
4698 blocked.block(
4699 vec![dep.id.clone(), other.id.clone()],
4700 Some("waits on both".to_owned()),
4701 );
4702 q.put(&mut blocked).unwrap();
4703
4704 let removed = q.remove(&dep.id, false, &questions).unwrap();
4705 assert!(removed.released.is_empty());
4706 assert_eq!(removed.still_blocked, [blocked.id.clone()]);
4707
4708 let after = q.get(&blocked.id).unwrap();
4709 assert_eq!(after.status, TaskStatus::Blocked);
4710 assert_eq!(after.blocked_by, [other.id.clone()]);
4711 assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
4714 }
4715
4716 #[test]
4717 fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
4718 let (dir, q) = queue();
4719 let questions = Questions::at(dir.path().join("questions"));
4720 let mut dep = task("dependency");
4721 q.put(&mut dep).unwrap();
4722 let mut blocked = task("waiting");
4723 blocked.block(vec![dep.id.clone()], None);
4724 q.put(&mut blocked).unwrap();
4725
4726 let claim = q.claim(&blocked.id).unwrap();
4728 let removed = q.remove(&dep.id, false, &questions).unwrap();
4729 assert!(removed.released.is_empty());
4730 drop(claim);
4731
4732 let mut task = q.get(&blocked.id).unwrap();
4733 assert_eq!(task.status, TaskStatus::Blocked);
4734 assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
4735 assert_eq!(
4736 q.apply_deleted_blockers(&mut task, &questions),
4737 [dep.id.clone()]
4738 );
4739 assert_eq!(task.status, TaskStatus::Queued);
4740 }
4741
4742 #[test]
4743 fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
4744 let (dir, q) = queue();
4745 let questions = Questions::at(dir.path().join("questions"));
4746 let mut dep = task("dependency");
4747 dep.id = "20261013-193611-ca5b".into();
4748 q.put(&mut dep).unwrap();
4749 let mut waiting = task("waiting");
4750 waiting.block(vec!["ca5b".into()], None);
4751 q.put(&mut waiting).unwrap();
4752 q.remove(&dep.id, false, &questions).unwrap();
4753 assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
4754 assert_eq!(
4755 dependency(&q, &questions, "ca5b"),
4756 Dependency::Deleted(dep.id)
4757 );
4758 }
4759
4760 #[test]
4761 fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
4762 let (dir, queue) = queue();
4763 let questions = Questions::at(dir.path().join("questions"));
4764 let mut dep = task("dependency");
4765 dep.id = "20261013-193611-ca5b".into();
4766 queue.put(&mut dep).unwrap();
4767 std::fs::create_dir_all(questions.root()).unwrap();
4768 let question_id = "20261014-193611-ca5b";
4769 std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
4770 assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
4771 assert_eq!(
4772 dependency(&queue, &questions, "ca5b"),
4773 Dependency::Ambiguous
4774 );
4775 assert_eq!(
4776 dependency(&queue, &questions, &dep.id),
4777 Dependency::Task(dep.id)
4778 );
4779 assert_eq!(
4780 dependency(&queue, &questions, question_id),
4781 Dependency::Question(question_id.into())
4782 );
4783 assert_eq!(
4784 dependency(&queue, &questions, "missing"),
4785 Dependency::Missing
4786 );
4787 }
4788
4789 #[test]
4790 fn dependency_resolution_directory_errors_are_unknown_not_missing() {
4791 let dir = tempfile::tempdir().unwrap();
4792 let q = Queue::at(dir.path().join("not-a-directory"));
4793 std::fs::write(q.root(), "file").unwrap();
4794 let questions = Questions::at(dir.path().join("questions"));
4795 assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
4796 assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
4797 }
4798
4799 #[test]
4800 fn a_dependency_without_a_tombstone_is_still_missing() {
4801 let (dir, q) = queue();
4802 let questions = Questions::at(dir.path().join("questions"));
4803 let mut dep = task("dependency");
4804 q.put(&mut dep).unwrap();
4805 std::fs::remove_file(q.path_of(&dep.id)).unwrap();
4806 assert_eq!(
4807 missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
4808 [dep.id.clone()]
4809 );
4810 assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
4811 }
4812
4813 #[test]
4814 fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
4815 let mut t = task("held");
4816 t.hold_machine(Some("because".to_owned()));
4817 t.blocked_by = vec!["gone".to_owned()];
4818 assert!(!t.dependency_deleted("gone"));
4819 assert_eq!(t.status, TaskStatus::Held);
4820 }
4821
4822 #[test]
4823 fn a_failed_record_removal_takes_the_tombstone_back() {
4824 let (_dir, q) = queue();
4825 let mut t = task("stays");
4826 q.put(&mut t).unwrap();
4827 q.write_tombstone(&t.id).unwrap();
4828 let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
4829 assert!(err.is_err());
4830 assert!(!q.was_deleted(&t.id));
4833 }
4834
4835 fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
4836 let p = dir.join(name);
4837 std::fs::write(&p, body).unwrap();
4838 p
4839 }
4840
4841 #[test]
4842 fn an_attachment_copy_survives_deleting_its_source() {
4843 let (dir, q) = queue();
4844 let src = source_file(dir.path(), "shot.png", "pixels");
4845 let mut t = task("with a picture");
4846 let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
4847 q.put(&mut t).unwrap();
4848 std::fs::remove_file(&src).unwrap();
4849 assert_eq!(names, ["shot.png"]);
4850 let loaded = q.get(&t.id).unwrap();
4851 let paths = q.attachment_paths(&loaded);
4852 assert_eq!(paths.len(), 1);
4853 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
4854 }
4855
4856 #[test]
4857 fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
4858 let (dir, q) = queue();
4859 let mut t = task("bad names");
4860 for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
4861 let src = source_file(dir.path(), name, "x");
4862 assert!(
4863 q.attach(&mut t, &[src]).is_err(),
4864 "`{name}` must be refused"
4865 );
4866 }
4867 assert!(!crate::ask::valid_asset_name("C:foo.png"));
4871 #[cfg(not(windows))]
4872 {
4873 let src = source_file(dir.path(), "C:foo.png", "x");
4874 assert!(q.attach(&mut t, &[src]).is_err());
4875 }
4876 let long = format!("{}.png", "a".repeat(70));
4877 let src = source_file(dir.path(), &long, "x");
4878 assert!(q.attach(&mut t, &[src]).is_err());
4879 assert!(t.attachments.is_empty());
4880 assert!(!q.attachments_dir(&t.id).exists());
4881 }
4882
4883 #[test]
4884 fn a_taken_attachment_name_is_numbered_not_overwritten() {
4885 let (dir, q) = queue();
4886 let a = source_file(dir.path(), "shot.png", "one");
4887 let sub = dir.path().join("other");
4888 std::fs::create_dir_all(&sub).unwrap();
4889 let b = source_file(&sub, "shot.png", "two");
4890 let mut t = task("collision");
4891 q.attach(&mut t, &[a]).unwrap();
4892 q.attach(&mut t, &[b]).unwrap();
4893 assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
4894 let paths = q.attachment_paths(&t);
4895 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
4896 assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
4897 }
4898
4899 #[test]
4900 fn a_renumbered_name_stays_inside_the_length_bound() {
4901 let (dir, q) = queue();
4902 let name = format!("{}.png", "a".repeat(60));
4903 assert_eq!(name.len(), 64);
4904 let a = source_file(dir.path(), &name, "one");
4905 let sub = dir.path().join("other");
4906 std::fs::create_dir_all(&sub).unwrap();
4907 let b = source_file(&sub, &name, "two");
4908 let mut t = task("long");
4909 q.attach(&mut t, &[a, b]).unwrap();
4910 assert_eq!(t.attachments.len(), 2);
4911 assert!(
4912 t.attachments
4913 .iter()
4914 .all(|n| crate::ask::valid_asset_name(n))
4915 );
4916 assert!(t.attachments[1].ends_with("-2.png"));
4917 }
4918
4919 #[test]
4920 fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
4921 let (dir, q) = queue();
4922 let good = source_file(dir.path(), "good.png", "ok");
4923 let mut t = task("partial");
4924 q.attach(&mut t, &[good]).unwrap();
4925 let more = source_file(dir.path(), "more.png", "ok");
4926 let missing = dir.path().join("missing.png");
4927 assert!(q.attach(&mut t, &[more, missing]).is_err());
4928 assert_eq!(t.attachments, ["good.png"]);
4929 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4930 .unwrap()
4931 .flatten()
4932 .collect();
4933 assert_eq!(on_disk.len(), 1);
4934 }
4935
4936 fn block_put(q: &Queue, t: &Task) -> PathBuf {
4938 let tmp = q.path_of(&t.id).with_extension("json.tmp");
4939 std::fs::create_dir_all(&tmp).unwrap();
4940 tmp
4941 }
4942
4943 #[test]
4944 fn a_failed_put_leaves_no_new_attachment_directory() {
4945 let (dir, q) = queue();
4946 let mut t = task("fresh");
4947 let tmp = block_put(&q, &t);
4948 let src = source_file(dir.path(), "shot.png", "x");
4949 assert!(q.attach_and_put(&mut t, &[src]).is_err());
4950 assert!(t.attachments.is_empty());
4951 assert!(!q.attachments_dir(&t.id).exists());
4952 assert!(!q.path_of(&t.id).exists());
4953 std::fs::remove_dir(tmp).unwrap();
4954 }
4955
4956 #[test]
4957 fn a_failed_put_removes_only_the_copy_it_just_made() {
4958 let (dir, q) = queue();
4959 let mut t = task("edited");
4960 let first = source_file(dir.path(), "first.png", "1");
4961 q.attach_and_put(&mut t, &[first]).unwrap();
4962 block_put(&q, &t);
4963 let second = source_file(dir.path(), "second.png", "2");
4964 assert!(q.attach_and_put(&mut t, &[second]).is_err());
4965 assert_eq!(t.attachments, ["first.png"]);
4966 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4967 .unwrap()
4968 .flatten()
4969 .map(|e| e.file_name().to_string_lossy().into_owned())
4970 .collect();
4971 assert_eq!(on_disk, ["first.png"]);
4972 assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
4973 }
4974
4975 #[test]
4976 fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
4977 let (dir, q) = queue();
4978 let questions = Questions::at(dir.path().join("questions"));
4979 let gone = task("gone");
4980 let mut other = task("other");
4981 let mut live = task("live");
4982 q.put(&mut other).unwrap();
4983 q.put(&mut live).unwrap();
4984 let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
4987 std::fs::create_dir_all(&orphan).unwrap();
4988 std::fs::write(orphan.join("shot.png"), "x").unwrap();
4989 let busy = q.root.join(format!("{}.attachments.removing", live.id));
4992 std::fs::create_dir_all(&busy).unwrap();
4993
4994 q.remove(&other.id, false, &questions).unwrap();
4995 assert!(!orphan.exists(), "an orphan is swept");
4996 assert!(busy.exists(), "a removal in progress is left alone");
4997 }
4998
4999 #[test]
5000 fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
5001 let (dir, q) = queue();
5002 let questions = Questions::at(dir.path().join("questions"));
5003 let mut t = task("stuck");
5004 let src = source_file(dir.path(), "shot.png", "x");
5005 q.attach_and_put(&mut t, &[src]).unwrap();
5006 let aside = q.root.join(format!("{}.attachments.removing", t.id));
5009 std::fs::create_dir_all(&aside).unwrap();
5010 std::fs::write(aside.join("old.png"), "o").unwrap();
5011 assert!(q.remove(&t.id, false, &questions).is_err());
5012 assert!(q.path_of(&t.id).exists());
5013 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5014 }
5015
5016 #[test]
5017 fn a_failed_record_removal_puts_the_attachments_back() {
5018 let (dir, q) = queue();
5019 let mut t = task("rollback");
5020 let src = source_file(dir.path(), "shot.png", "x");
5021 q.attach_and_put(&mut t, &[src]).unwrap();
5022 let err = q
5023 .remove_record_with_attachments(&t.id, |_| {
5024 Err(std::io::Error::other("injected failure"))
5025 })
5026 .unwrap_err();
5027 assert!(format!("{err:#}").contains("injected failure"));
5028 assert!(q.path_of(&t.id).exists());
5029 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5030 assert!(
5031 !q.root
5032 .join(format!("{}.attachments.removing", t.id))
5033 .exists()
5034 );
5035 }
5036
5037 #[test]
5038 fn editing_a_task_keeps_its_attachments() {
5039 let (dir, q) = queue();
5040 let src = source_file(dir.path(), "shot.png", "x");
5041 let mut t = task("editable");
5042 q.attach(&mut t, &[src]).unwrap();
5043 t.edit("new".to_owned(), "new text".to_owned()).unwrap();
5044 q.put(&mut t).unwrap();
5045 assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
5046 }
5047
5048 #[test]
5049 fn removing_a_task_deletes_its_attachments() {
5050 let (dir, q) = queue();
5051 let questions = Questions::at(dir.path().join("questions"));
5052 let src = source_file(dir.path(), "shot.png", "x");
5053 let mut t = task("doomed");
5054 q.attach(&mut t, &[src]).unwrap();
5055 q.put(&mut t).unwrap();
5056 assert!(q.attachments_dir(&t.id).is_dir());
5057 q.remove(&t.id, false, &questions).unwrap();
5058 assert!(!q.attachments_dir(&t.id).exists());
5059 assert!(q.list().is_empty());
5060 }
5061
5062 #[test]
5063 fn a_task_written_before_attachments_still_reads() {
5064 let (_dir, q) = queue();
5065 let mut t = task("old");
5066 q.put(&mut t).unwrap();
5067 let path = q.path_of(&t.id);
5068 let mut v: serde_json::Value =
5069 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
5070 v.as_object_mut().unwrap().remove("attachments");
5071 std::fs::write(&path, v.to_string()).unwrap();
5072 assert!(q.get(&t.id).unwrap().attachments.is_empty());
5073 }
5074
5075 #[test]
5076 fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
5077 let q = Queue::at(PathBuf::from("relative-queue"));
5078 let mut t = task("rel");
5079 t.attachments.push("shot.png".to_owned());
5080 let paths = q.attachment_paths(&t);
5081 assert!(paths[0].is_absolute(), "{}", paths[0].display());
5082 assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
5083 }
5084
5085 #[test]
5086 fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
5087 let dir = tempfile::tempdir().unwrap();
5088 let q = Queue::at(dir.path().join("queue"));
5089 let mut t = task("resume");
5090 t.start("A".into());
5091 t.record_pass("A", 0, 1, None);
5092 t.start("B".into());
5093 q.put(&mut t).unwrap();
5094 assert_eq!(t.runs, vec!["A", "B"]);
5095 t.start("A".into());
5096 t.record_pass("A", 1, 3, None);
5097 q.put(&mut t).unwrap();
5098 let saved = q.get(&t.id).unwrap();
5099 assert_eq!(saved.runs, vec!["B", "A"]);
5100 assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
5101 assert_eq!(saved.attempts, 3);
5102 assert_eq!(saved.passes.len(), 2);
5103 assert_eq!(saved.passes[0].run, "A");
5104 assert_eq!(saved.passes[1].run, "A");
5105 }
5106
5107 #[test]
5108 fn repeated_start_records_one_run_and_separate_passes() {
5109 let mut t = task("resume");
5110 t.start("same-run".into());
5111 t.record_pass("same-run", 0, 1, None);
5112 t.start("same-run".into());
5113 t.record_pass("same-run", 1, 2, None);
5114 assert_eq!(t.runs, vec!["same-run"]);
5115 assert_eq!(t.attempts, 2);
5116 assert_eq!(t.passes.len(), 2);
5117 }
5118
5119 #[test]
5120 fn link_run_adds_a_run_once_and_touches_nothing_else() {
5121 let dir = tempfile::tempdir().unwrap();
5122 let queue = Queue::at(dir.path().join("queue"));
5123 let mut t = Task::new(
5124 "t".to_owned(),
5125 "do it".to_owned(),
5126 PathBuf::from("."),
5127 Source::Human,
5128 );
5129 queue.put(&mut t).unwrap();
5130 let before = queue.get(&t.id).unwrap();
5131
5132 let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
5133 assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
5134 assert_eq!(linked.status, before.status);
5135 assert_eq!(linked.attempts, before.attempts);
5136 assert_eq!(linked.interrupt, before.interrupt);
5137
5138 let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5139 assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
5140 assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
5141 assert!(queue.link_run("no-such-task", "r").is_err());
5142 }
5143
5144 #[test]
5145 fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
5146 let dir = tempfile::tempdir().unwrap();
5147 let queue = Queue::at(dir.path().join("queue"));
5148 let mut t = Task::new(
5149 "t".to_owned(),
5150 "do it".to_owned(),
5151 PathBuf::from("."),
5152 Source::Human,
5153 );
5154 queue.put(&mut t).unwrap();
5155 let mut snapshot = queue.get(&t.id).unwrap();
5157 queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5158
5159 snapshot.start("20260930-000000-aaaa".to_owned());
5160 queue.put(&mut snapshot).unwrap();
5161
5162 let stored = queue.get(&t.id).unwrap();
5163 assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
5164 assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
5165 assert_eq!(stored.attempts, 1);
5166 }
5167
5168 #[test]
5169 fn concurrent_links_and_daemon_saves_lose_nothing() {
5170 let dir = tempfile::tempdir().unwrap();
5171 let queue = Queue::at(dir.path().join("queue"));
5172 let mut t = Task::new(
5173 "t".to_owned(),
5174 "do it".to_owned(),
5175 PathBuf::from("."),
5176 Source::Human,
5177 );
5178 queue.put(&mut t).unwrap();
5179 let id = t.id.clone();
5180
5181 let linkers: Vec<_> = (0..4)
5182 .map(|n| {
5183 let (queue, id) = (queue.clone(), id.clone());
5184 std::thread::spawn(move || {
5185 for k in 0..10 {
5186 queue
5187 .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
5188 .unwrap();
5189 }
5190 })
5191 })
5192 .collect();
5193 let mut mine = queue.get(&id).unwrap();
5196 for k in 0..10 {
5197 mine.start(format!("20260930-000009-d{k:03}"));
5198 queue.put(&mut mine).unwrap();
5199 }
5200 for l in linkers {
5201 l.join().unwrap();
5202 }
5203
5204 let stored = queue.get(&id).unwrap();
5205 assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
5206 assert_eq!(
5207 stored.attempts, 10,
5208 "linking never rewinds the daemon's work"
5209 );
5210 }
5211
5212 fn age_lock(path: &Path) {
5213 let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
5214 f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
5215 .unwrap();
5216 }
5217
5218 #[test]
5219 fn concurrent_stale_takeover_yields_one_holder() {
5220 use std::sync::atomic::{AtomicUsize, Ordering};
5221 use std::sync::{Arc, Barrier};
5222 for _ in 0..5 {
5223 let dir = tempfile::tempdir().unwrap();
5224 let q = Queue::at(dir.path().to_path_buf());
5225 let lock = dir.path().join("t.write-lock");
5226 std::fs::write(&lock, "dead-0000").unwrap();
5227 age_lock(&lock);
5228 let n = 6;
5229 let barrier = Arc::new(Barrier::new(n));
5230 let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
5231 let handles: Vec<_> = (0..n)
5232 .map(|_| {
5233 let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
5234 std::thread::spawn(move || {
5235 b.wait();
5236 let g = q.lock_task("t").unwrap();
5237 let held = now.fetch_add(1, Ordering::SeqCst) + 1;
5238 max.fetch_max(held, Ordering::SeqCst);
5239 std::thread::sleep(std::time::Duration::from_millis(20));
5240 now.fetch_sub(1, Ordering::SeqCst);
5241 drop(g);
5242 })
5243 })
5244 .collect();
5245 for h in handles {
5246 h.join().unwrap();
5247 }
5248 assert_eq!(max.load(Ordering::SeqCst), 1);
5249 assert!(!lock.exists());
5250 }
5251 }
5252
5253 #[test]
5254 fn dropping_a_replaced_lock_leaves_the_new_holders_token() {
5255 let dir = tempfile::tempdir().unwrap();
5256 let q = Queue::at(dir.path().to_path_buf());
5257 let lock = dir.path().join("t.write-lock");
5258 let a = q.lock_task("t").unwrap();
5259 let replacement = owner_token();
5262 std::fs::write(&lock, &replacement).unwrap();
5263 drop(a);
5264 assert_eq!(std::fs::read_to_string(&lock).unwrap(), replacement);
5265 }
5266
5267 #[test]
5268 fn aged_live_task_lock_excludes_updates_until_retirement_releases_it() {
5269 let dir = tempfile::tempdir().unwrap();
5270 let q = Queue::at(dir.path().to_path_buf());
5271 let held = q.lock_task("t").unwrap();
5272 let path = dir.path().join("t.write-lock");
5273 age_lock(&path);
5274 assert!(!break_stale(&path, &held.token));
5275 let (started_tx, started_rx) = std::sync::mpsc::channel();
5276 let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
5277 let updater = std::thread::spawn(move || {
5278 started_tx.send(()).unwrap();
5279 let _write = q.lock_task("t").unwrap();
5280 acquired_tx.send(()).unwrap();
5281 });
5282 started_rx.recv().unwrap();
5283 assert!(
5284 acquired_rx
5285 .recv_timeout(std::time::Duration::from_millis(100))
5286 .is_err()
5287 );
5288 assert_eq!(std::fs::read_to_string(&path).unwrap(), held.token);
5289 drop(held);
5290 acquired_rx
5291 .recv_timeout(std::time::Duration::from_secs(2))
5292 .unwrap();
5293 updater.join().unwrap();
5294 assert!(!path.exists());
5295 }
5296
5297 #[test]
5298 fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
5299 let dir = tempfile::tempdir().unwrap();
5300 let lock = dir.path().join("t.write-lock");
5301 std::fs::write(&lock, "new-token").unwrap();
5302 assert!(!break_stale(&lock, "old-token"));
5303 assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
5304 assert!(!break_stale(&lock, "new-token"));
5306 assert!(lock.exists());
5307 age_lock(&lock);
5308 assert!(break_stale(&lock, "new-token"));
5309 assert!(!lock.exists());
5310 }
5311
5312 #[test]
5313 fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
5314 let dir = tempfile::tempdir().unwrap();
5315 let lock = dir.path().join("t.write-lock");
5316 std::fs::write(&lock, "dead-1").unwrap();
5317 age_lock(&lock);
5318 let marker = break_marker(&lock, "dead-1");
5319 std::fs::write(&marker, "crashed-remover").unwrap();
5320 assert!(!break_stale(&lock, "dead-1"));
5322 assert!(lock.exists());
5323 age_lock(&marker);
5325 assert!(break_stale(&lock, "dead-1"));
5326 assert!(!lock.exists());
5327 assert!(!marker.exists());
5328 }
5329
5330 fn chat_task() -> Task {
5331 Task::new(
5332 "t".to_owned(),
5333 "do t".to_owned(),
5334 PathBuf::from("."),
5335 Source::Agent {
5336 run: "20260902-000000-beef".to_owned(),
5337 node: CHAT_NODE.to_owned(),
5338 },
5339 )
5340 }
5341
5342 #[test]
5343 fn a_blocked_hold_keeps_its_anchor_but_a_new_run_starts_a_new_hold() {
5344 let mut t = chat_task();
5345 t.hold_machine(Some("waiting".into()));
5346 let old: Timestamp = "2020-01-01T00:00:00Z".parse().unwrap();
5347 t.held_at = Some(old);
5348 t.block(vec!["decision".into()], None);
5349 t.hold_machine(Some("same waiting decision".into()));
5350 assert_eq!(t.held_at, Some(old));
5351 t.start("real-new-run".into());
5352 t.fail("new failure", 1);
5353 assert_ne!(
5354 t.held_at,
5355 Some(old),
5356 "old answers cannot approve a new hold"
5357 );
5358 }
5359
5360 #[test]
5361 fn legacy_delivery_survives_upgrade_and_a_no_progress_run_link() {
5362 let (_dir, queue) = queue();
5363 let mut t = chat_task();
5364 t.start("old-run".into());
5365 t.hold_machine(Some("operator is considering recovery".into()));
5366 t.chat_notice = Some(ChatNotice {
5367 key: format!("held#{}", t.held_at.unwrap()),
5368 at: Timestamp::now(),
5369 outcome: ChatNoticeOutcome::Posted,
5370 });
5371 let mut raw = serde_json::to_value(&t).unwrap();
5372 raw.as_object_mut().unwrap().remove("chat_notices");
5373 raw.as_object_mut().unwrap().remove("hold_notice_cause");
5374 std::fs::create_dir_all(queue.path_of(&t.id).parent().unwrap()).unwrap();
5375 std::fs::write(queue.path_of(&t.id), serde_json::to_string(&raw).unwrap()).unwrap();
5376 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5377 queue
5378 .modify(&t.id, |task| {
5379 task.requeue();
5380 task.link_run("unprepared-run");
5381 task.startup_notice_cause("startup-source");
5382 task.fail("same unresolved preflight", 0);
5383 true
5384 })
5385 .unwrap();
5386 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5387 queue
5388 .modify(&t.id, |task| {
5389 task.succeed();
5390 true
5391 })
5392 .unwrap();
5393 assert_eq!(
5394 queue.get(&t.id).unwrap().chat_notice_due().as_deref(),
5395 Some("done")
5396 );
5397 }
5398
5399 #[test]
5400 fn a_dependency_wait_does_not_change_when_released_and_blocked_again() {
5401 let mut t = chat_task();
5402 t.block(vec!["a".into(), "b".into()], Some("waiting".into()));
5403 let key = t.chat_notice_key().unwrap();
5404 t.chat_notices.push(key);
5405 t.release();
5406 t.block(
5407 vec!["b".into(), "a".into(), "a".into()],
5408 Some("still waiting on prerequisites".into()),
5409 );
5410 assert_eq!(t.chat_notice_due(), None);
5411 t.block(
5412 vec!["new-question".into()],
5413 Some("different decision".into()),
5414 );
5415 assert!(t.chat_notice_due().is_some());
5416 }
5417
5418 #[test]
5419 fn running_startup_failures_preserve_stage_and_report_distinct_blockers() {
5420 let mut t = chat_task();
5421 t.start("existing-run".into());
5422 t.fail("implementation failed", 1);
5423 let failed = t.chat_notice_key().unwrap();
5424 t.chat_notices.push(failed.clone());
5425 t.release_operator("test", None);
5426 t.start("existing-run".into());
5427 t.fail_startup("startup-source", "same unresolved recovery source", 1);
5428 assert_eq!(
5429 t.chat_notice_due(),
5430 None,
5431 "no-progress recovery stays quiet"
5432 );
5433 t.release_operator("test", None);
5434 t.start("existing-run".into());
5435 t.fail_startup("startup-attachments", "attachment missing", 1);
5436 let attachments = t.chat_notice_due().expect("new attachment blocker");
5437 assert_ne!(attachments, failed);
5438 assert_eq!(
5439 t.hold_notice_cause.as_ref().unwrap().stage,
5440 "startup-attachments"
5441 );
5442 t.chat_notices.push(attachments);
5443 t.release_operator("test", None);
5444 t.start("existing-run".into());
5445 t.fail_startup(
5446 "startup-attachments",
5447 "same missing attachment, reworded",
5448 1,
5449 );
5450 assert_eq!(
5451 t.chat_notice_due(),
5452 None,
5453 "same stage after attempt counter reset"
5454 );
5455 t.release_operator("test", None);
5456 t.start("existing-run".into());
5457 t.fail_startup("startup-config", "different preflight stage", 1);
5458 assert!(t.chat_notice_due().is_some());
5459 assert_eq!(
5460 t.hold_notice_cause.as_ref().unwrap().stage,
5461 "startup-config"
5462 );
5463 }
5464
5465 #[test]
5466 fn structured_causes_preserve_no_progress_retries_and_report_new_passes() {
5467 let mut t = chat_task();
5468 t.hold_machine(Some("startup source missing".into()));
5469 t.startup_notice_cause("startup-source");
5470 let first = t.chat_notice_key().unwrap();
5471 t.chat_notices.push(first.clone());
5472 t.release_operator("test", None);
5473 t.startup_notice_cause("startup-source");
5474 t.hold_machine(Some("same source cannot be read".into()));
5475 assert_eq!(t.chat_notice_due(), None);
5476 t.startup_notice_cause("startup-config");
5477 assert!(t.chat_notice_due().is_some(), "different startup stage");
5478 t.notice_cause("failed", Some("existing-run"), Some(0));
5479 let run_failure = t.chat_notice_key().unwrap();
5480 t.chat_notices.push(run_failure.clone());
5481 t.release_operator("test", None);
5482 t.startup_notice_cause("startup-source");
5483 t.hold_machine(Some("no new work happened".into()));
5484 assert_eq!(t.chat_notice_due(), None);
5485 t.notice_cause("failed", Some("existing-run"), Some(1));
5486 assert!(t.chat_notice_due().is_some(), "a real resumed pass failed");
5487 assert_ne!(t.chat_notice_key().unwrap(), run_failure);
5488 }
5489
5490 #[test]
5491 fn legacy_hold_baseline_does_not_replay_or_hide_new_failures_and_completion() {
5492 let mut t = chat_task();
5493 t.hold_machine(Some("still waiting".into()));
5494 t.chat_notice = Some(ChatNotice {
5495 key: "held#2026-10-14T00:00:00Z".into(),
5496 at: Timestamp::now(),
5497 outcome: ChatNoticeOutcome::Posted,
5498 });
5499 assert_eq!(t.chat_notice_due(), None, "old hold baseline");
5500 t.import_chat_notice();
5501 t.notice_cause("failed", Some("new-run"), Some(0));
5502 assert!(t.chat_notice_due().is_some(), "new producer evidence");
5503 t.succeed();
5504 assert_eq!(t.chat_notice_due().as_deref(), Some("done"));
5505 }
5506
5507 #[test]
5508 fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
5509 let mut t = chat_task();
5510 assert_eq!(t.chat_notice_due(), None, "queued");
5511 t.start("r1".to_owned());
5512 assert_eq!(t.chat_notice_due(), None, "running");
5513 t.fail("boom", 3);
5514 assert_eq!(t.status, TaskStatus::Failed);
5515 assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
5516 t.start("r2".to_owned());
5517 t.park("upgrade");
5518 assert_eq!(t.chat_notice_due(), None, "parked");
5519 t.fail("boom", 1);
5520 assert_eq!(t.status, TaskStatus::Held);
5521 let key = t
5522 .chat_notice_due()
5523 .expect("held on the last attempt is final");
5524 t.chat_notice = Some(ChatNotice {
5525 key,
5526 at: Timestamp::now(),
5527 outcome: ChatNoticeOutcome::Posted,
5528 });
5529 assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
5530 t.release();
5531 assert_eq!(t.chat_notice_due(), None, "requeued after a release");
5532 t.start("r3".to_owned());
5533 t.succeed();
5534 assert!(t.chat_notice_due().is_some(), "done is a new final state");
5535 }
5536
5537 #[test]
5538 fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
5539 for key in ["done", "done#0", "done#1", "done#12"] {
5540 let mut t = chat_task();
5541 t.start("r1".to_owned());
5542 t.succeed();
5543 t.chat_notice = Some(ChatNotice {
5544 key: key.to_owned(),
5545 at: Timestamp::now(),
5546 outcome: ChatNoticeOutcome::Posted,
5547 });
5548 t.requeue();
5549 t.start("r2".to_owned());
5550 t.succeed();
5551 assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
5552 assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
5553 }
5554 }
5555
5556 #[test]
5557 fn a_release_without_progress_keeps_the_unresolved_notice_cause() {
5558 let mut t = chat_task();
5559 t.hold_manual(Some("wait".to_owned()));
5560 let first = t.chat_notice_due().expect("held");
5561 t.chat_notice = Some(ChatNotice {
5562 key: first.clone(),
5563 at: Timestamp::now(),
5564 outcome: ChatNoticeOutcome::Posted,
5565 });
5566 t.hold_manual(Some("still waiting".to_owned()));
5567 assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
5568 t.release();
5569 std::thread::sleep(std::time::Duration::from_millis(5));
5570 t.hold_manual(None);
5571 assert_eq!(t.chat_notice_due(), None, "no progress, same cause");
5572 }
5573
5574 #[test]
5575 fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
5576 let mut human = task("h");
5577 human.succeed();
5578 assert_eq!(human.chat_notice_due(), None);
5579 let mut inherited = task("i");
5581 inherited.origin_chat = Some("20260902-000000-beef".to_owned());
5582 inherited.succeed();
5583 assert_eq!(inherited.chat_notice_due(), None);
5584 let mut old = chat_task();
5585 old.schema = 12;
5586 old.status = TaskStatus::Done;
5587 assert_eq!(old.chat_notice_due(), None, "history is not replayed");
5588 let mut live = chat_task();
5589 live.schema = 12;
5590 live.start("r1".to_owned());
5591 live.succeed();
5592 assert!(live.chat_notice_due().is_some(), "started by this build");
5593 let mut held = chat_task();
5594 held.schema = 12;
5595 held.hold_machine(Some("config".to_owned()));
5596 assert!(held.chat_notice_due().is_some(), "held before any start");
5597 }
5598
5599 #[test]
5600 fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
5601 let tmp = tempfile::tempdir().expect("tempdir");
5602 let q = Queue::at(tmp.path().join("queue"));
5603 let mut t = chat_task();
5604 t.succeed();
5605 q.put(&mut t).expect("put");
5606 let mut stale = q.get(&t.id).expect("get");
5607 let key = t.chat_notice_due().expect("due");
5608 q.modify(&t.id, |x| {
5609 x.chat_notice = Some(ChatNotice {
5610 key: key.clone(),
5611 at: Timestamp::now(),
5612 outcome: ChatNoticeOutcome::Posted,
5613 });
5614 true
5615 })
5616 .expect("modify");
5617 q.put(&mut stale).expect("stale put");
5618 assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
5619 }
5620}