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(path: &Path, token: &str, depth: u8) -> Option<(PathBuf, String)> {
2054 use std::io::Write;
2055 let marker = break_marker(path, token);
2056 for _ in 0..2 {
2057 match std::fs::OpenOptions::new()
2058 .write(true)
2059 .create_new(true)
2060 .open(&marker)
2061 {
2062 Ok(mut f) => {
2063 let mine = owner_token();
2064 if f.write_all(mine.as_bytes()).is_err() {
2065 drop(f);
2066 let _ = std::fs::remove_file(&marker);
2067 return None;
2068 }
2069 return Some((marker, mine));
2070 }
2071 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2072 let judged = std::fs::read_to_string(&marker).ok();
2073 match judged.filter(|_| older_than_stale(&marker)) {
2074 Some(judged) if depth < MAX_MARKER_DEPTH => {
2075 remove_lock_if(&marker, &judged, true, depth + 1)?;
2076 }
2077 _ => return None,
2078 }
2079 }
2080 Err(_) => return None,
2081 }
2082 }
2083 None
2084}
2085
2086fn remove_lock_if(path: &Path, token: &str, require_stale: bool, depth: u8) -> Option<bool> {
2091 let (marker, mine) = take_marker(path, token, depth)?;
2092 let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
2093 && (!require_stale || older_than_stale(path) && writer_may_be_gone(token));
2094 if still {
2095 let _ = std::fs::remove_file(path);
2096 }
2097 if depth >= MAX_MARKER_DEPTH {
2099 if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
2100 let _ = std::fs::remove_file(&marker);
2101 }
2102 } else {
2103 let _ = remove_lock_if(&marker, &mine, false, depth + 1);
2104 }
2105 Some(still)
2106}
2107
2108fn break_stale(path: &Path, judged: &str) -> bool {
2111 remove_lock_if(path, judged, true, 0).unwrap_or(false)
2112}
2113
2114impl Drop for TaskLock {
2115 fn drop(&mut self) {
2116 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
2117 loop {
2118 match remove_lock_if(&self.path, &self.token, false, 0) {
2119 Some(_) => return,
2120 None if std::time::Instant::now() > deadline => return,
2121 None => std::thread::sleep(std::time::Duration::from_millis(5)),
2122 }
2123 }
2124 }
2125}
2126
2127#[derive(Debug, Clone)]
2129pub struct Queue {
2130 root: PathBuf,
2131}
2132
2133impl Queue {
2134 pub fn open() -> Self {
2136 Self::at(crate::run::home().join("queue"))
2137 }
2138
2139 pub fn at(root: PathBuf) -> Self {
2142 Self { root }
2143 }
2144
2145 pub fn root(&self) -> &Path {
2147 &self.root
2148 }
2149
2150 pub fn path_of(&self, id: &str) -> PathBuf {
2152 self.root.join(format!("{id}.json"))
2153 }
2154
2155 pub fn attachments_dir(&self, id: &str) -> PathBuf {
2158 self.root.join(format!("{id}.attachments"))
2159 }
2160
2161 pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2172 let mut wanted = Vec::new();
2173 for src in sources {
2174 let name = src
2175 .file_name()
2176 .and_then(|n| n.to_str())
2177 .with_context(|| format!("`{}` has no usable file name", src.display()))?;
2178 if !crate::ask::valid_asset_name(name) {
2179 bail!(
2180 "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
2181 with no `..`; rename the file and try again"
2182 );
2183 }
2184 if !src.is_file() {
2185 bail!("attachment `{}` is not a file", src.display());
2186 }
2187 wanted.push((src, name));
2188 }
2189 let dir = self.attachments_dir(&task.id);
2190 let existed = dir.is_dir();
2191 std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
2192 let mut created: Vec<PathBuf> = Vec::new();
2193 let mut names = Vec::new();
2194 let mut copy_all = || -> Result<()> {
2195 for (src, name) in &wanted {
2196 let (stored, path) = copy_new(&dir, src, name)?;
2197 created.push(path);
2198 names.push(stored);
2199 }
2200 Ok(())
2201 };
2202 if let Err(e) = copy_all() {
2203 for path in &created {
2204 let _ = std::fs::remove_file(path);
2205 }
2206 if !existed {
2207 let _ = std::fs::remove_dir(&dir);
2208 }
2209 return Err(e);
2210 }
2211 task.attachments.extend(names.iter().cloned());
2212 Ok(names)
2213 }
2214
2215 pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2226 let dir = self.attachments_dir(&task.id);
2227 let existed = dir.is_dir();
2228 let before = task.attachments.len();
2229 let names = self.attach(task, sources)?;
2230 if let Err(e) = self.put(task) {
2231 for name in &names {
2232 let _ = std::fs::remove_file(dir.join(name));
2233 }
2234 if !existed {
2235 let _ = std::fs::remove_dir(&dir);
2236 }
2237 task.attachments.truncate(before);
2238 return Err(e);
2239 }
2240 Ok(names)
2241 }
2242
2243 pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
2247 let dir = self.attachments_dir(&task.id);
2248 task.attachments
2249 .iter()
2250 .map(|n| {
2251 let p = dir.join(n);
2252 std::path::absolute(&p).unwrap_or(p)
2253 })
2254 .collect()
2255 }
2256
2257 pub fn put(&self, task: &mut Task) -> Result<()> {
2266 let _lock = self.lock_task(&task.id)?;
2267 self.put_unlocked(task)
2268 }
2269
2270 pub fn create_new(&self, task: &mut Task) -> Result<bool> {
2275 let _lock = self.lock_task(&task.id)?;
2276 if self.path_of(&task.id).exists() {
2277 return Ok(false);
2278 }
2279 self.put_unlocked(task)?;
2280 Ok(true)
2281 }
2282
2283 fn put_unlocked(&self, task: &mut Task) -> Result<()> {
2285 if let Ok(mut stored) = read_path(&self.path_of(&task.id)) {
2286 stored.import_chat_notice();
2287 for key in stored.chat_notices {
2288 if !task.chat_notices.contains(&key) {
2289 task.chat_notices.push(key);
2290 }
2291 }
2292 for run in stored.runs {
2293 if !task.runs.contains(&run) {
2294 task.runs.push(run);
2295 }
2296 }
2297 for p in stored.passes {
2298 if !task
2299 .passes
2300 .iter()
2301 .any(|q| q.run == p.run && q.pass_index == p.pass_index)
2302 {
2303 task.passes.push(p);
2304 }
2305 }
2306 if let Some(n) = stored.chat_notice
2309 && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
2310 {
2311 task.chat_notice = Some(n);
2312 }
2313 }
2314 task.import_chat_notice();
2315 task.updated_at = Timestamp::now();
2316 std::fs::create_dir_all(&self.root)
2317 .with_context(|| format!("create {}", self.root.display()))?;
2318 let body = serde_json::to_string_pretty(task).context("serialize task")?;
2319 let path = self.path_of(&task.id);
2320 let tmp = path.with_extension("json.tmp");
2321 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
2322 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
2323 if let (Some(notice), Some(home)) = (
2328 crate::notices::task_held(task),
2329 self.root.parent().filter(|p| !p.as_os_str().is_empty()),
2330 ) {
2331 crate::notices::raise_in(home, notice);
2332 }
2333 Ok(())
2334 }
2335
2336 pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
2341 let id = self.resolve_id(id)?;
2342 let _lock = self.lock_task(&id)?;
2343 let mut task = self.get(&id)?;
2344 if !f(&mut task) {
2345 return Ok(false);
2346 }
2347 self.put_unlocked(&mut task)?;
2348 Ok(true)
2349 }
2350
2351 pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
2356 let id = self.resolve_id(id)?;
2357 let _lock = self.lock_task(&id)?;
2360 let mut task = self.get(&id)?;
2361 if task.link_run(run) {
2362 self.put_unlocked(&mut task)?;
2363 }
2364 Ok(task)
2365 }
2366
2367 pub(crate) fn lock_task(&self, id: &str) -> Result<TaskLock> {
2375 std::fs::create_dir_all(&self.root)
2376 .with_context(|| format!("create {}", self.root.display()))?;
2377 let path = self.root.join(format!("{id}.write-lock"));
2378 let started = std::time::Instant::now();
2379 loop {
2380 match std::fs::OpenOptions::new()
2381 .write(true)
2382 .create_new(true)
2383 .open(&path)
2384 {
2385 Ok(mut file) => {
2386 use std::io::Write;
2387 let token = owner_token();
2388 if let Err(e) = file.write_all(token.as_bytes()) {
2389 drop(file);
2390 let _ = std::fs::remove_file(&path);
2391 return Err(e).with_context(|| format!("lock {}", path.display()));
2392 }
2393 return Ok(TaskLock { path, token });
2394 }
2395 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2396 let judged = std::fs::read_to_string(&path).ok();
2399 let broken = judged
2400 .filter(|_| older_than_stale(&path))
2401 .is_some_and(|judged| break_stale(&path, &judged));
2402 if !broken {
2406 if started.elapsed() > TASK_LOCK_STALE {
2407 bail!("could not lock task {id}");
2408 }
2409 std::thread::sleep(std::time::Duration::from_millis(15));
2410 }
2411 }
2412 Err(e)
2416 if e.kind() == std::io::ErrorKind::PermissionDenied
2417 && started.elapsed() <= TASK_LOCK_STALE =>
2418 {
2419 std::thread::sleep(std::time::Duration::from_millis(15));
2420 }
2421 Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
2422 }
2423 }
2424 }
2425
2426 pub fn get(&self, id: &str) -> Result<Task> {
2428 let resolved = self.resolve_id(id)?;
2429 read_path(&self.path_of(&resolved))
2430 }
2431
2432 pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
2464 let resolved = self.resolve_id(id)?;
2465 if in_flight {
2466 bail!("task {resolved} is being run by a live daemon right now");
2467 }
2468 self.write_tombstone(&resolved)?;
2469 if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
2470 {
2471 if self.path_of(&resolved).exists() {
2475 let _ = std::fs::remove_file(self.tombstone_path(&resolved));
2476 }
2477 return Err(e);
2478 }
2479 let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
2480 Ok(Removal {
2481 id: resolved,
2482 released,
2483 still_blocked,
2484 })
2485 }
2486
2487 fn tombstone_path(&self, id: &str) -> PathBuf {
2490 self.root.join(format!("{id}.removed"))
2491 }
2492
2493 fn write_tombstone(&self, id: &str) -> Result<()> {
2494 let path = self.tombstone_path(id);
2495 let tmp = self.root.join(format!("{id}.removed.tmp"));
2496 std::fs::write(&tmp, b"")
2497 .and_then(|()| std::fs::rename(&tmp, &path))
2498 .with_context(|| format!("write {}", path.display()))
2499 }
2500
2501 fn was_deleted(&self, id: &str) -> bool {
2505 self.tombstone_path(id).is_file() && !self.path_of(id).exists()
2506 }
2507
2508 pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
2512 let deleted = deleted_blockers(self, questions, &task.blocked_by);
2513 deleted
2514 .into_iter()
2515 .filter(|id| task.dependency_deleted(id))
2516 .collect()
2517 }
2518
2519 pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
2523 let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
2524 return;
2525 };
2526 let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
2527 let waiting = dependent.status == TaskStatus::Blocked;
2528 let message = match (ja, waiting) {
2529 (true, false) => format!(
2530 "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
2531 dependent.short(),
2532 short(deleted)
2533 ),
2534 (true, true) => format!(
2535 "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
2536 dependent.short(),
2537 short(deleted)
2538 ),
2539 (false, false) => format!(
2540 "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
2541 dependent.short(),
2542 short(deleted)
2543 ),
2544 (false, true) => format!(
2545 "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
2546 dependent.short(),
2547 short(deleted)
2548 ),
2549 };
2550 crate::notices::raise_in(
2551 home,
2552 crate::notices::Notice::info(
2553 &format!("unblocked:{}:{}", dependent.id, deleted),
2554 message,
2555 )
2556 .link(crate::notices::Link::Task {
2557 id: dependent.id.clone(),
2558 }),
2559 );
2560 }
2561
2562 fn release_dependents_of(
2567 &self,
2568 dependency: &str,
2569 questions: &Questions,
2570 ) -> (Vec<String>, Vec<String>) {
2571 let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
2572 for listed in self.list() {
2573 if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
2574 matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
2575 }) {
2576 continue;
2577 }
2578 let Ok(_claim) = self.claim(&listed.id) else {
2579 continue;
2580 };
2581 let Ok(mut task) = self.get(&listed.id) else {
2582 continue;
2583 };
2584 let references: Vec<String> = task.blocked_by.iter().filter(|id| {
2585 matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
2586 }).cloned().collect();
2587 if references.is_empty() {
2588 continue;
2589 }
2590 for reference in references {
2591 task.dependency_deleted(&reference);
2592 }
2593 if self.put(&mut task).is_err() {
2594 continue;
2595 }
2596 self.note_dependency_deleted(&task, dependency);
2597 if task.status == TaskStatus::Blocked {
2598 still_blocked.push(task.id.clone());
2599 } else {
2600 released.push(task.id.clone());
2601 }
2602 }
2603 (released, still_blocked)
2604 }
2605
2606 fn remove_record_with_attachments(
2610 &self,
2611 resolved: &str,
2612 remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
2613 ) -> Result<()> {
2614 self.sweep_removed_attachments();
2621 let attachments = self.attachments_dir(resolved);
2622 let aside = self.root.join(format!("{resolved}.attachments.removing"));
2623 let moved = match std::fs::rename(&attachments, &aside) {
2624 Ok(()) => true,
2625 Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
2626 Err(e) => {
2627 return Err(e).with_context(|| format!("remove {}", attachments.display()));
2628 }
2629 };
2630 let path = self.path_of(resolved);
2631 if let Err(e) = remove_record(&path) {
2632 if moved {
2633 let _ = std::fs::rename(&aside, &attachments);
2634 }
2635 return Err(e).with_context(|| format!("remove {}", path.display()));
2636 }
2637 if moved {
2638 if let Err(e) = std::fs::remove_dir_all(&aside) {
2639 tracing::warn!("leftover attachments {}: {e}", aside.display());
2640 }
2641 }
2642 let lock = self.lock_path(resolved);
2643 if let Err(e) = std::fs::remove_file(&lock) {
2644 if e.kind() != std::io::ErrorKind::NotFound {
2645 return Err(e).with_context(|| format!("remove {}", lock.display()));
2646 }
2647 }
2648 Ok(())
2649 }
2650
2651 fn sweep_removed_attachments(&self) {
2657 let Ok(entries) = std::fs::read_dir(&self.root) else {
2658 return;
2659 };
2660 for entry in entries.flatten() {
2661 let name = entry.file_name();
2662 let name = name.to_string_lossy();
2663 let Some(id) = name.strip_suffix(".attachments.removing") else {
2664 continue;
2665 };
2666 if !self.path_of(id).exists() {
2669 if let Err(e) = std::fs::remove_dir_all(entry.path()) {
2670 tracing::warn!("leftover attachments {}: {e}", entry.path().display());
2671 }
2672 }
2673 }
2674 self.sweep_tombstones();
2675 }
2676
2677 fn sweep_tombstones(&self) {
2686 let Ok(entries) = std::fs::read_dir(&self.root) else {
2687 return;
2688 };
2689 let tasks = self.list();
2690 for entry in entries.flatten() {
2691 let name = entry.file_name();
2692 let name = name.to_string_lossy();
2693 let Some(id) = name.strip_suffix(".removed") else {
2694 continue;
2695 };
2696 if self.path_of(id).exists()
2697 || tasks.iter().any(|t| {
2698 t.blocked_by
2699 .iter()
2700 .any(|b| id.starts_with(b) || id.ends_with(b))
2701 })
2702 {
2703 continue;
2704 }
2705 let old = entry
2706 .metadata()
2707 .and_then(|m| m.modified())
2708 .ok()
2709 .and_then(|m| m.elapsed().ok())
2710 .is_some_and(|age| age >= TOMBSTONE_GRACE);
2711 if old {
2712 let _ = std::fs::remove_file(entry.path());
2713 }
2714 }
2715 }
2716
2717 fn lock_path(&self, id: &str) -> PathBuf {
2720 self.root.join(format!("{id}.lock"))
2721 }
2722
2723 pub fn list(&self) -> Vec<Task> {
2736 let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
2737 .into_iter()
2738 .flatten()
2739 .flatten()
2740 .map(|e| e.path())
2741 .filter(|p| p.extension().is_some_and(|x| x == "json"))
2742 .filter_map(|p| read_path(&p).ok())
2743 .collect();
2744 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
2745 tasks
2746 }
2747
2748 pub fn superseded(&self) -> HashMap<String, String> {
2761 let mut by = HashMap::new();
2762 for task in self.list() {
2763 for earlier in &task.runs {
2764 if let Some(later) = task.successor_of(earlier) {
2765 by.insert(earlier.clone(), later.clone());
2766 }
2767 }
2768 }
2769 by
2770 }
2771
2772 pub fn superseded_by(&self, run: &str) -> Option<String> {
2780 for task in self.list() {
2781 if task.runs.iter().any(|r| r == run) {
2782 return task.successor_of(run).cloned();
2783 }
2784 }
2785 None
2786 }
2787
2788 pub fn latest_attempt(&self, run: &str) -> Option<String> {
2799 for task in self.list() {
2800 if task.runs.iter().any(|r| r == run) {
2801 return task.runs.last().filter(|last| **last != run).cloned();
2802 }
2803 }
2804 None
2805 }
2806
2807 pub fn next_runnable(&self) -> Option<Task> {
2812 let mut runnable: Vec<Task> = self
2813 .list()
2814 .into_iter()
2815 .filter(|t| t.status.runnable())
2816 .collect();
2817 runnable.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
2818 runnable.into_iter().next()
2819 }
2820
2821 pub fn claim(&self, id: &str) -> Result<Claim> {
2828 std::fs::create_dir_all(&self.root)
2829 .with_context(|| format!("create {}", self.root.display()))?;
2830 let path = self.lock_path(id);
2831 match std::fs::OpenOptions::new()
2832 .write(true)
2833 .create_new(true)
2834 .open(&path)
2835 {
2836 Ok(mut f) => {
2837 use std::io::Write as _;
2838 let _ = writeln!(f, "{}", std::process::id());
2840 Ok(Claim { path })
2841 }
2842 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2843 bail!("task {id} is already claimed ({} exists)", path.display())
2844 }
2845 Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
2846 }
2847 }
2848
2849 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
2851 if !prefix.is_empty()
2852 && prefix != "."
2853 && prefix != ".."
2854 && !prefix.contains(['/', '\\'])
2855 && self.path_of(prefix).is_file()
2856 {
2857 return Ok(prefix.to_owned());
2858 }
2859 let ids = record_ids(&self.root, "json")?;
2860 match resolve_reference(prefix, ids.iter().map(String::as_str)) {
2861 Reference::Found(id) => Ok(id),
2862 Reference::Missing => bail!("no task matches `{prefix}`"),
2863 Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
2864 }
2865 }
2866
2867 pub fn revision(&self) -> u64 {
2874 self.revision_excluding(&std::collections::BTreeSet::new())
2875 }
2876
2877 pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
2880 use std::hash::{Hash as _, Hasher as _};
2881
2882 let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
2883 .into_iter()
2884 .flatten()
2885 .flatten()
2886 .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
2887 .filter(|e| {
2888 let path = e.path();
2889 !path
2890 .file_stem()
2891 .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
2892 })
2893 .filter_map(|e| {
2894 let name = e.file_name().to_string_lossy().into_owned();
2895 let mtime = e
2896 .metadata()
2897 .ok()?
2898 .modified()
2899 .ok()?
2900 .duration_since(std::time::UNIX_EPOCH)
2901 .ok()?
2902 .as_millis() as u64;
2903 Some((name, mtime))
2904 })
2905 .collect();
2906
2907 if entries.is_empty() {
2908 return 0;
2909 }
2910
2911 entries.sort_unstable();
2912 let mut hasher = std::hash::DefaultHasher::new();
2913 for (name, mtime) in &entries {
2914 name.hash(&mut hasher);
2915 mtime.hash(&mut hasher);
2916 }
2917 let h = hasher.finish();
2918 if h == 0 { 1 } else { h }
2919 }
2920}
2921
2922const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
2925
2926#[derive(Debug, Clone)]
2928pub struct Removal {
2929 pub id: String,
2931 pub released: Vec<String>,
2935 pub still_blocked: Vec<String>,
2938}
2939
2940#[derive(Debug)]
2942pub struct Claim {
2943 path: PathBuf,
2944}
2945
2946impl Drop for Claim {
2947 fn drop(&mut self) {
2948 let _ = std::fs::remove_file(&self.path);
2949 }
2950}
2951
2952pub fn title_from(instruction: &str, max: usize) -> String {
2955 let Some(line) = first_line(instruction) else {
2956 return "(empty task)".to_owned();
2957 };
2958 if line.chars().count() <= max {
2959 return line.to_owned();
2960 }
2961 let head: String = line.chars().take(max.saturating_sub(1)).collect();
2962 format!("{head}…")
2963}
2964
2965pub(crate) fn first_line(instruction: &str) -> Option<&str> {
2971 let line = instruction
2972 .lines()
2973 .map(str::trim)
2974 .find(|l| !l.is_empty())?
2975 .trim_start_matches(['#', '-', '*', '>', ' '])
2976 .trim();
2977 (!line.is_empty()).then_some(line)
2978}
2979
2980fn read_path(path: &Path) -> Result<Task> {
2981 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
2982 let task: Task =
2983 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2984 if task.schema > SCHEMA {
2990 bail!(
2991 "task {} was written by a different magi (schema {}, this build \
2992 speaks {SCHEMA})",
2993 task.id,
2994 task.schema
2995 );
2996 }
2997 Ok(task)
2998}
2999
3000#[derive(Debug, PartialEq, Eq)]
3003pub(crate) enum Reference {
3004 Found(String),
3005 Missing,
3006 Ambiguous,
3007}
3008
3009pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
3010 let mut hits = Vec::new();
3011 for candidate in ids {
3012 if candidate == id {
3013 return Reference::Found(candidate.to_owned());
3014 }
3015 if candidate.starts_with(id) || candidate.ends_with(id) {
3016 hits.push(candidate);
3017 }
3018 }
3019 match hits.as_slice() {
3020 [] => Reference::Missing,
3021 [one] => Reference::Found((*one).to_owned()),
3022 _ => Reference::Ambiguous,
3023 }
3024}
3025
3026pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
3029 let entries = match std::fs::read_dir(root) {
3030 Ok(entries) => entries,
3031 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
3032 Err(e) => return Err(e.into()),
3033 };
3034 let mut ids = Vec::new();
3035 for entry in entries {
3036 let path = entry?.path();
3037 if path.extension().is_some_and(|e| e == extension)
3038 && let Some(id) = path.file_stem().and_then(|s| s.to_str())
3039 {
3040 ids.push(id.to_owned());
3041 }
3042 }
3043 Ok(ids)
3044}
3045
3046#[derive(Debug, PartialEq, Eq)]
3049pub enum Dependency {
3050 Task(String),
3052 Question(String),
3054 Deleted(String),
3056 Missing,
3058 Ambiguous,
3060 Unknown,
3062}
3063
3064pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
3066 let (Ok(tasks), Ok(qs), Ok(removed)) = (
3067 record_ids(queue.root(), "json"),
3068 record_ids(questions.root(), "json"),
3069 record_ids(queue.root(), "removed"),
3070 ) else {
3071 return Dependency::Unknown;
3072 };
3073 let mut ids = tasks
3076 .iter()
3077 .chain(qs.iter())
3078 .chain(removed.iter())
3079 .cloned()
3080 .collect::<Vec<_>>();
3081 ids.sort();
3082 ids.dedup();
3083 match resolve_reference(id, ids.iter().map(String::as_str)) {
3084 Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
3085 Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
3086 Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
3087 Reference::Found(_) => Dependency::Unknown,
3088 Reference::Missing => Dependency::Missing,
3089 Reference::Ambiguous => Dependency::Ambiguous,
3090 }
3091}
3092
3093pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
3095 for id in ids {
3096 if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
3097 dependency(queue, questions, id)
3098 {
3099 *id = full;
3100 }
3101 }
3102}
3103
3104pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
3107 let inventory = crate::blockers::Inventory::load(queue, questions);
3108 let mut text = String::from(
3109 "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
3110 );
3111 match record_ids(queue.root(), "json") {
3112 Ok(ids) => {
3113 for id in ids {
3114 match queue.get(&id) {
3115 Ok(task) => text.push_str(&format!(
3116 "{}: {}; blocked_by={:?}; chain={:?}\n",
3117 task.id,
3118 task.status.as_str(),
3119 task.blocked_by,
3120 inventory.waits_on(&task)
3121 )),
3122 Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
3123 }
3124 }
3125 }
3126 Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
3127 }
3128 text
3129}
3130
3131pub fn missing_blockers(
3133 queue: &Queue,
3134 questions: &Questions,
3135 blocked_by: &[String],
3136) -> Vec<String> {
3137 blocked_by
3138 .iter()
3139 .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
3140 .cloned()
3141 .collect()
3142}
3143
3144pub fn deleted_blockers(
3147 queue: &Queue,
3148 questions: &Questions,
3149 blocked_by: &[String],
3150) -> Vec<String> {
3151 blocked_by
3152 .iter()
3153 .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
3154 .cloned()
3155 .collect()
3156}
3157
3158pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
3170 missing_blocker_hold_reason_in(blocked_by, missing, "en")
3171}
3172
3173pub fn missing_blocker_hold_reason_in(
3175 blocked_by: &[String],
3176 missing: &[String],
3177 language: &str,
3178) -> String {
3179 if crate::lang::is_japanese(language) {
3180 format!(
3181 "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
3182 blocked_by.join(", "),
3183 missing.join(", "),
3184 )
3185 } else {
3186 format!(
3187 "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
3188 blocked_by.join(", "),
3189 missing.join(", "),
3190 )
3191 }
3192}
3193
3194pub fn short(id: &str) -> &str {
3196 id.split('-').next_back().unwrap_or(id)
3197}
3198
3199fn new_id() -> String {
3200 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
3201 let seed = crate::rng::entropy();
3202 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
3203}
3204
3205#[cfg(test)]
3206mod tests {
3207 #[test]
3208 fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
3209 let t = Task::new(
3210 "retry".into(),
3211 "do it".into(),
3212 PathBuf::from("."),
3213 Source::Human,
3214 );
3215 let mut value = serde_json::to_value(&t).unwrap();
3216 for field in ["resumption", "resumption_history", "operator_decided_at"] {
3217 value.as_object_mut().unwrap().remove(field);
3218 }
3219 value["schema"] = serde_json::json!(15);
3220 let mut legacy: Task = serde_json::from_value(value).unwrap();
3221 legacy.hold_machine(Some("automatic retry".into()));
3222 legacy.release();
3223 assert!(legacy.resumption.is_none());
3224 assert!(legacy.resumption_history.is_empty());
3225 assert!(legacy.operator_decided_at.is_none());
3226 legacy.release_operator("cli", None);
3227 legacy.start("run".into());
3228 let reloaded: Task =
3229 serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
3230 assert!(reloaded.resume_override.is_none());
3231 assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
3232 }
3233
3234 #[test]
3235 fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
3236 let mut base = RunOverrides {
3237 merge: Some("pr".to_owned()),
3238 candidates: Some(3),
3239 ..RunOverrides::default()
3240 };
3241 base.merge_over(&RunOverrides {
3242 merge: Some("none".to_owned()),
3243 seed: Some(7),
3244 ..RunOverrides::default()
3245 });
3246 assert_eq!(base.merge.as_deref(), Some("none"));
3247 assert_eq!(base.candidates, Some(3));
3248 assert_eq!(base.seed, Some(7));
3249 }
3250
3251 #[test]
3252 fn an_already_landed_task_is_done_with_its_attempt_refunded() {
3253 let mut t = task("relanded");
3254 t.attempts = 1;
3255 t.status = TaskStatus::Running;
3256 t.already_landed("already in main as 0e368de");
3257 assert_eq!(t.status, TaskStatus::Done);
3258 assert_eq!(t.attempts, 0);
3259 assert!(t.hold_reason.is_none());
3260 assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
3261 }
3262
3263 #[test]
3264 fn missing_blocker_reason_follows_the_language() {
3265 let b = vec!["a".to_owned()];
3266 let en = missing_blocker_hold_reason_in(&b, &b, "en");
3267 assert_eq!(en, missing_blocker_hold_reason(&b, &b));
3268 assert!(en.starts_with("blocked on a"));
3269 assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
3270 assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
3271 }
3272
3273 use super::*;
3274
3275 #[test]
3276 fn triage_applied_survives_release_and_old_records_read_as_empty() {
3277 let mut t = Task::new(
3278 "t".to_owned(),
3279 "i".to_owned(),
3280 PathBuf::from("r"),
3281 Source::Human,
3282 );
3283 t.mark_triage_applied("q1");
3284 t.mark_triage_applied("q1");
3285 t.hold_machine(Some("x".to_owned()));
3286 t.release();
3287 assert_eq!(t.triage_applied, ["q1"]);
3288 assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
3289
3290 let mut v = serde_json::to_value(&t).unwrap();
3291 v.as_object_mut().unwrap().remove("triage_applied");
3292 let old: Task = serde_json::from_value(v).unwrap();
3293 assert!(old.triage_applied.is_empty());
3294 }
3295
3296 #[test]
3297 fn task_counts_of_empty_is_all_zero() {
3298 assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
3299 }
3300
3301 #[test]
3302 fn task_counts_of_tallies_every_status() {
3303 let mut queued = Task::new(
3304 "q".to_owned(),
3305 "i".to_owned(),
3306 PathBuf::from("."),
3307 Source::Human,
3308 );
3309 queued.status = TaskStatus::Queued;
3310 let mut running = queued.clone();
3311 running.status = TaskStatus::Running;
3312 let mut done = queued.clone();
3313 done.status = TaskStatus::Done;
3314 let mut failed = queued.clone();
3315 failed.status = TaskStatus::Failed;
3316 let mut held = queued.clone();
3317 held.status = TaskStatus::Held;
3318 let mut blocked = queued.clone();
3319 blocked.status = TaskStatus::Blocked;
3320
3321 let mut parked = queued.clone();
3322 parked.status = TaskStatus::Parked;
3323
3324 let counts = TaskCounts::of(&[
3325 queued,
3326 running,
3327 done.clone(),
3328 done,
3329 failed,
3330 held,
3331 blocked,
3332 parked,
3333 ]);
3334 assert_eq!(
3335 counts,
3336 TaskCounts {
3337 queued: 1,
3338 running: 1,
3339 done: 2,
3340 failed: 1,
3341 held: 1,
3342 blocked: 1,
3343 parked: 1,
3344 }
3345 );
3346 }
3347
3348 #[test]
3349 fn task_implementers_persist_normalized_without_losing_history() {
3350 let (_dir, q) = queue();
3351 let mut t = task("keep me");
3352 t.hold_manual(Some("waiting".into()));
3353 t.runs.push("earlier-run".into());
3354 t.attachments.push("reference.txt".into());
3355 let cfg: crate::config::Config =
3356 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
3357 .unwrap();
3358 q.put(&mut t).unwrap();
3359 let original = serde_json::to_value(&t).unwrap();
3360 for n in [2, 1, 2] {
3361 t.set_implementers(n, &cfg).unwrap();
3362 q.put(&mut t).unwrap();
3363 t = q.get(&t.id).unwrap();
3364 assert_eq!(t.implementers(), Some(usize::from(n)));
3365 assert_eq!(t.solo, n == 1);
3366 if n == 1 {
3367 assert!(t.overrides.is_none());
3368 }
3369 let mut actual = serde_json::to_value(&t).unwrap();
3370 for field in ["solo", "overrides", "updated_at"] {
3371 actual[field] = original[field].clone();
3372 }
3373 assert_eq!(actual, original);
3374 }
3375 t.overrides.as_mut().unwrap().judges = Some(2);
3376 t.set_implementers(1, &cfg).unwrap();
3377 assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
3378 assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
3379 }
3380
3381 #[test]
3382 fn task_implementers_old_records_keep_inheritance_until_changed() {
3383 let mut value = serde_json::to_value(task("old task")).unwrap();
3384 value.as_object_mut().unwrap().remove("solo");
3385 value.as_object_mut().unwrap().remove("overrides");
3386 let mut old: Task = serde_json::from_value(value).unwrap();
3387 assert_eq!(old.implementers(), None);
3388 old.set_implementers(1, &crate::config::Config::default())
3389 .unwrap();
3390 assert!(old.solo);
3391 assert!(old.overrides.is_none());
3392 }
3393
3394 #[test]
3395 fn task_implementers_reject_invalid_counts_and_states() {
3396 let cfg = crate::config::Config::default();
3397 let mut t = task("unchanged");
3398 for n in [0, 4, 255] {
3399 assert!(t.set_implementers(n, &cfg).is_err());
3400 }
3401 for status in [
3402 TaskStatus::Running,
3403 TaskStatus::Done,
3404 TaskStatus::Failed,
3405 TaskStatus::Blocked,
3406 TaskStatus::Parked,
3407 ] {
3408 t.status = status;
3409 let before = serde_json::to_value(&t).unwrap();
3410 assert!(t.set_implementers(1, &cfg).is_err());
3411 assert_eq!(serde_json::to_value(&t).unwrap(), before);
3412 }
3413 }
3414
3415 fn queue() -> (tempfile::TempDir, Queue) {
3418 let dir = tempfile::tempdir().unwrap();
3419 let q = Queue::at(dir.path().join("queue"));
3420 (dir, q)
3421 }
3422
3423 #[test]
3424 fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
3425 let dir = tempfile::tempdir().unwrap();
3426 let q = Queue::at(dir.path().join("queue"));
3427 let mut t = task("held");
3428 q.put(&mut t).unwrap();
3429 assert_eq!(
3430 crate::notices::Notices::at(dir.path().join("notifications"))
3431 .list()
3432 .len(),
3433 0
3434 );
3435 t.hold_machine(Some("out of attempts".to_owned()));
3436 q.put(&mut t).unwrap();
3437 let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
3438 assert_eq!(listed.len(), 1);
3439 assert!(listed[0].message.contains("out of attempts"));
3440 }
3441
3442 fn task(title: &str) -> Task {
3443 Task::new(
3444 title.to_owned(),
3445 format!("do {title}"),
3446 PathBuf::from("."),
3447 Source::Human,
3448 )
3449 }
3450
3451 #[test]
3452 fn attempt_limit_inherits_old_records_and_controls_failure() {
3453 let mut t = task("budget");
3454 let mut old = serde_json::to_value(&t).unwrap();
3455 old.as_object_mut().unwrap().remove("attempt_limit");
3456 old["schema"] = serde_json::json!(15);
3457 let legacy: Task = serde_json::from_value(old).unwrap();
3458 assert_eq!(legacy.attempt_limit, None);
3459 assert_eq!(legacy.effective_attempt_limit(2), 2);
3460 t.attempts = 2;
3461 t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
3462 t.fail("retry", 1);
3463 assert_eq!(t.status, TaskStatus::Failed);
3464 t.attempts = 3;
3465 t.fail("stop", 10);
3466 assert_eq!(t.status, TaskStatus::Held);
3467 t.change_attempt_limit(None).unwrap();
3468 assert_eq!(t.status, TaskStatus::Held);
3469 assert_eq!(t.attempts, 3);
3470 t.fail("inherit", 4);
3471 assert_eq!(t.status, TaskStatus::Failed);
3472 for status in [
3473 TaskStatus::Running,
3474 TaskStatus::Done,
3475 TaskStatus::Failed,
3476 TaskStatus::Blocked,
3477 TaskStatus::Parked,
3478 ] {
3479 t.status = status;
3480 assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
3481 assert_eq!(t.attempt_limit, None);
3482 }
3483 }
3484
3485 #[test]
3486 fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
3487 let mut t = task("retried");
3488 assert!(
3489 t.earlier_attempts().is_empty(),
3490 "a first attempt takes nothing over"
3491 );
3492 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3493 assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
3494 assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
3496 assert_eq!(t.successor_of("bbbb"), None);
3497 assert_eq!(t.successor_of("zzzz"), None);
3498 }
3499
3500 #[test]
3501 fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
3502 let (_dir, q) = queue();
3503 let mut t = task("resumed");
3504 for (runs, successor) in [
3505 (vec!["aaaa", "aaaa"], None),
3506 (vec!["aaaa", "bbbb", "aaaa"], None),
3507 (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
3508 (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
3509 ] {
3510 t.runs = runs.into_iter().map(str::to_owned).collect();
3511 q.put(&mut t).unwrap();
3512
3513 assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
3514 assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
3515 assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
3516 }
3517 }
3518
3519 #[test]
3520 fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
3521 let (_dir, q) = queue();
3522 let mut t = task("retried");
3523 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3524 q.put(&mut t).unwrap();
3525
3526 assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
3527 assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
3528 assert_eq!(
3529 q.superseded_by("cccc"),
3530 None,
3531 "the latest attempt replaces nothing"
3532 );
3533 assert_eq!(
3534 q.superseded_by("never-heard-of-it"),
3535 None,
3536 "a run belonging to no task on this queue is not superseded"
3537 );
3538
3539 let mut by = HashMap::new();
3540 by.insert("aaaa".to_owned(), "bbbb".to_owned());
3541 by.insert("bbbb".to_owned(), "cccc".to_owned());
3542 assert_eq!(
3543 q.superseded(),
3544 by,
3545 "the whole-map and single-run forms must agree"
3546 );
3547 }
3548
3549 #[test]
3550 fn superseded_attempts_is_empty_until_the_task_is_done() {
3551 let mut t = task("retried");
3552 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3553 t.status = TaskStatus::Failed;
3554 assert_eq!(
3555 t.superseded_attempts(true),
3556 &[] as &[String],
3557 "a task still retrying has no attempt yet that a later one made moot"
3558 );
3559
3560 t.status = TaskStatus::Running;
3561 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3562 }
3563
3564 #[test]
3565 fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
3566 let mut t = task("retried");
3567 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3568 t.status = TaskStatus::Done;
3569 assert_eq!(
3570 t.superseded_attempts(true),
3571 &["aaaa".to_owned(), "bbbb".to_owned()],
3572 "cccc is the attempt whose success made the task done, and stays out"
3573 );
3574 }
3575
3576 #[test]
3577 fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
3578 let mut t = task("first try landed");
3579 t.runs = vec!["aaaa".to_owned()];
3580 t.status = TaskStatus::Done;
3581 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3582 }
3583
3584 #[test]
3585 fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
3586 let mut t = task("closed by hand after a manual merge");
3593 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3594 t.status = TaskStatus::Done;
3595 assert_eq!(
3596 t.superseded_attempts(false),
3597 &[] as &[String],
3598 "nothing here is provably why the task is done, so nothing is superseded"
3599 );
3600 }
3601
3602 #[test]
3603 fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
3604 let (_dir, q) = queue();
3605 let mut t = task("retried twice");
3606 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3607 q.put(&mut t).unwrap();
3608
3609 assert_eq!(
3610 q.latest_attempt("aaaa"),
3611 Some("cccc".to_owned()),
3612 "an old attempt points straight at the chain's current head, not the \
3613 next attempt in the middle of it"
3614 );
3615 assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
3616 assert_eq!(
3617 q.latest_attempt("cccc"),
3618 None,
3619 "the latest attempt is not superseded by anything"
3620 );
3621 assert_eq!(
3622 q.latest_attempt("never-heard-of-it"),
3623 None,
3624 "a run belonging to no task on this queue is not superseded"
3625 );
3626 }
3627
3628 #[test]
3629 fn a_markdown_heading_is_the_title_not_decoration() {
3630 assert_eq!(
3635 title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
3636 "Rework the config loader"
3637 );
3638 assert_eq!(title_from("- fix the thing", 40), "fix the thing");
3639 assert_eq!(title_from("> quoted task", 40), "quoted task");
3640 assert_eq!(title_from(" \n\n", 40), "(empty task)");
3642 assert_eq!(title_from("###\n", 40), "(empty task)");
3643 }
3644
3645 #[test]
3646 fn a_long_title_is_elided_by_characters_not_bytes() {
3647 let long = "課題".repeat(30);
3649 let title = title_from(&long, 10);
3650 assert_eq!(title.chars().count(), 10);
3651 assert!(title.ends_with('…'));
3652 }
3653
3654 #[test]
3655 fn priority_wins_and_ties_break_oldest_first() {
3656 let (_dir, q) = queue();
3657 let mut a = task("first");
3658 let mut b = task("second");
3659 let mut c = task("urgent");
3660 a.id = "20260101-000001-aaaa".to_owned();
3662 b.id = "20260101-000002-bbbb".to_owned();
3663 c.id = "20260101-000003-cccc".to_owned();
3664 c.priority = 5;
3665 for t in [&mut a, &mut b, &mut c] {
3666 q.put(t).unwrap();
3667 }
3668
3669 assert_eq!(q.next_runnable().unwrap().id, c.id);
3671 c.hold_machine(None);
3672 q.put(&mut c).unwrap();
3673 assert_eq!(q.next_runnable().unwrap().id, a.id);
3675 assert_eq!(q.list().len(), 3, "b is still waiting its turn");
3676 }
3677
3678 #[test]
3679 fn a_blocked_task_never_starves_another_runnable_one() {
3680 let (_dir, q) = queue();
3681 let mut blocked = task("blocked");
3682 blocked.block(vec!["something".to_owned()], None);
3683 q.put(&mut blocked).unwrap();
3684
3685 let mut runnable = task("free to go");
3686 q.put(&mut runnable).unwrap();
3687
3688 let next = q.next_runnable().expect("a runnable task is still offered");
3689 assert_eq!(next.id, runnable.id);
3690 }
3691
3692 #[test]
3693 fn a_held_task_is_never_offered_to_the_loop() {
3694 let (_dir, q) = queue();
3695 let mut t = task("held");
3696 q.put(&mut t).unwrap();
3697 assert!(q.next_runnable().is_some());
3698
3699 t.hold_machine(None);
3700 q.put(&mut t).unwrap();
3701 assert!(
3702 q.next_runnable().is_none(),
3703 "a held task must wait for a human"
3704 );
3705
3706 t.status = TaskStatus::Failed;
3708 q.put(&mut t).unwrap();
3709 assert!(q.next_runnable().is_some());
3710 }
3711
3712 #[test]
3713 fn attempts_are_capped_and_then_the_task_is_held() {
3714 let mut t = task("doomed");
3715
3716 t.start("run-1".to_owned());
3717 t.fail("gate red", 2);
3718 assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
3719
3720 t.start("run-2".to_owned());
3721 t.fail("gate red", 2);
3722 assert_eq!(
3723 t.status,
3724 TaskStatus::Held,
3725 "out of attempts: stop spending money on it"
3726 );
3727 assert_eq!(t.runs, ["run-1", "run-2"]);
3728 assert_eq!(t.last_error.as_deref(), Some("gate red"));
3729 assert_eq!(
3730 t.hold_reason.as_deref(),
3731 Some("gate red"),
3732 "the hold must say why, not leave hold_reason null next to a \
3733 populated last_error"
3734 );
3735 }
3736
3737 #[test]
3738 fn handing_off_a_task_records_a_hold_reason_too() {
3739 let mut t = task("left a pull request");
3740 t.start("run-1".to_owned());
3741 t.handed_off("run ended with a pull request open [run run-1]");
3742 assert_eq!(t.status, TaskStatus::Held);
3743 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3744 assert_eq!(
3745 t.hold_reason.as_deref(),
3746 Some("run ended with a pull request open [run run-1]")
3747 );
3748 assert_eq!(t.hold_reason, t.last_error);
3749 }
3750
3751 #[test]
3752 fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
3753 let mut t = task("stalled by quota");
3754
3755 t.start("run-1".to_owned());
3756 assert_eq!(t.attempts, 1);
3757 t.stall("judge-1, judge-2 out of quota");
3758 assert_eq!(
3759 t.attempts, 0,
3760 "a closed quota window must not spend the task's retry budget"
3761 );
3762 assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
3763 assert_eq!(
3764 t.last_error.as_deref(),
3765 Some("judge-1, judge-2 out of quota")
3766 );
3767
3768 for _ in 0..20 {
3771 t.start("run-n".to_owned());
3772 t.stall("still out of quota");
3773 }
3774 t.start("run-real".to_owned());
3775 t.fail("gate red", 2);
3776 assert_eq!(
3777 t.status,
3778 TaskStatus::Failed,
3779 "the first attempt that was really judged is attempt one"
3780 );
3781 }
3782
3783 #[test]
3784 fn releasing_a_held_task_gives_it_a_real_second_chance() {
3785 let mut t = task("retry me");
3786 t.start("run-1".to_owned());
3787 t.fail("gate red", 1);
3788 assert_eq!(t.status, TaskStatus::Held);
3789
3790 t.release();
3791 assert_eq!(t.status, TaskStatus::Queued);
3792 assert_eq!(t.attempts, 0);
3795 assert!(t.last_error.is_none());
3796 assert_eq!(
3797 t.runs.len(),
3798 1,
3799 "history is kept: attempts reset, evidence does not"
3800 );
3801 }
3802
3803 #[test]
3804 fn a_hold_reason_survives_and_a_release_clears_it() {
3805 let mut t = task("waiting on something else");
3806 t.hold_manual(Some(
3807 "waiting for 20260101-000000-aaaa to land first".to_owned(),
3808 ));
3809 assert_eq!(t.status, TaskStatus::Held);
3810 assert_eq!(
3811 t.hold_reason.as_deref(),
3812 Some("waiting for 20260101-000000-aaaa to land first")
3813 );
3814
3815 t.hold_manual(None);
3817 assert_eq!(
3818 t.hold_reason.as_deref(),
3819 Some("waiting for 20260101-000000-aaaa to land first"),
3820 "a bare re-hold keeps whatever a human already wrote down"
3821 );
3822
3823 let mut plain = task("no reason given");
3825 plain.hold_manual(None);
3826 assert_eq!(plain.status, TaskStatus::Held);
3827 assert!(plain.hold_reason.is_none());
3828
3829 t.release();
3830 assert_eq!(t.status, TaskStatus::Queued);
3831 assert!(
3832 t.hold_reason.is_none(),
3833 "a stale reason must not greet the next person who holds this task"
3834 );
3835 }
3836
3837 #[test]
3838 fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
3839 let mut t = task("landed by hand while held");
3844 t.hold_manual(Some("waiting on 3ed9".to_owned()));
3845 assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
3846
3847 t.succeed();
3848 assert_eq!(t.status, TaskStatus::Done);
3849 assert!(
3850 t.hold_reason.is_none(),
3851 "a done task cannot still be waiting on something"
3852 );
3853 }
3854
3855 #[test]
3856 fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
3857 let mut held = task("held straight out of blocked");
3864 held.block(
3865 vec!["20260101-000000-dead".to_owned()],
3866 Some("waiting on the migration script".to_owned()),
3867 );
3868 assert_eq!(held.status, TaskStatus::Blocked);
3869
3870 held.hold_manual(None);
3871 assert_eq!(held.status, TaskStatus::Held);
3872 assert!(
3873 held.blocked_by.is_empty(),
3874 "hold overrides the wait, same as release"
3875 );
3876 assert!(held.block_reason.is_none());
3877
3878 let mut done = task("closed straight out of blocked");
3879 done.block(
3880 vec!["20260101-000000-dead".to_owned()],
3881 Some("waiting on the migration script".to_owned()),
3882 );
3883 done.succeed();
3884 assert_eq!(done.status, TaskStatus::Done);
3885 assert!(
3886 done.blocked_by.is_empty(),
3887 "a done task cannot still be waiting on a dependency"
3888 );
3889 assert!(done.block_reason.is_none());
3890 }
3891
3892 #[test]
3893 fn a_blocked_task_is_never_offered_to_the_loop() {
3894 let mut t = task("blocked");
3895 assert!(t.status.runnable());
3896 t.block(
3897 vec!["dep-id".to_owned()],
3898 Some("waits on dep-id".to_owned()),
3899 );
3900 assert_eq!(t.status, TaskStatus::Blocked);
3901 assert!(!t.status.runnable());
3902 assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
3903 }
3904
3905 #[test]
3906 fn unblocking_the_last_dependency_returns_the_task_to_queued() {
3907 let mut t = task("blocked on two");
3908 t.block(
3909 vec!["a".to_owned(), "b".to_owned()],
3910 Some("waits on a and b".to_owned()),
3911 );
3912
3913 t.unblock("a");
3914 assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
3915 assert_eq!(t.blocked_by, ["b"]);
3916
3917 t.unblock("b");
3918 assert_eq!(t.status, TaskStatus::Queued);
3919 assert!(t.blocked_by.is_empty());
3920 assert!(t.block_reason.is_none());
3921 }
3922
3923 #[test]
3924 fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
3925 let mut t = task("never blocked");
3926 t.unblock("whatever");
3927 assert_eq!(t.status, TaskStatus::Queued);
3928 }
3929
3930 #[test]
3931 fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
3932 let mut t = task("held, then asked about");
3937 t.hold_machine(Some("out of attempts".to_owned()));
3938 assert_eq!(t.status, TaskStatus::Held);
3939
3940 t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
3941 assert_eq!(t.status, TaskStatus::Blocked);
3942
3943 t.record_answer("what now?".to_owned(), "leave it held".to_owned());
3944 t.unblock("q1");
3945 assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
3946 assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
3947 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3948 assert!(t.blocked_from.is_none(), "consumed once restored");
3949 }
3950
3951 #[test]
3952 fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
3953 let mut t = task("manually held, then asked about");
3954 t.hold_manual(Some("waiting on a dependency".to_owned()));
3955
3956 t.block(vec!["q1".to_owned()], None);
3957 t.unblock("q1");
3958
3959 assert_eq!(t.status, TaskStatus::Held);
3960 assert_eq!(t.hold_source, Some(HoldSource::Manual));
3961 }
3962
3963 #[test]
3964 fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
3965 let mut t = task("held, blocked twice");
3969 t.hold_machine(None);
3970 t.block(vec!["q1".to_owned()], Some("first".to_owned()));
3971 t.block(
3972 vec!["q1".to_owned(), "q2".to_owned()],
3973 Some("second".to_owned()),
3974 );
3975
3976 t.unblock("q1");
3977 assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
3978 t.unblock("q2");
3979 assert_eq!(t.status, TaskStatus::Held);
3980 }
3981
3982 #[test]
3983 fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
3984 let mut t = task("blocked mid-run");
3988 t.start("run-1".to_owned());
3989 assert_eq!(t.status, TaskStatus::Running);
3990
3991 t.block(vec!["q1".to_owned()], None);
3992 t.unblock("q1");
3993 assert_eq!(t.status, TaskStatus::Queued);
3994 }
3995
3996 #[test]
3997 fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
3998 let mut t = task("legacy record, held before it was blocked");
4004 t.hold_source = Some(HoldSource::Machine);
4005 t.hold_reason = Some("legacy hold reason".to_owned());
4006 t.status = TaskStatus::Blocked;
4007 t.blocked_by = vec!["q1".to_owned()];
4008 t.blocked_from = None;
4009
4010 t.unblock("q1");
4011 assert_eq!(t.status, TaskStatus::Held);
4012 }
4013
4014 #[test]
4015 fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
4016 let mut t = task("legacy record, ordinary dependency block");
4017 t.status = TaskStatus::Blocked;
4018 t.blocked_by = vec!["dep".to_owned()];
4019 t.blocked_from = None;
4020
4021 t.unblock("dep");
4022 assert_eq!(t.status, TaskStatus::Queued);
4023 }
4024
4025 #[test]
4026 fn answering_a_question_is_recorded_and_survives_a_release() {
4027 let mut t = task("asked something");
4028 t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
4029 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
4030 t.unblock("q1");
4031 assert_eq!(t.status, TaskStatus::Queued);
4032 assert_eq!(t.answers.len(), 1);
4033 assert_eq!(t.answers[0].answer, "SQLite");
4034
4035 t.release();
4039 assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
4040 }
4041
4042 #[test]
4043 fn resumption_review_branch_requires_unchanged_authored_criteria() {
4044 for source in ["cli", "web"] {
4045 let mut t = task("review old candidate");
4046 t.request_review("candidate-A".into());
4047 t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
4048 let mut unchanged = t.clone();
4049 unchanged.release_operator(source, None);
4050 assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
4051 t.edit(
4052 "review approved head".into(),
4053 "Reuse measured PR head B; fix Clippy; review".into(),
4054 )
4055 .unwrap();
4056 t.release_operator(source, None);
4057 assert!(t.review_branch.is_none());
4058 assert!(t.review_instruction.is_none());
4059 let mut legacy = unchanged;
4061 legacy.review_instruction = None;
4062 legacy.hold_for_handover(None, "still checked out".into());
4063 legacy
4064 .edit("new plan".into(), "Reuse measured PR head B".into())
4065 .unwrap();
4066 legacy.release_operator(source, None);
4067 assert!(legacy.review_branch.is_none());
4068 }
4069 }
4070
4071 #[test]
4072 fn a_refused_handover_keeps_the_review_branch_across_release() {
4073 let mut t = task("refused takeover");
4074 t.start("run-1".to_owned());
4075 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
4076 assert_eq!(t.status, TaskStatus::Held);
4077 assert_eq!(t.attempts, 1);
4078 t.release();
4079 assert_eq!(t.status, TaskStatus::Queued);
4080 assert_eq!(t.attempts, 0);
4081 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4082
4083 t.hold_for_handover(None, "again".to_owned());
4085 t.requeue();
4086 assert!(t.review_branch.is_none());
4087
4088 let mut m = task("manual");
4090 m.review_branch = Some("magi/x/A".to_owned());
4091 m.hold_manual(None);
4092 m.release();
4093 assert!(m.review_branch.is_none());
4094 }
4095
4096 #[test]
4097 fn requesting_review_requeues_the_task_and_remembers_the_branch() {
4098 let mut t = task("blocked run with a surviving branch");
4099 t.start("run-1".to_owned());
4100 t.fail("blocked with major findings", 5);
4101 assert_eq!(t.status, TaskStatus::Failed);
4102
4103 t.request_review("magi/eba2/A".to_owned());
4104 assert_eq!(t.status, TaskStatus::Queued);
4105 assert_eq!(t.attempts, 0);
4106 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4107
4108 t.release();
4110 assert!(t.review_branch.is_none());
4111 }
4112
4113 #[test]
4114 fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
4115 let mut t = task("retry");
4116 t.start("run-1".to_owned());
4117 t.requeue();
4118 assert!(t.fresh_start);
4119
4120 t.release();
4121 assert!(!t.fresh_start);
4122 }
4123
4124 #[test]
4125 fn urgent_only_changes_queued_or_held_and_preserves_settings() {
4126 for status in [
4127 TaskStatus::Queued,
4128 TaskStatus::Held,
4129 TaskStatus::Running,
4130 TaskStatus::Done,
4131 TaskStatus::Failed,
4132 TaskStatus::Blocked,
4133 TaskStatus::Parked,
4134 ] {
4135 let mut t = task("urgent settings");
4136 t.status = status;
4137 t.attempts = 2;
4138 t.hold_reason = Some("waiting".into());
4139 t.priority = 7;
4140 t.interrupt = true;
4141 let before = serde_json::to_value(&t).unwrap();
4142 for urgent in [true, true, false, false] {
4143 let result = t.set_urgent(urgent);
4144 let mut expected = before.clone();
4145 if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
4146 result.unwrap();
4147 expected["urgent"] = urgent.into();
4148 } else {
4149 let error = result.unwrap_err().to_string();
4150 assert!(error.contains(status.as_str()), "{error}");
4151 assert!(error.contains("queued or held"), "{error}");
4152 }
4153 assert_eq!(serde_json::to_value(&t).unwrap(), expected);
4154 }
4155 }
4156 }
4157
4158 #[test]
4159 fn priority_can_be_changed_while_queued_but_not_while_running() {
4160 let mut t = task("reprioritise me");
4161 t.set_priority(5).unwrap();
4162 assert_eq!(t.priority, 5);
4163
4164 t.start("run-1".to_owned());
4165 let err = t.set_priority(9).unwrap_err().to_string();
4166 assert!(err.contains("running"), "{err}");
4167 assert_eq!(t.priority, 5, "the rejected write must not partially apply");
4168 }
4169
4170 #[test]
4171 fn interrupt_can_be_marked_while_queued_but_not_while_running() {
4172 let mut t = task("interrupt me");
4173 assert!(!t.interrupt, "off unless asked, same as any other task");
4174
4175 t.set_interrupt(true).unwrap();
4176 assert!(t.interrupt);
4177
4178 t.start("run-1".to_owned());
4179 assert!(
4180 !t.interrupt,
4181 "the mark is one-shot: dispatching the task fulfils it, \
4182 whatever the run that follows ends up doing"
4183 );
4184 let err = t.set_interrupt(true).unwrap_err().to_string();
4185 assert!(err.contains("running"), "{err}");
4186 t.set_interrupt(false).unwrap();
4189 assert!(!t.interrupt);
4190 }
4191
4192 #[test]
4196 fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
4197 let mut t = task("interrupt me");
4198 t.set_interrupt(true).unwrap();
4199 t.start("run-1".to_owned());
4200 t.fail("mock failure", 5);
4201 assert_eq!(t.status, TaskStatus::Failed);
4202 assert!(
4203 !t.interrupt,
4204 "one attempt already spent the mark; a retry is an ordinary \
4205 requeue, not a fresh interrupt request"
4206 );
4207 }
4208
4209 #[test]
4210 fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
4211 let (_dir, q) = queue();
4212 let mut a = task("first filed");
4213 let mut b = task("second filed");
4214 a.id = "20260101-000001-aaaa".to_owned();
4215 b.id = "20260101-000002-bbbb".to_owned();
4216 q.put(&mut a).unwrap();
4217 q.put(&mut b).unwrap();
4218
4219 assert_eq!(
4220 q.next_runnable().unwrap().id,
4221 a.id,
4222 "with equal priority the older task goes first, so a burst of \
4223 new work cannot starve it"
4224 );
4225 assert_eq!(
4226 q.list()[0].id,
4227 b.id,
4228 "but the list an operator reads is newest first, the same as \
4229 before priority existed - a's turn to run does not make it the \
4230 newest task"
4231 );
4232
4233 let mut a = q.get(&a.id).unwrap();
4234 a.set_priority(10).unwrap();
4235 q.put(&mut a).unwrap();
4236
4237 assert_eq!(
4238 q.next_runnable().unwrap().id,
4239 a.id,
4240 "a raised priority must be reflected the moment it is saved"
4241 );
4242 assert_eq!(
4246 q.list()[0].id,
4247 a.id,
4248 "the raised task must sort first in the list an operator reads, \
4249 not only in next_runnable's own ordering"
4250 );
4251 }
4252
4253 #[test]
4254 fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
4255 let mut t = Task::new(
4256 "old title".to_owned(),
4257 "old instruction".to_owned(),
4258 PathBuf::from("/repo"),
4259 Source::Agent {
4260 run: "20260101-000000-beef".to_owned(),
4261 node: "implement".to_owned(),
4262 },
4263 );
4264 let id = t.id.clone();
4265 let created_at = t.created_at;
4266 t.runs.push("20260101-000000-beef".to_owned());
4267
4268 t.edit("new title".to_owned(), "new instruction".to_owned())
4269 .unwrap();
4270
4271 assert_eq!(t.title, "new title");
4272 assert_eq!(t.instruction, "new instruction");
4273 assert_eq!(t.id, id, "editing must not mint a new id");
4274 assert_eq!(t.created_at, created_at);
4275 assert_eq!(
4276 t.source,
4277 Source::Agent {
4278 run: "20260101-000000-beef".to_owned(),
4279 node: "implement".to_owned(),
4280 },
4281 "editing must not turn agent attribution into human"
4282 );
4283 assert_eq!(t.runs, ["20260101-000000-beef"]);
4284 }
4285
4286 #[test]
4287 fn editing_is_refused_once_a_task_is_running_or_finished() {
4288 let mut running = task("in flight");
4289 running.start("run-1".to_owned());
4290 let err = running
4291 .edit("x".to_owned(), "y".to_owned())
4292 .unwrap_err()
4293 .to_string();
4294 assert!(err.contains("running"), "{err}");
4295
4296 let mut done = task("finished");
4297 done.succeed();
4298 let err = done
4299 .edit("x".to_owned(), "y".to_owned())
4300 .unwrap_err()
4301 .to_string();
4302 assert!(err.contains("done"), "{err}");
4303
4304 let mut queued = task("waiting");
4306 queued.edit("x".to_owned(), "y".to_owned()).unwrap();
4307 let mut held = task("parked");
4308 held.hold_machine(None);
4309 held.edit("x".to_owned(), "y".to_owned()).unwrap();
4310 }
4311
4312 #[test]
4313 fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
4314 let (_dir, q) = queue();
4315 let path = q.path_of("20260101-000000-aaaa");
4316 std::fs::create_dir_all(q.root()).unwrap();
4317 std::fs::write(
4318 &path,
4319 serde_json::json!({
4320 "schema": SCHEMA,
4321 "id": "20260101-000000-aaaa",
4322 "title": "from before hold reasons existed",
4323 "instruction": "from before hold reasons existed",
4324 "repo": ".",
4325 "source": { "kind": "human" },
4326 "status": "held",
4327 "created_at": Timestamp::now().to_string(),
4328 "updated_at": Timestamp::now().to_string(),
4329 })
4330 .to_string(),
4331 )
4332 .unwrap();
4333
4334 let task = q.get("20260101-000000-aaaa").expect("must still read");
4335 assert!(task.hold_reason.is_none());
4336 assert!(task.operator_held());
4337 }
4338
4339 #[test]
4340 fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
4341 let (_dir, q) = queue();
4342 let path = q.path_of("20260101-000000-bbbb");
4343 std::fs::create_dir_all(q.root()).unwrap();
4344 std::fs::write(
4345 &path,
4346 serde_json::json!({
4347 "schema": 2,
4348 "id": "20260101-000000-bbbb",
4349 "title": "old manual recovery",
4350 "instruction": "old manual recovery",
4351 "repo": ".",
4352 "source": { "kind": "human" },
4353 "status": "held",
4354 "hold_reason": "active manual recovery run20260912-224242-daf5",
4355 "created_at": Timestamp::now().to_string(),
4356 "updated_at": Timestamp::now().to_string(),
4357 })
4358 .to_string(),
4359 )
4360 .unwrap();
4361
4362 let task = q.get("20260101-000000-bbbb").expect("must still read");
4363 assert_eq!(task.hold_source, None);
4364 assert!(task.operator_held());
4365 }
4366
4367 #[test]
4368 fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
4369 let (_dir, q) = queue();
4370 let path = q.path_of("20260101-000000-aaaa");
4371 std::fs::create_dir_all(q.root()).unwrap();
4372 std::fs::write(
4373 &path,
4374 serde_json::json!({
4375 "schema": SCHEMA,
4376 "id": "20260101-000000-aaaa",
4377 "title": "from before diagnostics existed",
4378 "instruction": "from before diagnostics existed",
4379 "repo": ".",
4380 "source": { "kind": "human" },
4381 "status": "held",
4382 "created_at": Timestamp::now().to_string(),
4383 "updated_at": Timestamp::now().to_string(),
4384 })
4385 .to_string(),
4386 )
4387 .unwrap();
4388
4389 let task = q.get("20260101-000000-aaaa").expect("must still read");
4390 assert!(task.diagnostic.is_none());
4391 }
4392
4393 #[test]
4394 fn a_schema_1_task_with_no_blocking_fields_still_reads() {
4395 let (_dir, q) = queue();
4399 let path = q.path_of("20260101-000000-aaaa");
4400 std::fs::create_dir_all(q.root()).unwrap();
4401 std::fs::write(
4402 &path,
4403 serde_json::json!({
4404 "schema": 1,
4405 "id": "20260101-000000-aaaa",
4406 "title": "from before blocking existed",
4407 "instruction": "from before blocking existed",
4408 "repo": ".",
4409 "source": { "kind": "human" },
4410 "status": "queued",
4411 "created_at": Timestamp::now().to_string(),
4412 "updated_at": Timestamp::now().to_string(),
4413 })
4414 .to_string(),
4415 )
4416 .unwrap();
4417
4418 let task = q.get("20260101-000000-aaaa").expect("must still read");
4419 assert!(task.blocked_by.is_empty());
4420 assert!(task.block_reason.is_none());
4421 assert!(task.answers.is_empty());
4422 assert!(task.review_branch.is_none());
4423 }
4424
4425 #[test]
4426 fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
4427 let mut held = task("diagnosed");
4432 held.start("run-1".to_owned());
4433 held.fail("gate red", 1);
4434 held.diagnostic = Some("cargo test failed: ...".to_owned());
4435 assert_eq!(held.status, TaskStatus::Held);
4436
4437 held.release();
4438 assert!(held.diagnostic.is_none());
4439
4440 held.diagnostic = Some("cargo test failed: ...".to_owned());
4441 held.succeed();
4442 assert!(held.diagnostic.is_none());
4443 }
4444
4445 #[test]
4446 fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
4447 let mut t = task("retried");
4448 t.start("run-1".to_owned());
4449 t.diagnostic = Some("stale evidence from a previous hold".to_owned());
4450 t.fail("unrelated config error", 5);
4451 assert_eq!(t.status, TaskStatus::Failed);
4452 assert!(
4453 t.diagnostic.is_none(),
4454 "fail() must not let an old diagnostic outlive the run that produced it"
4455 );
4456 }
4457
4458 #[test]
4459 fn a_claim_is_exclusive_and_releases_on_drop() {
4460 let (_dir, q) = queue();
4461 let mut t = task("contended");
4462 q.put(&mut t).unwrap();
4463
4464 let held = q.claim(&t.id).unwrap();
4465 assert!(
4466 q.claim(&t.id).is_err(),
4467 "two daemons must not drive one task into two runs"
4468 );
4469 drop(held);
4470 assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
4471 }
4472
4473 #[test]
4474 fn a_round_trip_survives_disk() {
4475 let (_dir, q) = queue();
4476 let mut t = Task::new(
4477 "titled".to_owned(),
4478 "body".to_owned(),
4479 PathBuf::from("/repo"),
4480 Source::Agent {
4481 run: "20260101-000000-beef".to_owned(),
4482 node: "implement".to_owned(),
4483 },
4484 );
4485 t.priority = 3;
4486 q.put(&mut t).unwrap();
4487
4488 let back = q.get(&t.id).unwrap();
4489 assert_eq!(back.id, t.id);
4490 assert_eq!(back.priority, 3);
4491 assert_eq!(back.source.label(), "implement@beef");
4492 assert_eq!(q.get(t.short()).unwrap().id, t.id);
4494 }
4495
4496 #[test]
4497 fn an_unreadable_task_does_not_take_the_queue_down() {
4498 let (_dir, q) = queue();
4499 let mut t = task("fine");
4500 q.put(&mut t).unwrap();
4501 std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
4502
4503 let listed = q.list();
4504 assert_eq!(listed.len(), 1, "the readable task still lists");
4505 assert_eq!(listed[0].id, t.id);
4506 }
4507
4508 #[test]
4509 fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
4510 let (_dir, q) = queue();
4511 let path = q.path_of("20260101-000000-aaaa");
4512 std::fs::create_dir_all(q.root()).unwrap();
4513 std::fs::write(
4514 &path,
4515 serde_json::json!({
4516 "schema": SCHEMA,
4517 "id": "20260101-000000-aaaa",
4518 "title": "from before solo existed",
4519 "instruction": "from before solo existed",
4520 "repo": ".",
4521 "source": { "kind": "human" },
4522 "status": "queued",
4523 "created_at": Timestamp::now().to_string(),
4524 "updated_at": Timestamp::now().to_string(),
4525 })
4526 .to_string(),
4527 )
4528 .unwrap();
4529
4530 let task = q.get("20260101-000000-aaaa").expect("must still read");
4531 assert!(!task.solo, "a queue file with no `solo` field means false");
4532 }
4533
4534 #[test]
4535 fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
4536 let (_dir, q) = queue();
4537 let path = q.path_of("20260101-000000-bbbb");
4538 std::fs::create_dir_all(q.root()).unwrap();
4539 std::fs::write(
4540 &path,
4541 serde_json::json!({
4542 "schema": SCHEMA,
4543 "id": "20260101-000000-bbbb",
4544 "title": "from before urgent existed",
4545 "instruction": "from before urgent existed",
4546 "repo": ".",
4547 "source": { "kind": "human" },
4548 "status": "queued",
4549 "created_at": Timestamp::now().to_string(),
4550 "updated_at": Timestamp::now().to_string(),
4551 })
4552 .to_string(),
4553 )
4554 .unwrap();
4555
4556 let task = q.get("20260101-000000-bbbb").expect("must still read");
4557 assert!(
4558 !task.urgent,
4559 "a queue file with no `urgent` field means false, same as `solo`"
4560 );
4561 }
4562
4563 #[test]
4564 fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
4565 let (_dir, q) = queue();
4566 let mut t = task("from the future");
4567 q.put(&mut t).unwrap();
4568 let path = q.path_of(&t.id);
4569 let body = std::fs::read_to_string(&path)
4570 .unwrap()
4571 .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
4572 std::fs::write(&path, body).unwrap();
4573
4574 let err = q.get(&t.id).unwrap_err().to_string();
4575 assert!(err.contains("schema 99"), "{err}");
4576 }
4577
4578 #[test]
4579 fn revision_moves_when_the_queue_changes() {
4580 let (_dir, q) = queue();
4581 assert_eq!(q.revision(), 0, "an empty queue has no revision");
4582 let mut t = task("first");
4583 q.put(&mut t).unwrap();
4584 assert!(q.revision() > 0, "a written task moves the revision");
4585 }
4586
4587 #[test]
4588 fn revision_moves_when_deleting_an_older_task() {
4589 let (dir, q) = queue();
4590 let questions = Questions::at(dir.path().join("questions"));
4591 let mut t1 = task("older");
4592 q.put(&mut t1).unwrap();
4593 std::thread::sleep(std::time::Duration::from_millis(10));
4595 let mut t2 = task("newer");
4596 q.put(&mut t2).unwrap();
4597
4598 let rev_before = q.revision();
4599 q.remove(&t1.id, false, &questions).unwrap();
4600 let rev_after = q.revision();
4601
4602 assert_ne!(
4603 rev_before, rev_after,
4604 "deleting an older task must change the revision so other clients see the deletion"
4605 );
4606 }
4607
4608 #[test]
4609 fn removing_a_task_takes_it_out_of_the_listing() {
4610 let (dir, q) = queue();
4611 let questions = Questions::at(dir.path().join("questions"));
4612 let mut t = task("delete me");
4613 q.put(&mut t).unwrap();
4614 let removed = q.remove(t.short(), false, &questions).unwrap();
4615 assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
4616 assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
4617 assert!(q.list().is_empty());
4618 assert!(
4619 q.remove(&t.id, false, &questions).is_err(),
4620 "removing twice is an error"
4621 );
4622 }
4623
4624 #[test]
4625 fn removing_a_task_takes_its_stale_lock_with_it() {
4626 let (dir, q) = queue();
4627 let questions = Questions::at(dir.path().join("questions"));
4628 let mut t = task("interrupted");
4629 q.put(&mut t).unwrap();
4630
4631 let claim = q.claim(&t.id).unwrap();
4634 std::mem::forget(claim);
4635 assert!(
4636 q.claim(&t.id).is_err(),
4637 "the orphaned lock is what makes the task look claimed"
4638 );
4639
4640 let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
4642 assert!(err.contains("live daemon"), "{err}");
4643 assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
4644
4645 q.remove(&t.id, false, &questions).unwrap();
4647 assert!(q.list().is_empty());
4648 let mut again = task("interrupted");
4649 again.id = t.id.clone();
4650 q.put(&mut again).unwrap();
4651 assert!(
4652 q.claim(&t.id).is_ok(),
4653 "a task that comes back must be claimable, which a left-behind lock would prevent"
4654 );
4655 }
4656
4657 fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
4658 crate::notices::Notices::at(dir.join("notifications")).list()
4659 }
4660
4661 #[test]
4662 fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
4663 let (dir, q) = queue();
4664 let questions = Questions::at(dir.path().join("questions"));
4665 let mut dep = task("dependency");
4666 q.put(&mut dep).unwrap();
4667 let mut blocked = task("waiting");
4668 blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
4669 q.put(&mut blocked).unwrap();
4670
4671 let removed = q.remove(&dep.id, false, &questions).unwrap();
4672 assert_eq!(removed.released, [blocked.id.clone()]);
4673 assert!(removed.still_blocked.is_empty());
4674
4675 let after = q.get(&blocked.id).unwrap();
4676 assert_eq!(after.status, TaskStatus::Queued);
4677 assert!(after.blocked_by.is_empty());
4678 assert!(after.block_reason.is_none());
4679 let notes = notices_of(dir.path());
4680 assert_eq!(notes.len(), 1, "{notes:?}");
4681 assert_eq!(notes[0].severity, crate::notices::Severity::Info);
4682 assert_eq!(
4683 notes[0].link,
4684 Some(crate::notices::Link::Task {
4685 id: blocked.id.clone(),
4686 })
4687 );
4688 }
4689
4690 #[test]
4691 fn removing_one_of_two_dependencies_keeps_the_other() {
4692 let (dir, q) = queue();
4693 let questions = Questions::at(dir.path().join("questions"));
4694 let mut dep = task("dependency");
4695 q.put(&mut dep).unwrap();
4696 let mut other = task("other");
4697 q.put(&mut other).unwrap();
4698 let mut blocked = task("waiting");
4699 blocked.block(
4700 vec![dep.id.clone(), other.id.clone()],
4701 Some("waits on both".to_owned()),
4702 );
4703 q.put(&mut blocked).unwrap();
4704
4705 let removed = q.remove(&dep.id, false, &questions).unwrap();
4706 assert!(removed.released.is_empty());
4707 assert_eq!(removed.still_blocked, [blocked.id.clone()]);
4708
4709 let after = q.get(&blocked.id).unwrap();
4710 assert_eq!(after.status, TaskStatus::Blocked);
4711 assert_eq!(after.blocked_by, [other.id.clone()]);
4712 assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
4715 }
4716
4717 #[test]
4718 fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
4719 let (dir, q) = queue();
4720 let questions = Questions::at(dir.path().join("questions"));
4721 let mut dep = task("dependency");
4722 q.put(&mut dep).unwrap();
4723 let mut blocked = task("waiting");
4724 blocked.block(vec![dep.id.clone()], None);
4725 q.put(&mut blocked).unwrap();
4726
4727 let claim = q.claim(&blocked.id).unwrap();
4729 let removed = q.remove(&dep.id, false, &questions).unwrap();
4730 assert!(removed.released.is_empty());
4731 drop(claim);
4732
4733 let mut task = q.get(&blocked.id).unwrap();
4734 assert_eq!(task.status, TaskStatus::Blocked);
4735 assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
4736 assert_eq!(
4737 q.apply_deleted_blockers(&mut task, &questions),
4738 [dep.id.clone()]
4739 );
4740 assert_eq!(task.status, TaskStatus::Queued);
4741 }
4742
4743 #[test]
4744 fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
4745 let (dir, q) = queue();
4746 let questions = Questions::at(dir.path().join("questions"));
4747 let mut dep = task("dependency");
4748 dep.id = "20261013-193611-ca5b".into();
4749 q.put(&mut dep).unwrap();
4750 let mut waiting = task("waiting");
4751 waiting.block(vec!["ca5b".into()], None);
4752 q.put(&mut waiting).unwrap();
4753 q.remove(&dep.id, false, &questions).unwrap();
4754 assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
4755 assert_eq!(
4756 dependency(&q, &questions, "ca5b"),
4757 Dependency::Deleted(dep.id)
4758 );
4759 }
4760
4761 #[test]
4762 fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
4763 let (dir, queue) = queue();
4764 let questions = Questions::at(dir.path().join("questions"));
4765 let mut dep = task("dependency");
4766 dep.id = "20261013-193611-ca5b".into();
4767 queue.put(&mut dep).unwrap();
4768 std::fs::create_dir_all(questions.root()).unwrap();
4769 let question_id = "20261014-193611-ca5b";
4770 std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
4771 assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
4772 assert_eq!(
4773 dependency(&queue, &questions, "ca5b"),
4774 Dependency::Ambiguous
4775 );
4776 assert_eq!(
4777 dependency(&queue, &questions, &dep.id),
4778 Dependency::Task(dep.id)
4779 );
4780 assert_eq!(
4781 dependency(&queue, &questions, question_id),
4782 Dependency::Question(question_id.into())
4783 );
4784 assert_eq!(
4785 dependency(&queue, &questions, "missing"),
4786 Dependency::Missing
4787 );
4788 }
4789
4790 #[test]
4791 fn dependency_resolution_directory_errors_are_unknown_not_missing() {
4792 let dir = tempfile::tempdir().unwrap();
4793 let q = Queue::at(dir.path().join("not-a-directory"));
4794 std::fs::write(q.root(), "file").unwrap();
4795 let questions = Questions::at(dir.path().join("questions"));
4796 assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
4797 assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
4798 }
4799
4800 #[test]
4801 fn a_dependency_without_a_tombstone_is_still_missing() {
4802 let (dir, q) = queue();
4803 let questions = Questions::at(dir.path().join("questions"));
4804 let mut dep = task("dependency");
4805 q.put(&mut dep).unwrap();
4806 std::fs::remove_file(q.path_of(&dep.id)).unwrap();
4807 assert_eq!(
4808 missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
4809 [dep.id.clone()]
4810 );
4811 assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
4812 }
4813
4814 #[test]
4815 fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
4816 let mut t = task("held");
4817 t.hold_machine(Some("because".to_owned()));
4818 t.blocked_by = vec!["gone".to_owned()];
4819 assert!(!t.dependency_deleted("gone"));
4820 assert_eq!(t.status, TaskStatus::Held);
4821 }
4822
4823 #[test]
4824 fn a_failed_record_removal_takes_the_tombstone_back() {
4825 let (_dir, q) = queue();
4826 let mut t = task("stays");
4827 q.put(&mut t).unwrap();
4828 q.write_tombstone(&t.id).unwrap();
4829 let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
4830 assert!(err.is_err());
4831 assert!(!q.was_deleted(&t.id));
4834 }
4835
4836 fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
4837 let p = dir.join(name);
4838 std::fs::write(&p, body).unwrap();
4839 p
4840 }
4841
4842 #[test]
4843 fn an_attachment_copy_survives_deleting_its_source() {
4844 let (dir, q) = queue();
4845 let src = source_file(dir.path(), "shot.png", "pixels");
4846 let mut t = task("with a picture");
4847 let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
4848 q.put(&mut t).unwrap();
4849 std::fs::remove_file(&src).unwrap();
4850 assert_eq!(names, ["shot.png"]);
4851 let loaded = q.get(&t.id).unwrap();
4852 let paths = q.attachment_paths(&loaded);
4853 assert_eq!(paths.len(), 1);
4854 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
4855 }
4856
4857 #[test]
4858 fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
4859 let (dir, q) = queue();
4860 let mut t = task("bad names");
4861 for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
4862 let src = source_file(dir.path(), name, "x");
4863 assert!(
4864 q.attach(&mut t, &[src]).is_err(),
4865 "`{name}` must be refused"
4866 );
4867 }
4868 assert!(!crate::ask::valid_asset_name("C:foo.png"));
4872 #[cfg(not(windows))]
4873 {
4874 let src = source_file(dir.path(), "C:foo.png", "x");
4875 assert!(q.attach(&mut t, &[src]).is_err());
4876 }
4877 let long = format!("{}.png", "a".repeat(70));
4878 let src = source_file(dir.path(), &long, "x");
4879 assert!(q.attach(&mut t, &[src]).is_err());
4880 assert!(t.attachments.is_empty());
4881 assert!(!q.attachments_dir(&t.id).exists());
4882 }
4883
4884 #[test]
4885 fn a_taken_attachment_name_is_numbered_not_overwritten() {
4886 let (dir, q) = queue();
4887 let a = source_file(dir.path(), "shot.png", "one");
4888 let sub = dir.path().join("other");
4889 std::fs::create_dir_all(&sub).unwrap();
4890 let b = source_file(&sub, "shot.png", "two");
4891 let mut t = task("collision");
4892 q.attach(&mut t, &[a]).unwrap();
4893 q.attach(&mut t, &[b]).unwrap();
4894 assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
4895 let paths = q.attachment_paths(&t);
4896 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
4897 assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
4898 }
4899
4900 #[test]
4901 fn a_renumbered_name_stays_inside_the_length_bound() {
4902 let (dir, q) = queue();
4903 let name = format!("{}.png", "a".repeat(60));
4904 assert_eq!(name.len(), 64);
4905 let a = source_file(dir.path(), &name, "one");
4906 let sub = dir.path().join("other");
4907 std::fs::create_dir_all(&sub).unwrap();
4908 let b = source_file(&sub, &name, "two");
4909 let mut t = task("long");
4910 q.attach(&mut t, &[a, b]).unwrap();
4911 assert_eq!(t.attachments.len(), 2);
4912 assert!(
4913 t.attachments
4914 .iter()
4915 .all(|n| crate::ask::valid_asset_name(n))
4916 );
4917 assert!(t.attachments[1].ends_with("-2.png"));
4918 }
4919
4920 #[test]
4921 fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
4922 let (dir, q) = queue();
4923 let good = source_file(dir.path(), "good.png", "ok");
4924 let mut t = task("partial");
4925 q.attach(&mut t, &[good]).unwrap();
4926 let more = source_file(dir.path(), "more.png", "ok");
4927 let missing = dir.path().join("missing.png");
4928 assert!(q.attach(&mut t, &[more, missing]).is_err());
4929 assert_eq!(t.attachments, ["good.png"]);
4930 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4931 .unwrap()
4932 .flatten()
4933 .collect();
4934 assert_eq!(on_disk.len(), 1);
4935 }
4936
4937 fn block_put(q: &Queue, t: &Task) -> PathBuf {
4939 let tmp = q.path_of(&t.id).with_extension("json.tmp");
4940 std::fs::create_dir_all(&tmp).unwrap();
4941 tmp
4942 }
4943
4944 #[test]
4945 fn a_failed_put_leaves_no_new_attachment_directory() {
4946 let (dir, q) = queue();
4947 let mut t = task("fresh");
4948 let tmp = block_put(&q, &t);
4949 let src = source_file(dir.path(), "shot.png", "x");
4950 assert!(q.attach_and_put(&mut t, &[src]).is_err());
4951 assert!(t.attachments.is_empty());
4952 assert!(!q.attachments_dir(&t.id).exists());
4953 assert!(!q.path_of(&t.id).exists());
4954 std::fs::remove_dir(tmp).unwrap();
4955 }
4956
4957 #[test]
4958 fn a_failed_put_removes_only_the_copy_it_just_made() {
4959 let (dir, q) = queue();
4960 let mut t = task("edited");
4961 let first = source_file(dir.path(), "first.png", "1");
4962 q.attach_and_put(&mut t, &[first]).unwrap();
4963 block_put(&q, &t);
4964 let second = source_file(dir.path(), "second.png", "2");
4965 assert!(q.attach_and_put(&mut t, &[second]).is_err());
4966 assert_eq!(t.attachments, ["first.png"]);
4967 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4968 .unwrap()
4969 .flatten()
4970 .map(|e| e.file_name().to_string_lossy().into_owned())
4971 .collect();
4972 assert_eq!(on_disk, ["first.png"]);
4973 assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
4974 }
4975
4976 #[test]
4977 fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
4978 let (dir, q) = queue();
4979 let questions = Questions::at(dir.path().join("questions"));
4980 let gone = task("gone");
4981 let mut other = task("other");
4982 let mut live = task("live");
4983 q.put(&mut other).unwrap();
4984 q.put(&mut live).unwrap();
4985 let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
4988 std::fs::create_dir_all(&orphan).unwrap();
4989 std::fs::write(orphan.join("shot.png"), "x").unwrap();
4990 let busy = q.root.join(format!("{}.attachments.removing", live.id));
4993 std::fs::create_dir_all(&busy).unwrap();
4994
4995 q.remove(&other.id, false, &questions).unwrap();
4996 assert!(!orphan.exists(), "an orphan is swept");
4997 assert!(busy.exists(), "a removal in progress is left alone");
4998 }
4999
5000 #[test]
5001 fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
5002 let (dir, q) = queue();
5003 let questions = Questions::at(dir.path().join("questions"));
5004 let mut t = task("stuck");
5005 let src = source_file(dir.path(), "shot.png", "x");
5006 q.attach_and_put(&mut t, &[src]).unwrap();
5007 let aside = q.root.join(format!("{}.attachments.removing", t.id));
5010 std::fs::create_dir_all(&aside).unwrap();
5011 std::fs::write(aside.join("old.png"), "o").unwrap();
5012 assert!(q.remove(&t.id, false, &questions).is_err());
5013 assert!(q.path_of(&t.id).exists());
5014 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5015 }
5016
5017 #[test]
5018 fn a_failed_record_removal_puts_the_attachments_back() {
5019 let (dir, q) = queue();
5020 let mut t = task("rollback");
5021 let src = source_file(dir.path(), "shot.png", "x");
5022 q.attach_and_put(&mut t, &[src]).unwrap();
5023 let err = q
5024 .remove_record_with_attachments(&t.id, |_| {
5025 Err(std::io::Error::other("injected failure"))
5026 })
5027 .unwrap_err();
5028 assert!(format!("{err:#}").contains("injected failure"));
5029 assert!(q.path_of(&t.id).exists());
5030 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5031 assert!(
5032 !q.root
5033 .join(format!("{}.attachments.removing", t.id))
5034 .exists()
5035 );
5036 }
5037
5038 #[test]
5039 fn editing_a_task_keeps_its_attachments() {
5040 let (dir, q) = queue();
5041 let src = source_file(dir.path(), "shot.png", "x");
5042 let mut t = task("editable");
5043 q.attach(&mut t, &[src]).unwrap();
5044 t.edit("new".to_owned(), "new text".to_owned()).unwrap();
5045 q.put(&mut t).unwrap();
5046 assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
5047 }
5048
5049 #[test]
5050 fn removing_a_task_deletes_its_attachments() {
5051 let (dir, q) = queue();
5052 let questions = Questions::at(dir.path().join("questions"));
5053 let src = source_file(dir.path(), "shot.png", "x");
5054 let mut t = task("doomed");
5055 q.attach(&mut t, &[src]).unwrap();
5056 q.put(&mut t).unwrap();
5057 assert!(q.attachments_dir(&t.id).is_dir());
5058 q.remove(&t.id, false, &questions).unwrap();
5059 assert!(!q.attachments_dir(&t.id).exists());
5060 assert!(q.list().is_empty());
5061 }
5062
5063 #[test]
5064 fn a_task_written_before_attachments_still_reads() {
5065 let (_dir, q) = queue();
5066 let mut t = task("old");
5067 q.put(&mut t).unwrap();
5068 let path = q.path_of(&t.id);
5069 let mut v: serde_json::Value =
5070 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
5071 v.as_object_mut().unwrap().remove("attachments");
5072 std::fs::write(&path, v.to_string()).unwrap();
5073 assert!(q.get(&t.id).unwrap().attachments.is_empty());
5074 }
5075
5076 #[test]
5077 fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
5078 let q = Queue::at(PathBuf::from("relative-queue"));
5079 let mut t = task("rel");
5080 t.attachments.push("shot.png".to_owned());
5081 let paths = q.attachment_paths(&t);
5082 assert!(paths[0].is_absolute(), "{}", paths[0].display());
5083 assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
5084 }
5085
5086 #[test]
5087 fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
5088 let dir = tempfile::tempdir().unwrap();
5089 let q = Queue::at(dir.path().join("queue"));
5090 let mut t = task("resume");
5091 t.start("A".into());
5092 t.record_pass("A", 0, 1, None);
5093 t.start("B".into());
5094 q.put(&mut t).unwrap();
5095 assert_eq!(t.runs, vec!["A", "B"]);
5096 t.start("A".into());
5097 t.record_pass("A", 1, 3, None);
5098 q.put(&mut t).unwrap();
5099 let saved = q.get(&t.id).unwrap();
5100 assert_eq!(saved.runs, vec!["B", "A"]);
5101 assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
5102 assert_eq!(saved.attempts, 3);
5103 assert_eq!(saved.passes.len(), 2);
5104 assert_eq!(saved.passes[0].run, "A");
5105 assert_eq!(saved.passes[1].run, "A");
5106 }
5107
5108 #[test]
5109 fn repeated_start_records_one_run_and_separate_passes() {
5110 let mut t = task("resume");
5111 t.start("same-run".into());
5112 t.record_pass("same-run", 0, 1, None);
5113 t.start("same-run".into());
5114 t.record_pass("same-run", 1, 2, None);
5115 assert_eq!(t.runs, vec!["same-run"]);
5116 assert_eq!(t.attempts, 2);
5117 assert_eq!(t.passes.len(), 2);
5118 }
5119
5120 #[test]
5121 fn link_run_adds_a_run_once_and_touches_nothing_else() {
5122 let dir = tempfile::tempdir().unwrap();
5123 let queue = Queue::at(dir.path().join("queue"));
5124 let mut t = Task::new(
5125 "t".to_owned(),
5126 "do it".to_owned(),
5127 PathBuf::from("."),
5128 Source::Human,
5129 );
5130 queue.put(&mut t).unwrap();
5131 let before = queue.get(&t.id).unwrap();
5132
5133 let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
5134 assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
5135 assert_eq!(linked.status, before.status);
5136 assert_eq!(linked.attempts, before.attempts);
5137 assert_eq!(linked.interrupt, before.interrupt);
5138
5139 let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5140 assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
5141 assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
5142 assert!(queue.link_run("no-such-task", "r").is_err());
5143 }
5144
5145 #[test]
5146 fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
5147 let dir = tempfile::tempdir().unwrap();
5148 let queue = Queue::at(dir.path().join("queue"));
5149 let mut t = Task::new(
5150 "t".to_owned(),
5151 "do it".to_owned(),
5152 PathBuf::from("."),
5153 Source::Human,
5154 );
5155 queue.put(&mut t).unwrap();
5156 let mut snapshot = queue.get(&t.id).unwrap();
5158 queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5159
5160 snapshot.start("20260930-000000-aaaa".to_owned());
5161 queue.put(&mut snapshot).unwrap();
5162
5163 let stored = queue.get(&t.id).unwrap();
5164 assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
5165 assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
5166 assert_eq!(stored.attempts, 1);
5167 }
5168
5169 #[test]
5170 fn concurrent_links_and_daemon_saves_lose_nothing() {
5171 let dir = tempfile::tempdir().unwrap();
5172 let queue = Queue::at(dir.path().join("queue"));
5173 let mut t = Task::new(
5174 "t".to_owned(),
5175 "do it".to_owned(),
5176 PathBuf::from("."),
5177 Source::Human,
5178 );
5179 queue.put(&mut t).unwrap();
5180 let id = t.id.clone();
5181
5182 let linkers: Vec<_> = (0..4)
5183 .map(|n| {
5184 let (queue, id) = (queue.clone(), id.clone());
5185 std::thread::spawn(move || {
5186 for k in 0..10 {
5187 queue
5188 .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
5189 .unwrap();
5190 }
5191 })
5192 })
5193 .collect();
5194 let mut mine = queue.get(&id).unwrap();
5197 for k in 0..10 {
5198 mine.start(format!("20260930-000009-d{k:03}"));
5199 queue.put(&mut mine).unwrap();
5200 }
5201 for l in linkers {
5202 l.join().unwrap();
5203 }
5204
5205 let stored = queue.get(&id).unwrap();
5206 assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
5207 assert_eq!(
5208 stored.attempts, 10,
5209 "linking never rewinds the daemon's work"
5210 );
5211 }
5212
5213 fn age_lock(path: &Path) {
5214 let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
5215 f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
5216 .unwrap();
5217 }
5218
5219 #[test]
5220 fn concurrent_stale_takeover_yields_one_holder() {
5221 use std::sync::atomic::{AtomicUsize, Ordering};
5222 use std::sync::{Arc, Barrier};
5223 for _ in 0..5 {
5224 let dir = tempfile::tempdir().unwrap();
5225 let q = Queue::at(dir.path().to_path_buf());
5226 let lock = dir.path().join("t.write-lock");
5227 std::fs::write(&lock, "dead-0000").unwrap();
5228 age_lock(&lock);
5229 let n = 6;
5230 let barrier = Arc::new(Barrier::new(n));
5231 let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
5232 let handles: Vec<_> = (0..n)
5233 .map(|_| {
5234 let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
5235 std::thread::spawn(move || {
5236 b.wait();
5237 let g = q.lock_task("t").unwrap();
5238 let held = now.fetch_add(1, Ordering::SeqCst) + 1;
5239 max.fetch_max(held, Ordering::SeqCst);
5240 std::thread::sleep(std::time::Duration::from_millis(20));
5241 now.fetch_sub(1, Ordering::SeqCst);
5242 drop(g);
5243 })
5244 })
5245 .collect();
5246 for h in handles {
5247 h.join().unwrap();
5248 }
5249 assert_eq!(max.load(Ordering::SeqCst), 1);
5250 assert!(!lock.exists());
5251 }
5252 }
5253
5254 #[test]
5255 fn dropping_a_replaced_lock_leaves_the_new_holders_token() {
5256 let dir = tempfile::tempdir().unwrap();
5257 let q = Queue::at(dir.path().to_path_buf());
5258 let lock = dir.path().join("t.write-lock");
5259 let a = q.lock_task("t").unwrap();
5260 let replacement = owner_token();
5263 std::fs::write(&lock, &replacement).unwrap();
5264 drop(a);
5265 assert_eq!(std::fs::read_to_string(&lock).unwrap(), replacement);
5266 }
5267
5268 #[test]
5269 fn aged_live_task_lock_excludes_updates_until_retirement_releases_it() {
5270 let dir = tempfile::tempdir().unwrap();
5271 let q = Queue::at(dir.path().to_path_buf());
5272 let held = q.lock_task("t").unwrap();
5273 let path = dir.path().join("t.write-lock");
5274 age_lock(&path);
5275 assert!(!break_stale(&path, &held.token));
5276 let (started_tx, started_rx) = std::sync::mpsc::channel();
5277 let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
5278 let updater = std::thread::spawn(move || {
5279 started_tx.send(()).unwrap();
5280 let _write = q.lock_task("t").unwrap();
5281 acquired_tx.send(()).unwrap();
5282 });
5283 started_rx.recv().unwrap();
5284 assert!(
5285 acquired_rx
5286 .recv_timeout(std::time::Duration::from_millis(100))
5287 .is_err()
5288 );
5289 assert_eq!(std::fs::read_to_string(&path).unwrap(), held.token);
5290 drop(held);
5291 acquired_rx
5292 .recv_timeout(std::time::Duration::from_secs(2))
5293 .unwrap();
5294 updater.join().unwrap();
5295 assert!(!path.exists());
5296 }
5297
5298 #[test]
5299 fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
5300 let dir = tempfile::tempdir().unwrap();
5301 let lock = dir.path().join("t.write-lock");
5302 std::fs::write(&lock, "new-token").unwrap();
5303 assert!(!break_stale(&lock, "old-token"));
5304 assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
5305 assert!(!break_stale(&lock, "new-token"));
5307 assert!(lock.exists());
5308 age_lock(&lock);
5309 assert!(break_stale(&lock, "new-token"));
5310 assert!(!lock.exists());
5311 }
5312
5313 #[test]
5314 fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
5315 let dir = tempfile::tempdir().unwrap();
5316 let lock = dir.path().join("t.write-lock");
5317 std::fs::write(&lock, "dead-1").unwrap();
5318 age_lock(&lock);
5319 let marker = break_marker(&lock, "dead-1");
5320 std::fs::write(&marker, "crashed-remover").unwrap();
5321 assert!(!break_stale(&lock, "dead-1"));
5323 assert!(lock.exists());
5324 age_lock(&marker);
5326 assert!(break_stale(&lock, "dead-1"));
5327 assert!(!lock.exists());
5328 assert!(!marker.exists());
5329 }
5330
5331 fn chat_task() -> Task {
5332 Task::new(
5333 "t".to_owned(),
5334 "do t".to_owned(),
5335 PathBuf::from("."),
5336 Source::Agent {
5337 run: "20260902-000000-beef".to_owned(),
5338 node: CHAT_NODE.to_owned(),
5339 },
5340 )
5341 }
5342
5343 #[test]
5344 fn a_blocked_hold_keeps_its_anchor_but_a_new_run_starts_a_new_hold() {
5345 let mut t = chat_task();
5346 t.hold_machine(Some("waiting".into()));
5347 let old: Timestamp = "2020-01-01T00:00:00Z".parse().unwrap();
5348 t.held_at = Some(old);
5349 t.block(vec!["decision".into()], None);
5350 t.hold_machine(Some("same waiting decision".into()));
5351 assert_eq!(t.held_at, Some(old));
5352 t.start("real-new-run".into());
5353 t.fail("new failure", 1);
5354 assert_ne!(
5355 t.held_at,
5356 Some(old),
5357 "old answers cannot approve a new hold"
5358 );
5359 }
5360
5361 #[test]
5362 fn legacy_delivery_survives_upgrade_and_a_no_progress_run_link() {
5363 let (_dir, queue) = queue();
5364 let mut t = chat_task();
5365 t.start("old-run".into());
5366 t.hold_machine(Some("operator is considering recovery".into()));
5367 t.chat_notice = Some(ChatNotice {
5368 key: format!("held#{}", t.held_at.unwrap()),
5369 at: Timestamp::now(),
5370 outcome: ChatNoticeOutcome::Posted,
5371 });
5372 let mut raw = serde_json::to_value(&t).unwrap();
5373 raw.as_object_mut().unwrap().remove("chat_notices");
5374 raw.as_object_mut().unwrap().remove("hold_notice_cause");
5375 std::fs::create_dir_all(queue.path_of(&t.id).parent().unwrap()).unwrap();
5376 std::fs::write(queue.path_of(&t.id), serde_json::to_string(&raw).unwrap()).unwrap();
5377 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5378 queue
5379 .modify(&t.id, |task| {
5380 task.requeue();
5381 task.link_run("unprepared-run");
5382 task.startup_notice_cause("startup-source");
5383 task.fail("same unresolved preflight", 0);
5384 true
5385 })
5386 .unwrap();
5387 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5388 queue
5389 .modify(&t.id, |task| {
5390 task.succeed();
5391 true
5392 })
5393 .unwrap();
5394 assert_eq!(
5395 queue.get(&t.id).unwrap().chat_notice_due().as_deref(),
5396 Some("done")
5397 );
5398 }
5399
5400 #[test]
5401 fn a_dependency_wait_does_not_change_when_released_and_blocked_again() {
5402 let mut t = chat_task();
5403 t.block(vec!["a".into(), "b".into()], Some("waiting".into()));
5404 let key = t.chat_notice_key().unwrap();
5405 t.chat_notices.push(key);
5406 t.release();
5407 t.block(
5408 vec!["b".into(), "a".into(), "a".into()],
5409 Some("still waiting on prerequisites".into()),
5410 );
5411 assert_eq!(t.chat_notice_due(), None);
5412 t.block(
5413 vec!["new-question".into()],
5414 Some("different decision".into()),
5415 );
5416 assert!(t.chat_notice_due().is_some());
5417 }
5418
5419 #[test]
5420 fn running_startup_failures_preserve_stage_and_report_distinct_blockers() {
5421 let mut t = chat_task();
5422 t.start("existing-run".into());
5423 t.fail("implementation failed", 1);
5424 let failed = t.chat_notice_key().unwrap();
5425 t.chat_notices.push(failed.clone());
5426 t.release_operator("test", None);
5427 t.start("existing-run".into());
5428 t.fail_startup("startup-source", "same unresolved recovery source", 1);
5429 assert_eq!(
5430 t.chat_notice_due(),
5431 None,
5432 "no-progress recovery stays quiet"
5433 );
5434 t.release_operator("test", None);
5435 t.start("existing-run".into());
5436 t.fail_startup("startup-attachments", "attachment missing", 1);
5437 let attachments = t.chat_notice_due().expect("new attachment blocker");
5438 assert_ne!(attachments, failed);
5439 assert_eq!(
5440 t.hold_notice_cause.as_ref().unwrap().stage,
5441 "startup-attachments"
5442 );
5443 t.chat_notices.push(attachments);
5444 t.release_operator("test", None);
5445 t.start("existing-run".into());
5446 t.fail_startup(
5447 "startup-attachments",
5448 "same missing attachment, reworded",
5449 1,
5450 );
5451 assert_eq!(
5452 t.chat_notice_due(),
5453 None,
5454 "same stage after attempt counter reset"
5455 );
5456 t.release_operator("test", None);
5457 t.start("existing-run".into());
5458 t.fail_startup("startup-config", "different preflight stage", 1);
5459 assert!(t.chat_notice_due().is_some());
5460 assert_eq!(
5461 t.hold_notice_cause.as_ref().unwrap().stage,
5462 "startup-config"
5463 );
5464 }
5465
5466 #[test]
5467 fn structured_causes_preserve_no_progress_retries_and_report_new_passes() {
5468 let mut t = chat_task();
5469 t.hold_machine(Some("startup source missing".into()));
5470 t.startup_notice_cause("startup-source");
5471 let first = t.chat_notice_key().unwrap();
5472 t.chat_notices.push(first.clone());
5473 t.release_operator("test", None);
5474 t.startup_notice_cause("startup-source");
5475 t.hold_machine(Some("same source cannot be read".into()));
5476 assert_eq!(t.chat_notice_due(), None);
5477 t.startup_notice_cause("startup-config");
5478 assert!(t.chat_notice_due().is_some(), "different startup stage");
5479 t.notice_cause("failed", Some("existing-run"), Some(0));
5480 let run_failure = t.chat_notice_key().unwrap();
5481 t.chat_notices.push(run_failure.clone());
5482 t.release_operator("test", None);
5483 t.startup_notice_cause("startup-source");
5484 t.hold_machine(Some("no new work happened".into()));
5485 assert_eq!(t.chat_notice_due(), None);
5486 t.notice_cause("failed", Some("existing-run"), Some(1));
5487 assert!(t.chat_notice_due().is_some(), "a real resumed pass failed");
5488 assert_ne!(t.chat_notice_key().unwrap(), run_failure);
5489 }
5490
5491 #[test]
5492 fn legacy_hold_baseline_does_not_replay_or_hide_new_failures_and_completion() {
5493 let mut t = chat_task();
5494 t.hold_machine(Some("still waiting".into()));
5495 t.chat_notice = Some(ChatNotice {
5496 key: "held#2026-10-14T00:00:00Z".into(),
5497 at: Timestamp::now(),
5498 outcome: ChatNoticeOutcome::Posted,
5499 });
5500 assert_eq!(t.chat_notice_due(), None, "old hold baseline");
5501 t.import_chat_notice();
5502 t.notice_cause("failed", Some("new-run"), Some(0));
5503 assert!(t.chat_notice_due().is_some(), "new producer evidence");
5504 t.succeed();
5505 assert_eq!(t.chat_notice_due().as_deref(), Some("done"));
5506 }
5507
5508 #[test]
5509 fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
5510 let mut t = chat_task();
5511 assert_eq!(t.chat_notice_due(), None, "queued");
5512 t.start("r1".to_owned());
5513 assert_eq!(t.chat_notice_due(), None, "running");
5514 t.fail("boom", 3);
5515 assert_eq!(t.status, TaskStatus::Failed);
5516 assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
5517 t.start("r2".to_owned());
5518 t.park("upgrade");
5519 assert_eq!(t.chat_notice_due(), None, "parked");
5520 t.fail("boom", 1);
5521 assert_eq!(t.status, TaskStatus::Held);
5522 let key = t
5523 .chat_notice_due()
5524 .expect("held on the last attempt is final");
5525 t.chat_notice = Some(ChatNotice {
5526 key,
5527 at: Timestamp::now(),
5528 outcome: ChatNoticeOutcome::Posted,
5529 });
5530 assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
5531 t.release();
5532 assert_eq!(t.chat_notice_due(), None, "requeued after a release");
5533 t.start("r3".to_owned());
5534 t.succeed();
5535 assert!(t.chat_notice_due().is_some(), "done is a new final state");
5536 }
5537
5538 #[test]
5539 fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
5540 for key in ["done", "done#0", "done#1", "done#12"] {
5541 let mut t = chat_task();
5542 t.start("r1".to_owned());
5543 t.succeed();
5544 t.chat_notice = Some(ChatNotice {
5545 key: key.to_owned(),
5546 at: Timestamp::now(),
5547 outcome: ChatNoticeOutcome::Posted,
5548 });
5549 t.requeue();
5550 t.start("r2".to_owned());
5551 t.succeed();
5552 assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
5553 assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
5554 }
5555 }
5556
5557 #[test]
5558 fn a_release_without_progress_keeps_the_unresolved_notice_cause() {
5559 let mut t = chat_task();
5560 t.hold_manual(Some("wait".to_owned()));
5561 let first = t.chat_notice_due().expect("held");
5562 t.chat_notice = Some(ChatNotice {
5563 key: first.clone(),
5564 at: Timestamp::now(),
5565 outcome: ChatNoticeOutcome::Posted,
5566 });
5567 t.hold_manual(Some("still waiting".to_owned()));
5568 assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
5569 t.release();
5570 std::thread::sleep(std::time::Duration::from_millis(5));
5571 t.hold_manual(None);
5572 assert_eq!(t.chat_notice_due(), None, "no progress, same cause");
5573 }
5574
5575 #[test]
5576 fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
5577 let mut human = task("h");
5578 human.succeed();
5579 assert_eq!(human.chat_notice_due(), None);
5580 let mut inherited = task("i");
5582 inherited.origin_chat = Some("20260902-000000-beef".to_owned());
5583 inherited.succeed();
5584 assert_eq!(inherited.chat_notice_due(), None);
5585 let mut old = chat_task();
5586 old.schema = 12;
5587 old.status = TaskStatus::Done;
5588 assert_eq!(old.chat_notice_due(), None, "history is not replayed");
5589 let mut live = chat_task();
5590 live.schema = 12;
5591 live.start("r1".to_owned());
5592 live.succeed();
5593 assert!(live.chat_notice_due().is_some(), "started by this build");
5594 let mut held = chat_task();
5595 held.schema = 12;
5596 held.hold_machine(Some("config".to_owned()));
5597 assert!(held.chat_notice_due().is_some(), "held before any start");
5598 }
5599
5600 #[test]
5601 fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
5602 let tmp = tempfile::tempdir().expect("tempdir");
5603 let q = Queue::at(tmp.path().join("queue"));
5604 let mut t = chat_task();
5605 t.succeed();
5606 q.put(&mut t).expect("put");
5607 let mut stale = q.get(&t.id).expect("get");
5608 let key = t.chat_notice_due().expect("due");
5609 q.modify(&t.id, |x| {
5610 x.chat_notice = Some(ChatNotice {
5611 key: key.clone(),
5612 at: Timestamp::now(),
5613 outcome: ChatNoticeOutcome::Posted,
5614 });
5615 true
5616 })
5617 .expect("modify");
5618 q.put(&mut stale).expect("stale put");
5619 assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
5620 }
5621}