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 let mut counts = Self::default();
261 for t in tasks {
262 match t.status {
263 TaskStatus::Queued => counts.queued += 1,
264 TaskStatus::Running => counts.running += 1,
265 TaskStatus::Done => counts.done += 1,
266 TaskStatus::Failed => counts.failed += 1,
267 TaskStatus::Held => counts.held += 1,
268 TaskStatus::Blocked => counts.blocked += 1,
269 TaskStatus::Parked => counts.parked += 1,
270 }
271 }
272 counts
273 }
274}
275
276#[derive(Debug, Clone, Serialize, Deserialize)]
278#[serde(deny_unknown_fields)]
279pub struct Task {
280 pub schema: u32,
282 pub id: String,
284 pub title: String,
286 pub instruction: String,
288 pub repo: PathBuf,
290 pub source: Source,
292 #[serde(default)]
294 pub priority: i32,
295 #[serde(default)]
306 pub solo: bool,
307 pub status: TaskStatus,
309 #[serde(default)]
311 pub attempts: usize,
312 #[serde(default)]
314 pub attempt_limit: Option<NonZeroUsize>,
315 #[serde(default)]
318 pub runs: Vec<String>,
319 #[serde(default)]
321 pub last_error: Option<String>,
322 #[serde(default)]
335 pub hold_reason: Option<String>,
336 #[serde(default)]
339 pub hold_source: Option<HoldSource>,
340 #[serde(default)]
352 pub diagnostic: Option<String>,
353 #[serde(default)]
363 pub blocked_by: Vec<String>,
364 #[serde(default)]
367 pub block_reason: Option<String>,
368 #[serde(default)]
381 pub blocked_from: Option<TaskStatus>,
382 #[serde(default)]
384 pub blocked_at: Option<Timestamp>,
385 #[serde(default)]
396 pub answers: Vec<AnsweredQuestion>,
397 #[serde(default)]
403 pub triage_applied: Vec<String>,
404 #[serde(default)]
410 pub actions_applied: Vec<String>,
411 #[serde(default)]
416 pub resume_override: Option<OperatorResume>,
417 #[serde(default)]
420 pub resumption: Option<Resumption>,
421 #[serde(default)]
423 pub resumption_history: Vec<Resumption>,
424 #[serde(default)]
426 pub operator_decided_at: Option<Timestamp>,
427 #[serde(default)]
439 pub review_branch: Option<String>,
440 #[serde(default)]
442 pub review_instruction: Option<String>,
443 #[serde(default)]
445 pub pr_recovery: Option<crate::pr_recovery::Selection>,
446 #[serde(default)]
449 pub fresh_start: bool,
450 #[serde(default)]
452 pub replacement: Option<crate::merge_decision::Replacement>,
453 #[serde(default)]
464 pub interrupt: bool,
465 #[serde(default)]
488 pub urgent: bool,
489 #[serde(default)]
495 pub attachments: Vec<String>,
496 #[serde(default)]
500 pub followup: Option<FollowUp>,
501 #[serde(default)]
507 pub origin_chat: Option<String>,
508 #[serde(default)]
515 pub overrides: Option<RunOverrides>,
516 #[serde(default)]
523 pub review_of: Option<String>,
524 #[serde(default)]
529 pub held_at: Option<Timestamp>,
530 #[serde(default)]
534 pub park_reason: Option<String>,
535 #[serde(default)]
538 pub chat_notice: Option<ChatNotice>,
539 #[serde(default)]
541 pub chat_notices: Vec<String>,
542 #[serde(default)]
545 pub hold_notice_cause: Option<NoticeCause>,
546 #[serde(default)]
551 pub passes: Vec<TaskPass>,
552 pub created_at: Timestamp,
554 pub updated_at: Timestamp,
556}
557
558#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
560pub struct NoticeCause {
561 pub stage: String,
563 #[serde(default, skip_serializing_if = "Option::is_none")]
566 pub target: Option<String>,
567 pub run: Option<String>,
569 pub pass: Option<usize>,
571}
572
573pub(crate) fn is_done_notice_key(key: &str) -> bool {
575 key == "done"
576 || key
577 .strip_prefix("done#")
578 .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
579}
580
581#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
583#[serde(rename_all = "snake_case")]
584pub enum PassCost {
585 Spent,
587 Refunded,
589}
590
591#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
593pub struct TaskPass {
594 pub run: String,
596 pub pass_index: usize,
598 pub cost: PassCost,
600 pub attempts_after: usize,
602 pub settled_at: Timestamp,
604 #[serde(default)]
606 pub parked: Option<crate::run::ParkCause>,
607}
608
609#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
611#[serde(rename_all = "lowercase")]
612pub enum ChatNoticeOutcome {
613 Posted,
615 Skipped,
617}
618
619#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
621pub struct ChatNotice {
622 pub key: String,
624 pub at: Timestamp,
626 pub outcome: ChatNoticeOutcome,
628}
629
630#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
632pub struct RunOverrides {
633 #[serde(default)]
636 pub inherit_daemon_options: bool,
637 #[serde(default)]
639 pub merge: Option<String>,
640 #[serde(default)]
642 pub candidates: Option<usize>,
643 #[serde(default)]
645 pub judges: Option<usize>,
646 #[serde(default)]
648 pub reviewers: Option<usize>,
649 #[serde(default)]
651 pub review_rounds: Option<usize>,
652 #[serde(default)]
654 pub seed: Option<u64>,
655 #[serde(default)]
657 pub config: Option<PathBuf>,
658}
659
660impl RunOverrides {
661 pub fn merge_over(&mut self, other: &RunOverrides) {
663 macro_rules! take {
664 ($($f:ident),*) => {$(
665 if other.$f.is_some() {
666 self.$f.clone_from(&other.$f);
667 }
668 )*};
669 }
670 take!(
671 merge,
672 candidates,
673 judges,
674 reviewers,
675 review_rounds,
676 seed,
677 config
678 );
679 }
680
681 pub fn apply(&self, config: &mut crate::config::Config) {
683 if let Some(n) = self.candidates {
684 config.graph.implementers = n;
685 }
686 if let Some(n) = self.judges {
687 config.graph.judges = n;
688 }
689 if let Some(n) = self.reviewers {
690 config.graph.reviewers = n;
691 }
692 if let Some(n) = self.review_rounds {
693 config.graph.review_rounds = n;
694 }
695 if let Some(m) = &self.merge
696 && let Ok(mode) = crate::daemon::merge_mode(m)
697 {
698 config.merge.mode = mode;
699 }
700 if let Some(s) = self.seed {
701 config.blind.seed = Some(s);
702 }
703 }
704}
705
706#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
708pub struct FollowUp {
709 pub run: String,
711 #[serde(default)]
713 pub origin_task: Option<String>,
714 pub pr: String,
716 pub findings: Vec<String>,
718 pub generation: u32,
721}
722
723#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
725pub struct Resumption {
726 pub generation: u64,
728 pub source: String,
730 pub question_id: Option<String>,
732 pub at: Timestamp,
734 pub instruction: String,
736 pub answers_through: usize,
738 pub runs_through: usize,
741 pub cleared_reason: Option<String>,
743 #[serde(default)]
745 pub blocking_evidence: Vec<String>,
746}
747
748#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
751pub struct OperatorResume {
752 pub question_id: String,
754 pub at: Timestamp,
756 #[serde(default)]
759 pub conductor_rehold: Option<String>,
760 #[serde(default)]
763 pub forced: bool,
764 #[serde(default)]
771 pub pinned_run: Option<String>,
772}
773
774#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
777pub struct AnsweredQuestion {
778 pub question: String,
780 pub answer: String,
782}
783
784impl Task {
785 pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
787 let now = Timestamp::now();
788 let origin_chat = match &source {
789 Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
790 _ => None,
791 };
792 Self {
793 schema: SCHEMA,
794 id: new_id(),
795 title,
796 instruction,
797 repo,
798 source,
799 priority: 0,
800 solo: false,
801 status: TaskStatus::Queued,
802 attempts: 0,
803 attempt_limit: None,
804 runs: Vec::new(),
805 last_error: None,
806 hold_reason: None,
807 hold_source: None,
808 diagnostic: None,
809 blocked_by: Vec::new(),
810 block_reason: None,
811 blocked_from: None,
812 blocked_at: None,
813 answers: Vec::new(),
814 triage_applied: Vec::new(),
815 actions_applied: Vec::new(),
816 resume_override: None,
817 resumption: None,
818 resumption_history: Vec::new(),
819 operator_decided_at: None,
820 review_branch: None,
821 review_instruction: None,
822 pr_recovery: None,
823 fresh_start: false,
824 replacement: None,
825 interrupt: false,
826 urgent: false,
827 attachments: Vec::new(),
828 followup: None,
829 origin_chat,
830 overrides: None,
831 review_of: None,
832 held_at: None,
833 park_reason: None,
834 chat_notice: None,
835 chat_notices: Vec::new(),
836 hold_notice_cause: None,
837 passes: Vec::new(),
838 created_at: now,
839 updated_at: now,
840 }
841 }
842
843 pub fn chat_talk(&self) -> Option<&str> {
846 if let Some(id) = &self.origin_chat {
847 return Some(id);
848 }
849 match &self.source {
850 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
851 _ => None,
852 }
853 }
854
855 pub fn filed_by_chat(&self) -> Option<&str> {
859 match &self.source {
860 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
861 _ => None,
862 }
863 }
864
865 pub fn chat_notice_key(&self) -> Option<String> {
868 match self.status {
869 TaskStatus::Done => Some("done".to_owned()),
870 TaskStatus::Held => {
871 let cause = self
872 .hold_notice_cause
873 .clone()
874 .unwrap_or_else(|| self.unresolved_hold_cause());
875 Some(format!(
876 "cause:{}",
877 serde_json::to_string(&cause).expect("cause")
878 ))
879 }
880 TaskStatus::Blocked => {
881 let ids: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
882 let evidence = self
885 .hold_notice_cause
886 .as_ref()
887 .filter(|c| c.stage.starts_with("conductor-evidence:"));
888 Some(format!(
889 "dependencies:{}",
890 serde_json::to_string(&(ids, evidence)).expect("ids")
891 ))
892 }
893 _ => None,
894 }
895 }
896
897 fn unresolved_hold_cause(&self) -> NoticeCause {
898 NoticeCause {
899 stage: "hold".into(),
900 target: None,
901 run: self.runs.last().cloned(),
902 pass: None,
903 }
904 }
905
906 pub(crate) fn legacy_chat_notice_key(&self) -> Option<String> {
908 let key = match self.status {
909 TaskStatus::Held if self.hold_notice_cause.is_none() => Some(format!(
910 "held#{}",
911 self.held_at.map(|t| t.to_string()).unwrap_or_default()
912 )),
913 TaskStatus::Blocked => Some(format!(
914 "blocked#{}#{}",
915 self.blocked_by.join(","),
916 self.block_reason.as_deref().unwrap_or_default()
917 )),
918 _ => None,
919 };
920 key.filter(|key| self.legacy_notice_matches(key))
921 }
922
923 pub(crate) fn legacy_notice_matches(&self, key: &str) -> bool {
924 match self.status {
925 TaskStatus::Held if self.hold_notice_cause.is_none() => key
926 .strip_prefix("held#")
927 .is_some_and(|time| time.is_empty() || time.parse::<Timestamp>().is_ok()),
928 TaskStatus::Blocked
929 if self
930 .hold_notice_cause
931 .as_ref()
932 .is_none_or(|c| !c.stage.starts_with("conductor-evidence:")) =>
933 {
934 key.strip_prefix("blocked#").is_some_and(|tail| {
935 let Some((ids, _)) = tail.split_once('#') else {
936 return false;
937 };
938 let old: std::collections::BTreeSet<_> =
939 ids.split(',').filter(|id| !id.is_empty()).collect();
940 let current: std::collections::BTreeSet<_> =
941 self.blocked_by.iter().map(String::as_str).collect();
942 old == current
943 })
944 }
945 _ => false,
946 }
947 }
948
949 fn import_chat_notice(&mut self) {
950 if let Some(n) = &self.chat_notice {
951 let key = if is_done_notice_key(&n.key) {
952 "done".to_owned()
953 } else if self.legacy_notice_matches(&n.key) {
954 self.chat_notice_key().unwrap_or_else(|| n.key.clone())
955 } else {
956 n.key.clone()
957 };
958 if !self.chat_notices.contains(&key) {
959 self.chat_notices.push(key);
960 }
961 }
962 }
963
964 pub(crate) fn notice_cause(&mut self, stage: &str, run: Option<&str>, pass: Option<usize>) {
967 self.import_chat_notice();
968 self.hold_notice_cause = Some(NoticeCause {
969 stage: stage.to_owned(),
970 target: None,
971 run: run.map(str::to_owned),
972 pass,
973 });
974 }
975
976 pub(crate) fn startup_notice_cause(&mut self, stage: &str) {
979 let recovery_only = matches!(
980 stage,
981 "startup-source" | "startup-candidates" | "startup-public-head"
982 );
983 match self.hold_notice_cause.as_ref() {
984 Some(cause) if recovery_only && !cause.stage.starts_with("startup-") => return,
985 None if recovery_only
986 && self
987 .chat_notice
988 .as_ref()
989 .is_some_and(|n| !is_done_notice_key(&n.key)) =>
990 {
991 return;
992 }
993 _ => {}
994 }
995 self.notice_cause(stage, None, None);
996 }
997
998 pub(crate) fn fail_startup(
1001 &mut self,
1002 stage: &str,
1003 why: impl Into<String>,
1004 max_attempts: usize,
1005 ) {
1006 let unresolved = self.hold_notice_cause.clone();
1007 self.fail(why, max_attempts);
1008 self.hold_notice_cause = unresolved;
1009 self.startup_notice_cause(stage);
1010 }
1011
1012 pub(crate) fn fail_startup_target(
1014 &mut self,
1015 stage: &str,
1016 target: String,
1017 why: impl Into<String>,
1018 max_attempts: usize,
1019 ) {
1020 self.fail_startup(stage, why, max_attempts);
1021 if let Some(cause) = self.hold_notice_cause.as_mut() {
1022 cause.target = Some(target);
1023 }
1024 }
1025
1026 pub fn chat_notice_due(&self) -> Option<String> {
1028 self.filed_by_chat()?;
1029 if self.schema < 13 {
1030 return None;
1031 }
1032 let key = self.chat_notice_key()?;
1033 if self.chat_notices.contains(&key) {
1034 return None;
1035 }
1036 match &self.chat_notice {
1037 Some(n)
1038 if n.key == key
1039 || (key == "done" && is_done_notice_key(&n.key))
1040 || self.legacy_notice_matches(&n.key) =>
1041 {
1042 None
1043 }
1044 _ => Some(key),
1045 }
1046 }
1047
1048 pub fn short(&self) -> &str {
1050 short(&self.id)
1051 }
1052
1053 pub fn mark_triage_applied(&mut self, question_id: &str) {
1055 if !self.triage_applied(question_id) {
1056 self.triage_applied.push(question_id.to_owned());
1057 }
1058 }
1059
1060 pub fn action_applied(&self, question_id: &str) -> bool {
1063 self.actions_applied.iter().any(|id| id == question_id)
1064 }
1065
1066 pub fn mark_action_applied(&mut self, question_id: &str) {
1068 if !self.action_applied(question_id) {
1069 self.actions_applied.push(question_id.to_owned());
1070 }
1071 }
1072
1073 pub fn triage_applied(&self, question_id: &str) -> bool {
1075 self.triage_applied.iter().any(|id| id == question_id)
1076 }
1077
1078 pub fn start(&mut self, run: String) {
1091 self.stamp_schema();
1092 self.status = TaskStatus::Running;
1093 self.attempts += 1;
1094 self.runs.retain(|id| id != &run);
1095 self.runs.push(run);
1096 self.last_error = None;
1097 self.park_reason = None;
1098 self.fresh_start = false;
1099 self.interrupt = false;
1100 self.resume_override = None;
1102 }
1103
1104 pub fn link_run(&mut self, run: &str) -> bool {
1109 if self.runs.iter().any(|r| r == run) {
1110 return false;
1111 }
1112 self.runs.push(run.to_owned());
1113 true
1114 }
1115
1116 pub fn succeed(&mut self) {
1126 self.stamp_schema();
1127 self.status = TaskStatus::Done;
1128 self.resume_override = None;
1129 self.last_error = None;
1130 self.park_reason = None;
1131 self.hold_reason = None;
1132 self.hold_source = None;
1133 self.diagnostic = None;
1134 self.blocked_by.clear();
1135 self.block_reason = None;
1136 self.blocked_from = None;
1137 }
1138
1139 pub fn already_landed(&mut self, note: impl Into<String>) {
1147 self.succeed();
1148 self.attempts = self.attempts.saturating_sub(1);
1149 self.last_error = Some(note.into());
1150 }
1151
1152 pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
1182 if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
1183 return &[];
1184 }
1185 &self.runs[..self.runs.len() - 1]
1186 }
1187
1188 pub fn successor_of(&self, run: &str) -> Option<&String> {
1196 let pos = self.runs.iter().rposition(|r| r == run)?;
1197 self.runs.get(pos + 1)
1198 }
1199
1200 pub fn earlier_attempts(&self) -> &[String] {
1207 &self.runs
1208 }
1209
1210 fn stamp_schema(&mut self) {
1217 self.schema = self.schema.max(SCHEMA);
1218 }
1219
1220 fn note_held(&mut self) {
1221 self.stamp_schema();
1222 let continuing = self.status == TaskStatus::Held
1223 || (self.status == TaskStatus::Blocked
1224 && (self.blocked_from == Some(TaskStatus::Held)
1225 || (self.blocked_from.is_none()
1226 && (self.hold_reason.is_some() || self.hold_source.is_some()))));
1227 if !continuing || self.held_at.is_none() {
1228 self.held_at = Some(Timestamp::now());
1229 }
1230 }
1231
1232 pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
1248 let why = why.into();
1249 if self.status == TaskStatus::Running {
1250 let run = self.runs.last().cloned();
1251 self.notice_cause("run-failed", run.as_deref(), Some(self.attempts));
1252 }
1253 self.diagnostic = None;
1254 self.park_reason = None;
1255 self.status = if self.attempts >= self.effective_attempt_limit(max_attempts) {
1256 self.note_held();
1257 self.hold_source = Some(HoldSource::Machine);
1258 self.hold_reason = Some(why.clone());
1259 TaskStatus::Held
1260 } else {
1261 TaskStatus::Failed
1262 };
1263 self.last_error = Some(why);
1264 }
1265
1266 pub fn stall(&mut self, why: impl Into<String>) {
1276 self.last_error = Some(why.into());
1277 self.diagnostic = None;
1278 self.attempts = self.attempts.saturating_sub(1);
1279 self.status = TaskStatus::Failed;
1280 }
1281
1282 pub fn park(&mut self, why: impl Into<String>) {
1288 self.diagnostic = None;
1289 self.attempts = self.attempts.saturating_sub(1);
1290 self.status = TaskStatus::Parked;
1291 self.park_reason = Some(why.into());
1292 self.schema = self.schema.max(SCHEMA);
1295 }
1296
1297 pub fn record_pass(
1300 &mut self,
1301 run: &str,
1302 pass_index: usize,
1303 attempts_before: usize,
1304 parked: Option<crate::run::ParkCause>,
1305 ) {
1306 if self
1307 .passes
1308 .iter()
1309 .any(|p| p.run == run && p.pass_index == pass_index)
1310 {
1311 return;
1312 }
1313 let cost = if self.attempts < attempts_before {
1314 PassCost::Refunded
1315 } else {
1316 PassCost::Spent
1317 };
1318 self.passes.push(TaskPass {
1319 run: run.to_owned(),
1320 pass_index,
1321 cost,
1322 attempts_after: self.attempts,
1323 settled_at: Timestamp::now(),
1324 parked,
1325 });
1326 self.schema = self.schema.max(SCHEMA);
1327 }
1328
1329 pub fn operator_held(&self) -> bool {
1335 self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
1336 }
1337
1338 pub fn hold_manual(&mut self, reason: Option<String>) {
1348 self.operator_decided_at = Some(Timestamp::now());
1350 self.hold_unrequested_retry(reason);
1351 }
1352
1353 pub(crate) fn hold_unrequested_retry(&mut self, reason: Option<String>) {
1356 self.park_reason = None;
1357 self.note_held();
1358 self.status = TaskStatus::Held;
1359 if reason.is_some() {
1360 self.hold_reason = reason;
1361 }
1362 self.hold_source = Some(HoldSource::Manual);
1363 self.blocked_by.clear();
1364 self.block_reason = None;
1365 self.blocked_from = None;
1366 }
1367
1368 pub fn hold_machine(&mut self, reason: Option<String>) {
1373 self.park_reason = None;
1374 self.note_held();
1375 self.status = TaskStatus::Held;
1376 if reason.is_some() {
1377 self.hold_reason = reason;
1378 }
1379 self.hold_source = Some(HoldSource::Machine);
1380 self.blocked_by.clear();
1381 self.block_reason = None;
1382 self.blocked_from = None;
1383 }
1384
1385 pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
1395 self.stamp_schema();
1396 let before: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
1397 let after: std::collections::BTreeSet<_> = blocked_by.iter().collect();
1398 if self.status != TaskStatus::Blocked || before != after {
1399 self.blocked_at = Some(Timestamp::now());
1400 }
1401 if self.status != TaskStatus::Blocked {
1402 self.blocked_from = Some(self.status);
1403 }
1404 self.status = TaskStatus::Blocked;
1405 self.blocked_by = blocked_by;
1406 self.block_reason = reason;
1407 }
1408
1409 pub fn unblock(&mut self, resolved_id: &str) {
1432 if self.status != TaskStatus::Blocked {
1433 return;
1434 }
1435 self.blocked_by.retain(|id| id != resolved_id);
1436 self.restore_if_unblocked();
1437 }
1438
1439 pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
1450 if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
1451 return false;
1452 }
1453 self.blocked_by.retain(|id| id != deleted_id);
1454 self.restore_if_unblocked();
1455 true
1456 }
1457
1458 fn restore_if_unblocked(&mut self) {
1462 if self.blocked_by.is_empty() {
1463 self.status = match self.blocked_from {
1464 Some(TaskStatus::Running) => TaskStatus::Queued,
1465 Some(other) => other,
1466 None if self.hold_reason.is_some() || self.hold_source.is_some() => {
1467 TaskStatus::Held
1468 }
1469 None => TaskStatus::Queued,
1470 };
1471 self.block_reason = None;
1472 self.blocked_from = None;
1473 }
1474 }
1475
1476 pub fn record_answer(&mut self, question: String, answer: String) {
1481 self.answers.push(AnsweredQuestion { question, answer });
1482 }
1483
1484 pub fn request_review(&mut self, branch: String) {
1488 self.release();
1489 self.review_branch = Some(branch);
1490 self.review_instruction = Some(self.instruction.clone());
1491 }
1492
1493 pub fn requeue(&mut self) {
1496 self.release();
1497 self.review_branch = None;
1499 self.review_instruction = None;
1500 self.fresh_start = true;
1501 }
1502
1503 pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
1512 if branch.is_some() {
1513 self.review_branch = branch;
1514 self.review_instruction = Some(self.instruction.clone());
1515 }
1516 self.hold_machine(Some(reason));
1517 }
1518
1519 pub fn set_urgent(&mut self, urgent: bool) -> Result<()> {
1521 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1522 bail!(
1523 "task {} is {}; urgent can only change while queued or held; \
1524 wait for the active attempt to finish and requeue or hold the task first",
1525 self.short(),
1526 self.status.as_str()
1527 );
1528 }
1529 self.urgent = urgent;
1530 Ok(())
1531 }
1532
1533 pub fn set_priority(&mut self, priority: i32) -> Result<()> {
1542 if self.status == TaskStatus::Running {
1543 bail!(
1544 "task {} is running; its priority cannot be changed until \
1545 this attempt finishes",
1546 self.short()
1547 );
1548 }
1549 self.priority = priority;
1550 Ok(())
1551 }
1552
1553 pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
1568 if interrupt && !self.status.runnable() {
1569 bail!(
1570 "task {} is {}; only a queued or failed task can be marked \
1571 to interrupt",
1572 self.short(),
1573 self.status.as_str()
1574 );
1575 }
1576 self.interrupt = interrupt;
1577 Ok(())
1578 }
1579
1580 pub fn effective_attempt_limit(&self, daemon_limit: usize) -> usize {
1582 self.attempt_limit.map_or(daemon_limit, NonZeroUsize::get)
1583 }
1584
1585 pub fn change_attempt_limit(&mut self, limit: Option<NonZeroUsize>) -> Result<()> {
1588 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1589 bail!("only queued or held tasks can change the attempt limit");
1590 }
1591 self.stamp_schema();
1592 self.attempt_limit = limit;
1593 Ok(())
1594 }
1595
1596 pub fn implementers(&self) -> Option<usize> {
1598 if self.solo {
1599 Some(1)
1600 } else {
1601 self.overrides.as_ref().and_then(|o| o.candidates)
1602 }
1603 }
1604
1605 pub fn change_implementers(&mut self, count: u8) -> Result<()> {
1608 self.check_implementers_editable()?;
1609 let explicit = self.overrides.as_ref().and_then(|o| o.config.as_deref());
1610 let (mut config, _) = crate::config::Config::discover(&self.repo, explicit)?;
1611 if let Some(o) = &self.overrides {
1612 o.apply(&mut config);
1613 }
1614 self.set_implementers(count, &config)
1615 }
1616
1617 fn check_implementers_editable(&self) -> Result<()> {
1618 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1619 bail!(
1620 "task {} is {}; only queued or held tasks can change implementers",
1621 self.short(),
1622 self.status.as_str()
1623 );
1624 }
1625 Ok(())
1626 }
1627
1628 pub fn set_implementers(&mut self, count: u8, config: &crate::config::Config) -> Result<()> {
1630 self.check_implementers_editable()?;
1631 crate::talk::check_implementers(count, config).map_err(anyhow::Error::msg)?;
1632 self.stamp_schema();
1633 self.solo = count == 1;
1634 let mut overrides = self.overrides.clone().unwrap_or_else(|| RunOverrides {
1635 inherit_daemon_options: true,
1636 ..Default::default()
1637 });
1638 overrides.candidates = (count > 1).then_some(usize::from(count));
1639 let mut settings = overrides.clone();
1640 settings.inherit_daemon_options = false;
1641 self.overrides = if overrides.inherit_daemon_options && settings == RunOverrides::default()
1643 {
1644 None
1645 } else {
1646 Some(overrides)
1647 };
1648 Ok(())
1649 }
1650
1651 pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
1663 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1664 bail!(
1665 "task {} is {}; only a queued or held task's instruction can \
1666 be edited",
1667 self.short(),
1668 self.status.as_str()
1669 );
1670 }
1671 if self.review_branch.is_some() && self.review_instruction.is_none() {
1674 self.review_instruction = Some(self.instruction.clone());
1675 }
1676 if self.instruction != instruction {
1677 self.resume_override = None;
1680 }
1681 self.title = title;
1682 self.instruction = instruction;
1683 Ok(())
1684 }
1685
1686 pub fn handed_off(&mut self, why: impl Into<String>) {
1706 let why = why.into();
1707 self.diagnostic = None;
1708 self.note_held();
1709 self.status = TaskStatus::Held;
1710 self.hold_source = Some(HoldSource::Machine);
1711 self.hold_reason = Some(why.clone());
1712 self.last_error = Some(why);
1713 }
1714
1715 pub fn release_operator(&mut self, source: &str, question_id: Option<&str>) {
1717 if self
1718 .resumption
1719 .as_ref()
1720 .is_some_and(|r| r.instruction != self.instruction)
1721 {
1722 self.resume_override = None;
1725 }
1726 let approval = Resumption {
1727 generation: self.resumption.as_ref().map_or(1, |r| r.generation + 1),
1728 source: source.to_owned(),
1729 question_id: question_id.map(str::to_owned),
1730 at: Timestamp::now(),
1731 instruction: self.instruction.clone(),
1732 answers_through: self.answers.len(),
1733 runs_through: self.runs.len(),
1734 blocking_evidence: Vec::new(),
1735 cleared_reason: self
1736 .hold_reason
1737 .clone()
1738 .or_else(|| self.block_reason.clone()),
1739 };
1740 self.release();
1741 self.operator_decided_at = Some(approval.at);
1742 if let Some(previous) = self.resumption.replace(approval) {
1743 self.resumption_history.push(previous);
1744 }
1745 }
1746
1747 pub fn release(&mut self) {
1751 self.import_chat_notice();
1754 if self.hold_notice_cause.is_none()
1755 && matches!(self.status, TaskStatus::Held | TaskStatus::Blocked)
1756 {
1757 self.hold_notice_cause = Some(self.unresolved_hold_cause());
1758 }
1759 let refused_handover = self.status == TaskStatus::Held
1760 && self.hold_source == Some(HoldSource::Machine)
1761 && self.review_branch.is_some()
1762 && self
1763 .review_instruction
1764 .as_ref()
1765 .is_none_or(|i| *i == self.instruction);
1766 self.status = TaskStatus::Queued;
1767 self.held_at = None;
1768 self.attempts = 0;
1769 self.last_error = None;
1770 self.park_reason = None;
1771 self.hold_reason = None;
1774 self.hold_source = None;
1775 self.diagnostic = None;
1776 self.blocked_by.clear();
1781 self.block_reason = None;
1782 self.blocked_from = None;
1783 if !refused_handover {
1787 self.review_branch = None;
1788 self.review_instruction = None;
1789 }
1790 self.fresh_start = false;
1791 }
1792}
1793
1794fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
1798 use std::io::ErrorKind;
1799 let (stem, ext) = match name.rfind('.') {
1800 Some(i) if i > 0 => (&name[..i], &name[i..]),
1801 _ => (name, ""),
1802 };
1803 for n in 1u32.. {
1804 let candidate = if n == 1 {
1805 name.to_owned()
1806 } else {
1807 let suffix = format!("-{n}");
1808 let room = 64usize.saturating_sub(suffix.len() + ext.len());
1809 let stem: String = stem.chars().take(room).collect();
1810 format!("{stem}{suffix}{ext}")
1811 };
1812 if !crate::ask::valid_asset_name(&candidate) {
1813 bail!("no valid attachment name is left for `{name}`");
1814 }
1815 let path = dir.join(&candidate);
1816 match std::fs::OpenOptions::new()
1817 .write(true)
1818 .create_new(true)
1819 .open(&path)
1820 {
1821 Ok(mut out) => {
1822 let copied = std::fs::File::open(src)
1823 .and_then(|mut input| std::io::copy(&mut input, &mut out));
1824 if let Err(e) = copied {
1825 drop(out);
1826 let _ = std::fs::remove_file(&path);
1827 return Err(e).with_context(|| format!("copy {}", src.display()));
1828 }
1829 return Ok((candidate, path));
1830 }
1831 Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
1832 Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
1833 }
1834 }
1835 unreachable!("the counter never runs out")
1836}
1837
1838const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
1841
1842struct TaskLock {
1845 path: PathBuf,
1846 token: String,
1847}
1848
1849fn owner_token() -> String {
1852 static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
1853 let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1854 let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
1855 format!("{}-{:016x}", std::process::id(), r.next_u64())
1856}
1857
1858fn break_marker(path: &Path, token: &str) -> PathBuf {
1861 let mut h: u64 = 0xcbf29ce484222325;
1862 for b in token.bytes() {
1863 h ^= u64::from(b);
1864 h = h.wrapping_mul(0x100000001b3);
1865 }
1866 let mut name = path.as_os_str().to_owned();
1867 name.push(format!(".break-{h:016x}"));
1868 PathBuf::from(name)
1869}
1870
1871fn older_than_stale(path: &Path) -> bool {
1872 std::fs::metadata(path)
1873 .and_then(|m| m.modified())
1874 .ok()
1875 .and_then(|t| t.elapsed().ok())
1876 .is_some_and(|age| age > TASK_LOCK_STALE)
1877}
1878
1879const MAX_MARKER_DEPTH: u8 = 3;
1883
1884fn take_marker(path: &Path, token: &str, depth: u8) -> Option<(PathBuf, String)> {
1890 use std::io::Write;
1891 let marker = break_marker(path, token);
1892 for _ in 0..2 {
1893 match std::fs::OpenOptions::new()
1894 .write(true)
1895 .create_new(true)
1896 .open(&marker)
1897 {
1898 Ok(mut f) => {
1899 let mine = owner_token();
1900 if f.write_all(mine.as_bytes()).is_err() {
1901 drop(f);
1902 let _ = std::fs::remove_file(&marker);
1903 return None;
1904 }
1905 return Some((marker, mine));
1906 }
1907 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
1908 let judged = std::fs::read_to_string(&marker).ok();
1909 match judged.filter(|_| older_than_stale(&marker)) {
1910 Some(judged) if depth < MAX_MARKER_DEPTH => {
1911 remove_lock_if(&marker, &judged, true, depth + 1)?;
1912 }
1913 _ => return None,
1914 }
1915 }
1916 Err(_) => return None,
1917 }
1918 }
1919 None
1920}
1921
1922fn remove_lock_if(path: &Path, token: &str, require_stale: bool, depth: u8) -> Option<bool> {
1927 let (marker, mine) = take_marker(path, token, depth)?;
1928 let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
1929 && (!require_stale || older_than_stale(path));
1930 if still {
1931 let _ = std::fs::remove_file(path);
1932 }
1933 if depth >= MAX_MARKER_DEPTH {
1935 if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
1936 let _ = std::fs::remove_file(&marker);
1937 }
1938 } else {
1939 let _ = remove_lock_if(&marker, &mine, false, depth + 1);
1940 }
1941 Some(still)
1942}
1943
1944fn break_stale(path: &Path, judged: &str) -> bool {
1947 remove_lock_if(path, judged, true, 0).unwrap_or(false)
1948}
1949
1950impl Drop for TaskLock {
1951 fn drop(&mut self) {
1952 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
1953 loop {
1954 match remove_lock_if(&self.path, &self.token, false, 0) {
1955 Some(_) => return,
1956 None if std::time::Instant::now() > deadline => return,
1957 None => std::thread::sleep(std::time::Duration::from_millis(5)),
1958 }
1959 }
1960 }
1961}
1962
1963#[derive(Debug, Clone)]
1965pub struct Queue {
1966 root: PathBuf,
1967}
1968
1969impl Queue {
1970 pub fn open() -> Self {
1972 Self::at(crate::run::home().join("queue"))
1973 }
1974
1975 pub fn at(root: PathBuf) -> Self {
1978 Self { root }
1979 }
1980
1981 pub fn root(&self) -> &Path {
1983 &self.root
1984 }
1985
1986 pub fn path_of(&self, id: &str) -> PathBuf {
1988 self.root.join(format!("{id}.json"))
1989 }
1990
1991 pub fn attachments_dir(&self, id: &str) -> PathBuf {
1994 self.root.join(format!("{id}.attachments"))
1995 }
1996
1997 pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2008 let mut wanted = Vec::new();
2009 for src in sources {
2010 let name = src
2011 .file_name()
2012 .and_then(|n| n.to_str())
2013 .with_context(|| format!("`{}` has no usable file name", src.display()))?;
2014 if !crate::ask::valid_asset_name(name) {
2015 bail!(
2016 "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
2017 with no `..`; rename the file and try again"
2018 );
2019 }
2020 if !src.is_file() {
2021 bail!("attachment `{}` is not a file", src.display());
2022 }
2023 wanted.push((src, name));
2024 }
2025 let dir = self.attachments_dir(&task.id);
2026 let existed = dir.is_dir();
2027 std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
2028 let mut created: Vec<PathBuf> = Vec::new();
2029 let mut names = Vec::new();
2030 let mut copy_all = || -> Result<()> {
2031 for (src, name) in &wanted {
2032 let (stored, path) = copy_new(&dir, src, name)?;
2033 created.push(path);
2034 names.push(stored);
2035 }
2036 Ok(())
2037 };
2038 if let Err(e) = copy_all() {
2039 for path in &created {
2040 let _ = std::fs::remove_file(path);
2041 }
2042 if !existed {
2043 let _ = std::fs::remove_dir(&dir);
2044 }
2045 return Err(e);
2046 }
2047 task.attachments.extend(names.iter().cloned());
2048 Ok(names)
2049 }
2050
2051 pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2062 let dir = self.attachments_dir(&task.id);
2063 let existed = dir.is_dir();
2064 let before = task.attachments.len();
2065 let names = self.attach(task, sources)?;
2066 if let Err(e) = self.put(task) {
2067 for name in &names {
2068 let _ = std::fs::remove_file(dir.join(name));
2069 }
2070 if !existed {
2071 let _ = std::fs::remove_dir(&dir);
2072 }
2073 task.attachments.truncate(before);
2074 return Err(e);
2075 }
2076 Ok(names)
2077 }
2078
2079 pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
2083 let dir = self.attachments_dir(&task.id);
2084 task.attachments
2085 .iter()
2086 .map(|n| {
2087 let p = dir.join(n);
2088 std::path::absolute(&p).unwrap_or(p)
2089 })
2090 .collect()
2091 }
2092
2093 pub fn put(&self, task: &mut Task) -> Result<()> {
2102 let _lock = self.lock_task(&task.id)?;
2103 self.put_unlocked(task)
2104 }
2105
2106 pub fn create_new(&self, task: &mut Task) -> Result<bool> {
2111 let _lock = self.lock_task(&task.id)?;
2112 if self.path_of(&task.id).exists() {
2113 return Ok(false);
2114 }
2115 self.put_unlocked(task)?;
2116 Ok(true)
2117 }
2118
2119 fn put_unlocked(&self, task: &mut Task) -> Result<()> {
2121 if let Ok(mut stored) = read_path(&self.path_of(&task.id)) {
2122 stored.import_chat_notice();
2123 for key in stored.chat_notices {
2124 if !task.chat_notices.contains(&key) {
2125 task.chat_notices.push(key);
2126 }
2127 }
2128 for run in stored.runs {
2129 if !task.runs.contains(&run) {
2130 task.runs.push(run);
2131 }
2132 }
2133 for p in stored.passes {
2134 if !task
2135 .passes
2136 .iter()
2137 .any(|q| q.run == p.run && q.pass_index == p.pass_index)
2138 {
2139 task.passes.push(p);
2140 }
2141 }
2142 if let Some(n) = stored.chat_notice
2145 && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
2146 {
2147 task.chat_notice = Some(n);
2148 }
2149 }
2150 task.import_chat_notice();
2151 task.updated_at = Timestamp::now();
2152 std::fs::create_dir_all(&self.root)
2153 .with_context(|| format!("create {}", self.root.display()))?;
2154 let body = serde_json::to_string_pretty(task).context("serialize task")?;
2155 let path = self.path_of(&task.id);
2156 let tmp = path.with_extension("json.tmp");
2157 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
2158 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
2159 if let (Some(notice), Some(home)) = (
2164 crate::notices::task_held(task),
2165 self.root.parent().filter(|p| !p.as_os_str().is_empty()),
2166 ) {
2167 crate::notices::raise_in(home, notice);
2168 }
2169 Ok(())
2170 }
2171
2172 pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
2177 let id = self.resolve_id(id)?;
2178 let _lock = self.lock_task(&id)?;
2179 let mut task = self.get(&id)?;
2180 if !f(&mut task) {
2181 return Ok(false);
2182 }
2183 self.put_unlocked(&mut task)?;
2184 Ok(true)
2185 }
2186
2187 pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
2192 let id = self.resolve_id(id)?;
2193 let _lock = self.lock_task(&id)?;
2196 let mut task = self.get(&id)?;
2197 if task.link_run(run) {
2198 self.put_unlocked(&mut task)?;
2199 }
2200 Ok(task)
2201 }
2202
2203 fn lock_task(&self, id: &str) -> Result<TaskLock> {
2211 std::fs::create_dir_all(&self.root)
2212 .with_context(|| format!("create {}", self.root.display()))?;
2213 let path = self.root.join(format!("{id}.write-lock"));
2214 let started = std::time::Instant::now();
2215 loop {
2216 match std::fs::OpenOptions::new()
2217 .write(true)
2218 .create_new(true)
2219 .open(&path)
2220 {
2221 Ok(mut file) => {
2222 use std::io::Write;
2223 let token = owner_token();
2224 if let Err(e) = file.write_all(token.as_bytes()) {
2225 drop(file);
2226 let _ = std::fs::remove_file(&path);
2227 return Err(e).with_context(|| format!("lock {}", path.display()));
2228 }
2229 return Ok(TaskLock { path, token });
2230 }
2231 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2232 let judged = std::fs::read_to_string(&path).ok();
2235 let broken = judged
2236 .filter(|_| older_than_stale(&path))
2237 .is_some_and(|judged| break_stale(&path, &judged));
2238 if !broken {
2242 if started.elapsed() > TASK_LOCK_STALE {
2243 bail!("could not lock task {id}");
2244 }
2245 std::thread::sleep(std::time::Duration::from_millis(15));
2246 }
2247 }
2248 Err(e)
2252 if e.kind() == std::io::ErrorKind::PermissionDenied
2253 && started.elapsed() <= TASK_LOCK_STALE =>
2254 {
2255 std::thread::sleep(std::time::Duration::from_millis(15));
2256 }
2257 Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
2258 }
2259 }
2260 }
2261
2262 pub fn get(&self, id: &str) -> Result<Task> {
2264 let resolved = self.resolve_id(id)?;
2265 read_path(&self.path_of(&resolved))
2266 }
2267
2268 pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
2300 let resolved = self.resolve_id(id)?;
2301 if in_flight {
2302 bail!("task {resolved} is being run by a live daemon right now");
2303 }
2304 self.write_tombstone(&resolved)?;
2305 if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
2306 {
2307 if self.path_of(&resolved).exists() {
2311 let _ = std::fs::remove_file(self.tombstone_path(&resolved));
2312 }
2313 return Err(e);
2314 }
2315 let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
2316 Ok(Removal {
2317 id: resolved,
2318 released,
2319 still_blocked,
2320 })
2321 }
2322
2323 fn tombstone_path(&self, id: &str) -> PathBuf {
2326 self.root.join(format!("{id}.removed"))
2327 }
2328
2329 fn write_tombstone(&self, id: &str) -> Result<()> {
2330 let path = self.tombstone_path(id);
2331 let tmp = self.root.join(format!("{id}.removed.tmp"));
2332 std::fs::write(&tmp, b"")
2333 .and_then(|()| std::fs::rename(&tmp, &path))
2334 .with_context(|| format!("write {}", path.display()))
2335 }
2336
2337 fn was_deleted(&self, id: &str) -> bool {
2341 self.tombstone_path(id).is_file() && !self.path_of(id).exists()
2342 }
2343
2344 pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
2348 let deleted = deleted_blockers(self, questions, &task.blocked_by);
2349 deleted
2350 .into_iter()
2351 .filter(|id| task.dependency_deleted(id))
2352 .collect()
2353 }
2354
2355 pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
2359 let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
2360 return;
2361 };
2362 let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
2363 let waiting = dependent.status == TaskStatus::Blocked;
2364 let message = match (ja, waiting) {
2365 (true, false) => format!(
2366 "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
2367 dependent.short(),
2368 short(deleted)
2369 ),
2370 (true, true) => format!(
2371 "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
2372 dependent.short(),
2373 short(deleted)
2374 ),
2375 (false, false) => format!(
2376 "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
2377 dependent.short(),
2378 short(deleted)
2379 ),
2380 (false, true) => format!(
2381 "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
2382 dependent.short(),
2383 short(deleted)
2384 ),
2385 };
2386 crate::notices::raise_in(
2387 home,
2388 crate::notices::Notice::info(
2389 &format!("unblocked:{}:{}", dependent.id, deleted),
2390 message,
2391 )
2392 .link(crate::notices::Link::Task {
2393 id: dependent.id.clone(),
2394 }),
2395 );
2396 }
2397
2398 fn release_dependents_of(
2403 &self,
2404 dependency: &str,
2405 questions: &Questions,
2406 ) -> (Vec<String>, Vec<String>) {
2407 let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
2408 for listed in self.list() {
2409 if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
2410 matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
2411 }) {
2412 continue;
2413 }
2414 let Ok(_claim) = self.claim(&listed.id) else {
2415 continue;
2416 };
2417 let Ok(mut task) = self.get(&listed.id) else {
2418 continue;
2419 };
2420 let references: Vec<String> = task.blocked_by.iter().filter(|id| {
2421 matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
2422 }).cloned().collect();
2423 if references.is_empty() {
2424 continue;
2425 }
2426 for reference in references {
2427 task.dependency_deleted(&reference);
2428 }
2429 if self.put(&mut task).is_err() {
2430 continue;
2431 }
2432 self.note_dependency_deleted(&task, dependency);
2433 if task.status == TaskStatus::Blocked {
2434 still_blocked.push(task.id.clone());
2435 } else {
2436 released.push(task.id.clone());
2437 }
2438 }
2439 (released, still_blocked)
2440 }
2441
2442 fn remove_record_with_attachments(
2446 &self,
2447 resolved: &str,
2448 remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
2449 ) -> Result<()> {
2450 self.sweep_removed_attachments();
2457 let attachments = self.attachments_dir(resolved);
2458 let aside = self.root.join(format!("{resolved}.attachments.removing"));
2459 let moved = match std::fs::rename(&attachments, &aside) {
2460 Ok(()) => true,
2461 Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
2462 Err(e) => {
2463 return Err(e).with_context(|| format!("remove {}", attachments.display()));
2464 }
2465 };
2466 let path = self.path_of(resolved);
2467 if let Err(e) = remove_record(&path) {
2468 if moved {
2469 let _ = std::fs::rename(&aside, &attachments);
2470 }
2471 return Err(e).with_context(|| format!("remove {}", path.display()));
2472 }
2473 if moved {
2474 if let Err(e) = std::fs::remove_dir_all(&aside) {
2475 tracing::warn!("leftover attachments {}: {e}", aside.display());
2476 }
2477 }
2478 let lock = self.lock_path(resolved);
2479 if let Err(e) = std::fs::remove_file(&lock) {
2480 if e.kind() != std::io::ErrorKind::NotFound {
2481 return Err(e).with_context(|| format!("remove {}", lock.display()));
2482 }
2483 }
2484 Ok(())
2485 }
2486
2487 fn sweep_removed_attachments(&self) {
2493 let Ok(entries) = std::fs::read_dir(&self.root) else {
2494 return;
2495 };
2496 for entry in entries.flatten() {
2497 let name = entry.file_name();
2498 let name = name.to_string_lossy();
2499 let Some(id) = name.strip_suffix(".attachments.removing") else {
2500 continue;
2501 };
2502 if !self.path_of(id).exists() {
2505 if let Err(e) = std::fs::remove_dir_all(entry.path()) {
2506 tracing::warn!("leftover attachments {}: {e}", entry.path().display());
2507 }
2508 }
2509 }
2510 self.sweep_tombstones();
2511 }
2512
2513 fn sweep_tombstones(&self) {
2522 let Ok(entries) = std::fs::read_dir(&self.root) else {
2523 return;
2524 };
2525 let tasks = self.list();
2526 for entry in entries.flatten() {
2527 let name = entry.file_name();
2528 let name = name.to_string_lossy();
2529 let Some(id) = name.strip_suffix(".removed") else {
2530 continue;
2531 };
2532 if self.path_of(id).exists()
2533 || tasks.iter().any(|t| {
2534 t.blocked_by
2535 .iter()
2536 .any(|b| id.starts_with(b) || id.ends_with(b))
2537 })
2538 {
2539 continue;
2540 }
2541 let old = entry
2542 .metadata()
2543 .and_then(|m| m.modified())
2544 .ok()
2545 .and_then(|m| m.elapsed().ok())
2546 .is_some_and(|age| age >= TOMBSTONE_GRACE);
2547 if old {
2548 let _ = std::fs::remove_file(entry.path());
2549 }
2550 }
2551 }
2552
2553 fn lock_path(&self, id: &str) -> PathBuf {
2556 self.root.join(format!("{id}.lock"))
2557 }
2558
2559 pub fn list(&self) -> Vec<Task> {
2572 let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
2573 .into_iter()
2574 .flatten()
2575 .flatten()
2576 .map(|e| e.path())
2577 .filter(|p| p.extension().is_some_and(|x| x == "json"))
2578 .filter_map(|p| read_path(&p).ok())
2579 .collect();
2580 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
2581 tasks
2582 }
2583
2584 pub fn superseded(&self) -> HashMap<String, String> {
2597 let mut by = HashMap::new();
2598 for task in self.list() {
2599 for earlier in &task.runs {
2600 if let Some(later) = task.successor_of(earlier) {
2601 by.insert(earlier.clone(), later.clone());
2602 }
2603 }
2604 }
2605 by
2606 }
2607
2608 pub fn superseded_by(&self, run: &str) -> Option<String> {
2616 for task in self.list() {
2617 if task.runs.iter().any(|r| r == run) {
2618 return task.successor_of(run).cloned();
2619 }
2620 }
2621 None
2622 }
2623
2624 pub fn latest_attempt(&self, run: &str) -> Option<String> {
2635 for task in self.list() {
2636 if task.runs.iter().any(|r| r == run) {
2637 return task.runs.last().filter(|last| **last != run).cloned();
2638 }
2639 }
2640 None
2641 }
2642
2643 pub fn next_runnable(&self) -> Option<Task> {
2648 let mut runnable: Vec<Task> = self
2649 .list()
2650 .into_iter()
2651 .filter(|t| t.status.runnable())
2652 .collect();
2653 runnable.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
2654 runnable.into_iter().next()
2655 }
2656
2657 pub fn claim(&self, id: &str) -> Result<Claim> {
2664 std::fs::create_dir_all(&self.root)
2665 .with_context(|| format!("create {}", self.root.display()))?;
2666 let path = self.lock_path(id);
2667 match std::fs::OpenOptions::new()
2668 .write(true)
2669 .create_new(true)
2670 .open(&path)
2671 {
2672 Ok(mut f) => {
2673 use std::io::Write as _;
2674 let _ = writeln!(f, "{}", std::process::id());
2676 Ok(Claim { path })
2677 }
2678 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2679 bail!("task {id} is already claimed ({} exists)", path.display())
2680 }
2681 Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
2682 }
2683 }
2684
2685 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
2687 let ids = record_ids(&self.root, "json")?;
2688 match resolve_reference(prefix, ids.iter().map(String::as_str)) {
2689 Reference::Found(id) => Ok(id),
2690 Reference::Missing => bail!("no task matches `{prefix}`"),
2691 Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
2692 }
2693 }
2694
2695 pub fn revision(&self) -> u64 {
2702 self.revision_excluding(&std::collections::BTreeSet::new())
2703 }
2704
2705 pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
2708 use std::hash::{Hash as _, Hasher as _};
2709
2710 let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
2711 .into_iter()
2712 .flatten()
2713 .flatten()
2714 .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
2715 .filter(|e| {
2716 let path = e.path();
2717 !path
2718 .file_stem()
2719 .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
2720 })
2721 .filter_map(|e| {
2722 let name = e.file_name().to_string_lossy().into_owned();
2723 let mtime = e
2724 .metadata()
2725 .ok()?
2726 .modified()
2727 .ok()?
2728 .duration_since(std::time::UNIX_EPOCH)
2729 .ok()?
2730 .as_millis() as u64;
2731 Some((name, mtime))
2732 })
2733 .collect();
2734
2735 if entries.is_empty() {
2736 return 0;
2737 }
2738
2739 entries.sort_unstable();
2740 let mut hasher = std::hash::DefaultHasher::new();
2741 for (name, mtime) in &entries {
2742 name.hash(&mut hasher);
2743 mtime.hash(&mut hasher);
2744 }
2745 let h = hasher.finish();
2746 if h == 0 { 1 } else { h }
2747 }
2748}
2749
2750const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
2753
2754#[derive(Debug, Clone)]
2756pub struct Removal {
2757 pub id: String,
2759 pub released: Vec<String>,
2763 pub still_blocked: Vec<String>,
2766}
2767
2768#[derive(Debug)]
2770pub struct Claim {
2771 path: PathBuf,
2772}
2773
2774impl Drop for Claim {
2775 fn drop(&mut self) {
2776 let _ = std::fs::remove_file(&self.path);
2777 }
2778}
2779
2780pub fn title_from(instruction: &str, max: usize) -> String {
2783 let Some(line) = first_line(instruction) else {
2784 return "(empty task)".to_owned();
2785 };
2786 if line.chars().count() <= max {
2787 return line.to_owned();
2788 }
2789 let head: String = line.chars().take(max.saturating_sub(1)).collect();
2790 format!("{head}…")
2791}
2792
2793pub(crate) fn first_line(instruction: &str) -> Option<&str> {
2799 let line = instruction
2800 .lines()
2801 .map(str::trim)
2802 .find(|l| !l.is_empty())?
2803 .trim_start_matches(['#', '-', '*', '>', ' '])
2804 .trim();
2805 (!line.is_empty()).then_some(line)
2806}
2807
2808fn read_path(path: &Path) -> Result<Task> {
2809 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
2810 let task: Task =
2811 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2812 if task.schema > SCHEMA {
2818 bail!(
2819 "task {} was written by a different magi (schema {}, this build \
2820 speaks {SCHEMA})",
2821 task.id,
2822 task.schema
2823 );
2824 }
2825 Ok(task)
2826}
2827
2828#[derive(Debug, PartialEq, Eq)]
2831pub(crate) enum Reference {
2832 Found(String),
2833 Missing,
2834 Ambiguous,
2835}
2836
2837pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
2838 let mut hits = Vec::new();
2839 for candidate in ids {
2840 if candidate == id {
2841 return Reference::Found(candidate.to_owned());
2842 }
2843 if candidate.starts_with(id) || candidate.ends_with(id) {
2844 hits.push(candidate);
2845 }
2846 }
2847 match hits.as_slice() {
2848 [] => Reference::Missing,
2849 [one] => Reference::Found((*one).to_owned()),
2850 _ => Reference::Ambiguous,
2851 }
2852}
2853
2854pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
2857 let entries = match std::fs::read_dir(root) {
2858 Ok(entries) => entries,
2859 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
2860 Err(e) => return Err(e.into()),
2861 };
2862 let mut ids = Vec::new();
2863 for entry in entries {
2864 let path = entry?.path();
2865 if path.extension().is_some_and(|e| e == extension)
2866 && let Some(id) = path.file_stem().and_then(|s| s.to_str())
2867 {
2868 ids.push(id.to_owned());
2869 }
2870 }
2871 Ok(ids)
2872}
2873
2874#[derive(Debug, PartialEq, Eq)]
2877pub enum Dependency {
2878 Task(String),
2880 Question(String),
2882 Deleted(String),
2884 Missing,
2886 Ambiguous,
2888 Unknown,
2890}
2891
2892pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
2894 let (Ok(tasks), Ok(qs), Ok(removed)) = (
2895 record_ids(queue.root(), "json"),
2896 record_ids(questions.root(), "json"),
2897 record_ids(queue.root(), "removed"),
2898 ) else {
2899 return Dependency::Unknown;
2900 };
2901 let mut ids = tasks
2904 .iter()
2905 .chain(qs.iter())
2906 .chain(removed.iter())
2907 .cloned()
2908 .collect::<Vec<_>>();
2909 ids.sort();
2910 ids.dedup();
2911 match resolve_reference(id, ids.iter().map(String::as_str)) {
2912 Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
2913 Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
2914 Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
2915 Reference::Found(_) => Dependency::Unknown,
2916 Reference::Missing => Dependency::Missing,
2917 Reference::Ambiguous => Dependency::Ambiguous,
2918 }
2919}
2920
2921pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
2923 for id in ids {
2924 if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
2925 dependency(queue, questions, id)
2926 {
2927 *id = full;
2928 }
2929 }
2930}
2931
2932pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
2935 let inventory = crate::blockers::Inventory::load(queue, questions);
2936 let mut text = String::from(
2937 "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
2938 );
2939 match record_ids(queue.root(), "json") {
2940 Ok(ids) => {
2941 for id in ids {
2942 match queue.get(&id) {
2943 Ok(task) => text.push_str(&format!(
2944 "{}: {}; blocked_by={:?}; chain={:?}\n",
2945 task.id,
2946 task.status.as_str(),
2947 task.blocked_by,
2948 inventory.waits_on(&task)
2949 )),
2950 Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
2951 }
2952 }
2953 }
2954 Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
2955 }
2956 text
2957}
2958
2959pub fn missing_blockers(
2961 queue: &Queue,
2962 questions: &Questions,
2963 blocked_by: &[String],
2964) -> Vec<String> {
2965 blocked_by
2966 .iter()
2967 .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
2968 .cloned()
2969 .collect()
2970}
2971
2972pub fn deleted_blockers(
2975 queue: &Queue,
2976 questions: &Questions,
2977 blocked_by: &[String],
2978) -> Vec<String> {
2979 blocked_by
2980 .iter()
2981 .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
2982 .cloned()
2983 .collect()
2984}
2985
2986pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
2998 missing_blocker_hold_reason_in(blocked_by, missing, "en")
2999}
3000
3001pub fn missing_blocker_hold_reason_in(
3003 blocked_by: &[String],
3004 missing: &[String],
3005 language: &str,
3006) -> String {
3007 if crate::lang::is_japanese(language) {
3008 format!(
3009 "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
3010 blocked_by.join(", "),
3011 missing.join(", "),
3012 )
3013 } else {
3014 format!(
3015 "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
3016 blocked_by.join(", "),
3017 missing.join(", "),
3018 )
3019 }
3020}
3021
3022pub fn short(id: &str) -> &str {
3024 id.split('-').next_back().unwrap_or(id)
3025}
3026
3027fn new_id() -> String {
3028 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
3029 let seed = crate::rng::entropy();
3030 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
3031}
3032
3033#[cfg(test)]
3034mod tests {
3035 #[test]
3036 fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
3037 let t = Task::new(
3038 "retry".into(),
3039 "do it".into(),
3040 PathBuf::from("."),
3041 Source::Human,
3042 );
3043 let mut value = serde_json::to_value(&t).unwrap();
3044 for field in ["resumption", "resumption_history", "operator_decided_at"] {
3045 value.as_object_mut().unwrap().remove(field);
3046 }
3047 value["schema"] = serde_json::json!(15);
3048 let mut legacy: Task = serde_json::from_value(value).unwrap();
3049 legacy.hold_machine(Some("automatic retry".into()));
3050 legacy.release();
3051 assert!(legacy.resumption.is_none());
3052 assert!(legacy.resumption_history.is_empty());
3053 assert!(legacy.operator_decided_at.is_none());
3054 legacy.release_operator("cli", None);
3055 legacy.start("run".into());
3056 let reloaded: Task =
3057 serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
3058 assert!(reloaded.resume_override.is_none());
3059 assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
3060 }
3061
3062 #[test]
3063 fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
3064 let mut base = RunOverrides {
3065 merge: Some("pr".to_owned()),
3066 candidates: Some(3),
3067 ..RunOverrides::default()
3068 };
3069 base.merge_over(&RunOverrides {
3070 merge: Some("none".to_owned()),
3071 seed: Some(7),
3072 ..RunOverrides::default()
3073 });
3074 assert_eq!(base.merge.as_deref(), Some("none"));
3075 assert_eq!(base.candidates, Some(3));
3076 assert_eq!(base.seed, Some(7));
3077 }
3078
3079 #[test]
3080 fn an_already_landed_task_is_done_with_its_attempt_refunded() {
3081 let mut t = task("relanded");
3082 t.attempts = 1;
3083 t.status = TaskStatus::Running;
3084 t.already_landed("already in main as 0e368de");
3085 assert_eq!(t.status, TaskStatus::Done);
3086 assert_eq!(t.attempts, 0);
3087 assert!(t.hold_reason.is_none());
3088 assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
3089 }
3090
3091 #[test]
3092 fn missing_blocker_reason_follows_the_language() {
3093 let b = vec!["a".to_owned()];
3094 let en = missing_blocker_hold_reason_in(&b, &b, "en");
3095 assert_eq!(en, missing_blocker_hold_reason(&b, &b));
3096 assert!(en.starts_with("blocked on a"));
3097 assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
3098 assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
3099 }
3100
3101 use super::*;
3102
3103 #[test]
3104 fn triage_applied_survives_release_and_old_records_read_as_empty() {
3105 let mut t = Task::new(
3106 "t".to_owned(),
3107 "i".to_owned(),
3108 PathBuf::from("r"),
3109 Source::Human,
3110 );
3111 t.mark_triage_applied("q1");
3112 t.mark_triage_applied("q1");
3113 t.hold_machine(Some("x".to_owned()));
3114 t.release();
3115 assert_eq!(t.triage_applied, ["q1"]);
3116 assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
3117
3118 let mut v = serde_json::to_value(&t).unwrap();
3119 v.as_object_mut().unwrap().remove("triage_applied");
3120 let old: Task = serde_json::from_value(v).unwrap();
3121 assert!(old.triage_applied.is_empty());
3122 }
3123
3124 #[test]
3125 fn task_counts_of_empty_is_all_zero() {
3126 assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
3127 }
3128
3129 #[test]
3130 fn task_counts_of_tallies_every_status() {
3131 let mut queued = Task::new(
3132 "q".to_owned(),
3133 "i".to_owned(),
3134 PathBuf::from("."),
3135 Source::Human,
3136 );
3137 queued.status = TaskStatus::Queued;
3138 let mut running = queued.clone();
3139 running.status = TaskStatus::Running;
3140 let mut done = queued.clone();
3141 done.status = TaskStatus::Done;
3142 let mut failed = queued.clone();
3143 failed.status = TaskStatus::Failed;
3144 let mut held = queued.clone();
3145 held.status = TaskStatus::Held;
3146 let mut blocked = queued.clone();
3147 blocked.status = TaskStatus::Blocked;
3148
3149 let mut parked = queued.clone();
3150 parked.status = TaskStatus::Parked;
3151
3152 let counts = TaskCounts::of(&[
3153 queued,
3154 running,
3155 done.clone(),
3156 done,
3157 failed,
3158 held,
3159 blocked,
3160 parked,
3161 ]);
3162 assert_eq!(
3163 counts,
3164 TaskCounts {
3165 queued: 1,
3166 running: 1,
3167 done: 2,
3168 failed: 1,
3169 held: 1,
3170 blocked: 1,
3171 parked: 1,
3172 }
3173 );
3174 }
3175
3176 #[test]
3177 fn task_implementers_persist_normalized_without_losing_history() {
3178 let (_dir, q) = queue();
3179 let mut t = task("keep me");
3180 t.hold_manual(Some("waiting".into()));
3181 t.runs.push("earlier-run".into());
3182 t.attachments.push("reference.txt".into());
3183 let cfg: crate::config::Config =
3184 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
3185 .unwrap();
3186 q.put(&mut t).unwrap();
3187 let original = serde_json::to_value(&t).unwrap();
3188 for n in [2, 1, 2] {
3189 t.set_implementers(n, &cfg).unwrap();
3190 q.put(&mut t).unwrap();
3191 t = q.get(&t.id).unwrap();
3192 assert_eq!(t.implementers(), Some(usize::from(n)));
3193 assert_eq!(t.solo, n == 1);
3194 if n == 1 {
3195 assert!(t.overrides.is_none());
3196 }
3197 let mut actual = serde_json::to_value(&t).unwrap();
3198 for field in ["solo", "overrides", "updated_at"] {
3199 actual[field] = original[field].clone();
3200 }
3201 assert_eq!(actual, original);
3202 }
3203 t.overrides.as_mut().unwrap().judges = Some(2);
3204 t.set_implementers(1, &cfg).unwrap();
3205 assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
3206 assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
3207 }
3208
3209 #[test]
3210 fn task_implementers_old_records_keep_inheritance_until_changed() {
3211 let mut value = serde_json::to_value(task("old task")).unwrap();
3212 value.as_object_mut().unwrap().remove("solo");
3213 value.as_object_mut().unwrap().remove("overrides");
3214 let mut old: Task = serde_json::from_value(value).unwrap();
3215 assert_eq!(old.implementers(), None);
3216 old.set_implementers(1, &crate::config::Config::default())
3217 .unwrap();
3218 assert!(old.solo);
3219 assert!(old.overrides.is_none());
3220 }
3221
3222 #[test]
3223 fn task_implementers_reject_invalid_counts_and_states() {
3224 let cfg = crate::config::Config::default();
3225 let mut t = task("unchanged");
3226 for n in [0, 4, 255] {
3227 assert!(t.set_implementers(n, &cfg).is_err());
3228 }
3229 for status in [
3230 TaskStatus::Running,
3231 TaskStatus::Done,
3232 TaskStatus::Failed,
3233 TaskStatus::Blocked,
3234 TaskStatus::Parked,
3235 ] {
3236 t.status = status;
3237 let before = serde_json::to_value(&t).unwrap();
3238 assert!(t.set_implementers(1, &cfg).is_err());
3239 assert_eq!(serde_json::to_value(&t).unwrap(), before);
3240 }
3241 }
3242
3243 fn queue() -> (tempfile::TempDir, Queue) {
3246 let dir = tempfile::tempdir().unwrap();
3247 let q = Queue::at(dir.path().join("queue"));
3248 (dir, q)
3249 }
3250
3251 #[test]
3252 fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
3253 let dir = tempfile::tempdir().unwrap();
3254 let q = Queue::at(dir.path().join("queue"));
3255 let mut t = task("held");
3256 q.put(&mut t).unwrap();
3257 assert_eq!(
3258 crate::notices::Notices::at(dir.path().join("notifications"))
3259 .list()
3260 .len(),
3261 0
3262 );
3263 t.hold_machine(Some("out of attempts".to_owned()));
3264 q.put(&mut t).unwrap();
3265 let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
3266 assert_eq!(listed.len(), 1);
3267 assert!(listed[0].message.contains("out of attempts"));
3268 }
3269
3270 fn task(title: &str) -> Task {
3271 Task::new(
3272 title.to_owned(),
3273 format!("do {title}"),
3274 PathBuf::from("."),
3275 Source::Human,
3276 )
3277 }
3278
3279 #[test]
3280 fn attempt_limit_inherits_old_records_and_controls_failure() {
3281 let mut t = task("budget");
3282 let mut old = serde_json::to_value(&t).unwrap();
3283 old.as_object_mut().unwrap().remove("attempt_limit");
3284 old["schema"] = serde_json::json!(15);
3285 let legacy: Task = serde_json::from_value(old).unwrap();
3286 assert_eq!(legacy.attempt_limit, None);
3287 assert_eq!(legacy.effective_attempt_limit(2), 2);
3288 t.attempts = 2;
3289 t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
3290 t.fail("retry", 1);
3291 assert_eq!(t.status, TaskStatus::Failed);
3292 t.attempts = 3;
3293 t.fail("stop", 10);
3294 assert_eq!(t.status, TaskStatus::Held);
3295 t.change_attempt_limit(None).unwrap();
3296 assert_eq!(t.status, TaskStatus::Held);
3297 assert_eq!(t.attempts, 3);
3298 t.fail("inherit", 4);
3299 assert_eq!(t.status, TaskStatus::Failed);
3300 for status in [
3301 TaskStatus::Running,
3302 TaskStatus::Done,
3303 TaskStatus::Failed,
3304 TaskStatus::Blocked,
3305 TaskStatus::Parked,
3306 ] {
3307 t.status = status;
3308 assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
3309 assert_eq!(t.attempt_limit, None);
3310 }
3311 }
3312
3313 #[test]
3314 fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
3315 let mut t = task("retried");
3316 assert!(
3317 t.earlier_attempts().is_empty(),
3318 "a first attempt takes nothing over"
3319 );
3320 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3321 assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
3322 assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
3324 assert_eq!(t.successor_of("bbbb"), None);
3325 assert_eq!(t.successor_of("zzzz"), None);
3326 }
3327
3328 #[test]
3329 fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
3330 let (_dir, q) = queue();
3331 let mut t = task("resumed");
3332 for (runs, successor) in [
3333 (vec!["aaaa", "aaaa"], None),
3334 (vec!["aaaa", "bbbb", "aaaa"], None),
3335 (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
3336 (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
3337 ] {
3338 t.runs = runs.into_iter().map(str::to_owned).collect();
3339 q.put(&mut t).unwrap();
3340
3341 assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
3342 assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
3343 assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
3344 }
3345 }
3346
3347 #[test]
3348 fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
3349 let (_dir, q) = queue();
3350 let mut t = task("retried");
3351 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3352 q.put(&mut t).unwrap();
3353
3354 assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
3355 assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
3356 assert_eq!(
3357 q.superseded_by("cccc"),
3358 None,
3359 "the latest attempt replaces nothing"
3360 );
3361 assert_eq!(
3362 q.superseded_by("never-heard-of-it"),
3363 None,
3364 "a run belonging to no task on this queue is not superseded"
3365 );
3366
3367 let mut by = HashMap::new();
3368 by.insert("aaaa".to_owned(), "bbbb".to_owned());
3369 by.insert("bbbb".to_owned(), "cccc".to_owned());
3370 assert_eq!(
3371 q.superseded(),
3372 by,
3373 "the whole-map and single-run forms must agree"
3374 );
3375 }
3376
3377 #[test]
3378 fn superseded_attempts_is_empty_until_the_task_is_done() {
3379 let mut t = task("retried");
3380 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3381 t.status = TaskStatus::Failed;
3382 assert_eq!(
3383 t.superseded_attempts(true),
3384 &[] as &[String],
3385 "a task still retrying has no attempt yet that a later one made moot"
3386 );
3387
3388 t.status = TaskStatus::Running;
3389 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3390 }
3391
3392 #[test]
3393 fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
3394 let mut t = task("retried");
3395 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3396 t.status = TaskStatus::Done;
3397 assert_eq!(
3398 t.superseded_attempts(true),
3399 &["aaaa".to_owned(), "bbbb".to_owned()],
3400 "cccc is the attempt whose success made the task done, and stays out"
3401 );
3402 }
3403
3404 #[test]
3405 fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
3406 let mut t = task("first try landed");
3407 t.runs = vec!["aaaa".to_owned()];
3408 t.status = TaskStatus::Done;
3409 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3410 }
3411
3412 #[test]
3413 fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
3414 let mut t = task("closed by hand after a manual merge");
3421 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3422 t.status = TaskStatus::Done;
3423 assert_eq!(
3424 t.superseded_attempts(false),
3425 &[] as &[String],
3426 "nothing here is provably why the task is done, so nothing is superseded"
3427 );
3428 }
3429
3430 #[test]
3431 fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
3432 let (_dir, q) = queue();
3433 let mut t = task("retried twice");
3434 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3435 q.put(&mut t).unwrap();
3436
3437 assert_eq!(
3438 q.latest_attempt("aaaa"),
3439 Some("cccc".to_owned()),
3440 "an old attempt points straight at the chain's current head, not the \
3441 next attempt in the middle of it"
3442 );
3443 assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
3444 assert_eq!(
3445 q.latest_attempt("cccc"),
3446 None,
3447 "the latest attempt is not superseded by anything"
3448 );
3449 assert_eq!(
3450 q.latest_attempt("never-heard-of-it"),
3451 None,
3452 "a run belonging to no task on this queue is not superseded"
3453 );
3454 }
3455
3456 #[test]
3457 fn a_markdown_heading_is_the_title_not_decoration() {
3458 assert_eq!(
3463 title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
3464 "Rework the config loader"
3465 );
3466 assert_eq!(title_from("- fix the thing", 40), "fix the thing");
3467 assert_eq!(title_from("> quoted task", 40), "quoted task");
3468 assert_eq!(title_from(" \n\n", 40), "(empty task)");
3470 assert_eq!(title_from("###\n", 40), "(empty task)");
3471 }
3472
3473 #[test]
3474 fn a_long_title_is_elided_by_characters_not_bytes() {
3475 let long = "課題".repeat(30);
3477 let title = title_from(&long, 10);
3478 assert_eq!(title.chars().count(), 10);
3479 assert!(title.ends_with('…'));
3480 }
3481
3482 #[test]
3483 fn priority_wins_and_ties_break_oldest_first() {
3484 let (_dir, q) = queue();
3485 let mut a = task("first");
3486 let mut b = task("second");
3487 let mut c = task("urgent");
3488 a.id = "20260101-000001-aaaa".to_owned();
3490 b.id = "20260101-000002-bbbb".to_owned();
3491 c.id = "20260101-000003-cccc".to_owned();
3492 c.priority = 5;
3493 for t in [&mut a, &mut b, &mut c] {
3494 q.put(t).unwrap();
3495 }
3496
3497 assert_eq!(q.next_runnable().unwrap().id, c.id);
3499 c.hold_machine(None);
3500 q.put(&mut c).unwrap();
3501 assert_eq!(q.next_runnable().unwrap().id, a.id);
3503 assert_eq!(q.list().len(), 3, "b is still waiting its turn");
3504 }
3505
3506 #[test]
3507 fn a_blocked_task_never_starves_another_runnable_one() {
3508 let (_dir, q) = queue();
3509 let mut blocked = task("blocked");
3510 blocked.block(vec!["something".to_owned()], None);
3511 q.put(&mut blocked).unwrap();
3512
3513 let mut runnable = task("free to go");
3514 q.put(&mut runnable).unwrap();
3515
3516 let next = q.next_runnable().expect("a runnable task is still offered");
3517 assert_eq!(next.id, runnable.id);
3518 }
3519
3520 #[test]
3521 fn a_held_task_is_never_offered_to_the_loop() {
3522 let (_dir, q) = queue();
3523 let mut t = task("held");
3524 q.put(&mut t).unwrap();
3525 assert!(q.next_runnable().is_some());
3526
3527 t.hold_machine(None);
3528 q.put(&mut t).unwrap();
3529 assert!(
3530 q.next_runnable().is_none(),
3531 "a held task must wait for a human"
3532 );
3533
3534 t.status = TaskStatus::Failed;
3536 q.put(&mut t).unwrap();
3537 assert!(q.next_runnable().is_some());
3538 }
3539
3540 #[test]
3541 fn attempts_are_capped_and_then_the_task_is_held() {
3542 let mut t = task("doomed");
3543
3544 t.start("run-1".to_owned());
3545 t.fail("gate red", 2);
3546 assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
3547
3548 t.start("run-2".to_owned());
3549 t.fail("gate red", 2);
3550 assert_eq!(
3551 t.status,
3552 TaskStatus::Held,
3553 "out of attempts: stop spending money on it"
3554 );
3555 assert_eq!(t.runs, ["run-1", "run-2"]);
3556 assert_eq!(t.last_error.as_deref(), Some("gate red"));
3557 assert_eq!(
3558 t.hold_reason.as_deref(),
3559 Some("gate red"),
3560 "the hold must say why, not leave hold_reason null next to a \
3561 populated last_error"
3562 );
3563 }
3564
3565 #[test]
3566 fn handing_off_a_task_records_a_hold_reason_too() {
3567 let mut t = task("left a pull request");
3568 t.start("run-1".to_owned());
3569 t.handed_off("run ended with a pull request open [run run-1]");
3570 assert_eq!(t.status, TaskStatus::Held);
3571 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3572 assert_eq!(
3573 t.hold_reason.as_deref(),
3574 Some("run ended with a pull request open [run run-1]")
3575 );
3576 assert_eq!(t.hold_reason, t.last_error);
3577 }
3578
3579 #[test]
3580 fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
3581 let mut t = task("stalled by quota");
3582
3583 t.start("run-1".to_owned());
3584 assert_eq!(t.attempts, 1);
3585 t.stall("judge-1, judge-2 out of quota");
3586 assert_eq!(
3587 t.attempts, 0,
3588 "a closed quota window must not spend the task's retry budget"
3589 );
3590 assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
3591 assert_eq!(
3592 t.last_error.as_deref(),
3593 Some("judge-1, judge-2 out of quota")
3594 );
3595
3596 for _ in 0..20 {
3599 t.start("run-n".to_owned());
3600 t.stall("still out of quota");
3601 }
3602 t.start("run-real".to_owned());
3603 t.fail("gate red", 2);
3604 assert_eq!(
3605 t.status,
3606 TaskStatus::Failed,
3607 "the first attempt that was really judged is attempt one"
3608 );
3609 }
3610
3611 #[test]
3612 fn releasing_a_held_task_gives_it_a_real_second_chance() {
3613 let mut t = task("retry me");
3614 t.start("run-1".to_owned());
3615 t.fail("gate red", 1);
3616 assert_eq!(t.status, TaskStatus::Held);
3617
3618 t.release();
3619 assert_eq!(t.status, TaskStatus::Queued);
3620 assert_eq!(t.attempts, 0);
3623 assert!(t.last_error.is_none());
3624 assert_eq!(
3625 t.runs.len(),
3626 1,
3627 "history is kept: attempts reset, evidence does not"
3628 );
3629 }
3630
3631 #[test]
3632 fn a_hold_reason_survives_and_a_release_clears_it() {
3633 let mut t = task("waiting on something else");
3634 t.hold_manual(Some(
3635 "waiting for 20260101-000000-aaaa to land first".to_owned(),
3636 ));
3637 assert_eq!(t.status, TaskStatus::Held);
3638 assert_eq!(
3639 t.hold_reason.as_deref(),
3640 Some("waiting for 20260101-000000-aaaa to land first")
3641 );
3642
3643 t.hold_manual(None);
3645 assert_eq!(
3646 t.hold_reason.as_deref(),
3647 Some("waiting for 20260101-000000-aaaa to land first"),
3648 "a bare re-hold keeps whatever a human already wrote down"
3649 );
3650
3651 let mut plain = task("no reason given");
3653 plain.hold_manual(None);
3654 assert_eq!(plain.status, TaskStatus::Held);
3655 assert!(plain.hold_reason.is_none());
3656
3657 t.release();
3658 assert_eq!(t.status, TaskStatus::Queued);
3659 assert!(
3660 t.hold_reason.is_none(),
3661 "a stale reason must not greet the next person who holds this task"
3662 );
3663 }
3664
3665 #[test]
3666 fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
3667 let mut t = task("landed by hand while held");
3672 t.hold_manual(Some("waiting on 3ed9".to_owned()));
3673 assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
3674
3675 t.succeed();
3676 assert_eq!(t.status, TaskStatus::Done);
3677 assert!(
3678 t.hold_reason.is_none(),
3679 "a done task cannot still be waiting on something"
3680 );
3681 }
3682
3683 #[test]
3684 fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
3685 let mut held = task("held straight out of blocked");
3692 held.block(
3693 vec!["20260101-000000-dead".to_owned()],
3694 Some("waiting on the migration script".to_owned()),
3695 );
3696 assert_eq!(held.status, TaskStatus::Blocked);
3697
3698 held.hold_manual(None);
3699 assert_eq!(held.status, TaskStatus::Held);
3700 assert!(
3701 held.blocked_by.is_empty(),
3702 "hold overrides the wait, same as release"
3703 );
3704 assert!(held.block_reason.is_none());
3705
3706 let mut done = task("closed straight out of blocked");
3707 done.block(
3708 vec!["20260101-000000-dead".to_owned()],
3709 Some("waiting on the migration script".to_owned()),
3710 );
3711 done.succeed();
3712 assert_eq!(done.status, TaskStatus::Done);
3713 assert!(
3714 done.blocked_by.is_empty(),
3715 "a done task cannot still be waiting on a dependency"
3716 );
3717 assert!(done.block_reason.is_none());
3718 }
3719
3720 #[test]
3721 fn a_blocked_task_is_never_offered_to_the_loop() {
3722 let mut t = task("blocked");
3723 assert!(t.status.runnable());
3724 t.block(
3725 vec!["dep-id".to_owned()],
3726 Some("waits on dep-id".to_owned()),
3727 );
3728 assert_eq!(t.status, TaskStatus::Blocked);
3729 assert!(!t.status.runnable());
3730 assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
3731 }
3732
3733 #[test]
3734 fn unblocking_the_last_dependency_returns_the_task_to_queued() {
3735 let mut t = task("blocked on two");
3736 t.block(
3737 vec!["a".to_owned(), "b".to_owned()],
3738 Some("waits on a and b".to_owned()),
3739 );
3740
3741 t.unblock("a");
3742 assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
3743 assert_eq!(t.blocked_by, ["b"]);
3744
3745 t.unblock("b");
3746 assert_eq!(t.status, TaskStatus::Queued);
3747 assert!(t.blocked_by.is_empty());
3748 assert!(t.block_reason.is_none());
3749 }
3750
3751 #[test]
3752 fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
3753 let mut t = task("never blocked");
3754 t.unblock("whatever");
3755 assert_eq!(t.status, TaskStatus::Queued);
3756 }
3757
3758 #[test]
3759 fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
3760 let mut t = task("held, then asked about");
3765 t.hold_machine(Some("out of attempts".to_owned()));
3766 assert_eq!(t.status, TaskStatus::Held);
3767
3768 t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
3769 assert_eq!(t.status, TaskStatus::Blocked);
3770
3771 t.record_answer("what now?".to_owned(), "leave it held".to_owned());
3772 t.unblock("q1");
3773 assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
3774 assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
3775 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3776 assert!(t.blocked_from.is_none(), "consumed once restored");
3777 }
3778
3779 #[test]
3780 fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
3781 let mut t = task("manually held, then asked about");
3782 t.hold_manual(Some("waiting on a dependency".to_owned()));
3783
3784 t.block(vec!["q1".to_owned()], None);
3785 t.unblock("q1");
3786
3787 assert_eq!(t.status, TaskStatus::Held);
3788 assert_eq!(t.hold_source, Some(HoldSource::Manual));
3789 }
3790
3791 #[test]
3792 fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
3793 let mut t = task("held, blocked twice");
3797 t.hold_machine(None);
3798 t.block(vec!["q1".to_owned()], Some("first".to_owned()));
3799 t.block(
3800 vec!["q1".to_owned(), "q2".to_owned()],
3801 Some("second".to_owned()),
3802 );
3803
3804 t.unblock("q1");
3805 assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
3806 t.unblock("q2");
3807 assert_eq!(t.status, TaskStatus::Held);
3808 }
3809
3810 #[test]
3811 fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
3812 let mut t = task("blocked mid-run");
3816 t.start("run-1".to_owned());
3817 assert_eq!(t.status, TaskStatus::Running);
3818
3819 t.block(vec!["q1".to_owned()], None);
3820 t.unblock("q1");
3821 assert_eq!(t.status, TaskStatus::Queued);
3822 }
3823
3824 #[test]
3825 fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
3826 let mut t = task("legacy record, held before it was blocked");
3832 t.hold_source = Some(HoldSource::Machine);
3833 t.hold_reason = Some("legacy hold reason".to_owned());
3834 t.status = TaskStatus::Blocked;
3835 t.blocked_by = vec!["q1".to_owned()];
3836 t.blocked_from = None;
3837
3838 t.unblock("q1");
3839 assert_eq!(t.status, TaskStatus::Held);
3840 }
3841
3842 #[test]
3843 fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
3844 let mut t = task("legacy record, ordinary dependency block");
3845 t.status = TaskStatus::Blocked;
3846 t.blocked_by = vec!["dep".to_owned()];
3847 t.blocked_from = None;
3848
3849 t.unblock("dep");
3850 assert_eq!(t.status, TaskStatus::Queued);
3851 }
3852
3853 #[test]
3854 fn answering_a_question_is_recorded_and_survives_a_release() {
3855 let mut t = task("asked something");
3856 t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
3857 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
3858 t.unblock("q1");
3859 assert_eq!(t.status, TaskStatus::Queued);
3860 assert_eq!(t.answers.len(), 1);
3861 assert_eq!(t.answers[0].answer, "SQLite");
3862
3863 t.release();
3867 assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
3868 }
3869
3870 #[test]
3871 fn resumption_review_branch_requires_unchanged_authored_criteria() {
3872 for source in ["cli", "web"] {
3873 let mut t = task("review old candidate");
3874 t.request_review("candidate-A".into());
3875 t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
3876 let mut unchanged = t.clone();
3877 unchanged.release_operator(source, None);
3878 assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
3879 t.edit(
3880 "review approved head".into(),
3881 "Reuse measured PR head B; fix Clippy; review".into(),
3882 )
3883 .unwrap();
3884 t.release_operator(source, None);
3885 assert!(t.review_branch.is_none());
3886 assert!(t.review_instruction.is_none());
3887 let mut legacy = unchanged;
3889 legacy.review_instruction = None;
3890 legacy.hold_for_handover(None, "still checked out".into());
3891 legacy
3892 .edit("new plan".into(), "Reuse measured PR head B".into())
3893 .unwrap();
3894 legacy.release_operator(source, None);
3895 assert!(legacy.review_branch.is_none());
3896 }
3897 }
3898
3899 #[test]
3900 fn a_refused_handover_keeps_the_review_branch_across_release() {
3901 let mut t = task("refused takeover");
3902 t.start("run-1".to_owned());
3903 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
3904 assert_eq!(t.status, TaskStatus::Held);
3905 assert_eq!(t.attempts, 1);
3906 t.release();
3907 assert_eq!(t.status, TaskStatus::Queued);
3908 assert_eq!(t.attempts, 0);
3909 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
3910
3911 t.hold_for_handover(None, "again".to_owned());
3913 t.requeue();
3914 assert!(t.review_branch.is_none());
3915
3916 let mut m = task("manual");
3918 m.review_branch = Some("magi/x/A".to_owned());
3919 m.hold_manual(None);
3920 m.release();
3921 assert!(m.review_branch.is_none());
3922 }
3923
3924 #[test]
3925 fn requesting_review_requeues_the_task_and_remembers_the_branch() {
3926 let mut t = task("blocked run with a surviving branch");
3927 t.start("run-1".to_owned());
3928 t.fail("blocked with major findings", 5);
3929 assert_eq!(t.status, TaskStatus::Failed);
3930
3931 t.request_review("magi/eba2/A".to_owned());
3932 assert_eq!(t.status, TaskStatus::Queued);
3933 assert_eq!(t.attempts, 0);
3934 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
3935
3936 t.release();
3938 assert!(t.review_branch.is_none());
3939 }
3940
3941 #[test]
3942 fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
3943 let mut t = task("retry");
3944 t.start("run-1".to_owned());
3945 t.requeue();
3946 assert!(t.fresh_start);
3947
3948 t.release();
3949 assert!(!t.fresh_start);
3950 }
3951
3952 #[test]
3953 fn urgent_only_changes_queued_or_held_and_preserves_settings() {
3954 for status in [
3955 TaskStatus::Queued,
3956 TaskStatus::Held,
3957 TaskStatus::Running,
3958 TaskStatus::Done,
3959 TaskStatus::Failed,
3960 TaskStatus::Blocked,
3961 TaskStatus::Parked,
3962 ] {
3963 let mut t = task("urgent settings");
3964 t.status = status;
3965 t.attempts = 2;
3966 t.hold_reason = Some("waiting".into());
3967 t.priority = 7;
3968 t.interrupt = true;
3969 let before = serde_json::to_value(&t).unwrap();
3970 for urgent in [true, true, false, false] {
3971 let result = t.set_urgent(urgent);
3972 let mut expected = before.clone();
3973 if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
3974 result.unwrap();
3975 expected["urgent"] = urgent.into();
3976 } else {
3977 let error = result.unwrap_err().to_string();
3978 assert!(error.contains(status.as_str()), "{error}");
3979 assert!(error.contains("queued or held"), "{error}");
3980 }
3981 assert_eq!(serde_json::to_value(&t).unwrap(), expected);
3982 }
3983 }
3984 }
3985
3986 #[test]
3987 fn priority_can_be_changed_while_queued_but_not_while_running() {
3988 let mut t = task("reprioritise me");
3989 t.set_priority(5).unwrap();
3990 assert_eq!(t.priority, 5);
3991
3992 t.start("run-1".to_owned());
3993 let err = t.set_priority(9).unwrap_err().to_string();
3994 assert!(err.contains("running"), "{err}");
3995 assert_eq!(t.priority, 5, "the rejected write must not partially apply");
3996 }
3997
3998 #[test]
3999 fn interrupt_can_be_marked_while_queued_but_not_while_running() {
4000 let mut t = task("interrupt me");
4001 assert!(!t.interrupt, "off unless asked, same as any other task");
4002
4003 t.set_interrupt(true).unwrap();
4004 assert!(t.interrupt);
4005
4006 t.start("run-1".to_owned());
4007 assert!(
4008 !t.interrupt,
4009 "the mark is one-shot: dispatching the task fulfils it, \
4010 whatever the run that follows ends up doing"
4011 );
4012 let err = t.set_interrupt(true).unwrap_err().to_string();
4013 assert!(err.contains("running"), "{err}");
4014 t.set_interrupt(false).unwrap();
4017 assert!(!t.interrupt);
4018 }
4019
4020 #[test]
4024 fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
4025 let mut t = task("interrupt me");
4026 t.set_interrupt(true).unwrap();
4027 t.start("run-1".to_owned());
4028 t.fail("mock failure", 5);
4029 assert_eq!(t.status, TaskStatus::Failed);
4030 assert!(
4031 !t.interrupt,
4032 "one attempt already spent the mark; a retry is an ordinary \
4033 requeue, not a fresh interrupt request"
4034 );
4035 }
4036
4037 #[test]
4038 fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
4039 let (_dir, q) = queue();
4040 let mut a = task("first filed");
4041 let mut b = task("second filed");
4042 a.id = "20260101-000001-aaaa".to_owned();
4043 b.id = "20260101-000002-bbbb".to_owned();
4044 q.put(&mut a).unwrap();
4045 q.put(&mut b).unwrap();
4046
4047 assert_eq!(
4048 q.next_runnable().unwrap().id,
4049 a.id,
4050 "with equal priority the older task goes first, so a burst of \
4051 new work cannot starve it"
4052 );
4053 assert_eq!(
4054 q.list()[0].id,
4055 b.id,
4056 "but the list an operator reads is newest first, the same as \
4057 before priority existed - a's turn to run does not make it the \
4058 newest task"
4059 );
4060
4061 let mut a = q.get(&a.id).unwrap();
4062 a.set_priority(10).unwrap();
4063 q.put(&mut a).unwrap();
4064
4065 assert_eq!(
4066 q.next_runnable().unwrap().id,
4067 a.id,
4068 "a raised priority must be reflected the moment it is saved"
4069 );
4070 assert_eq!(
4074 q.list()[0].id,
4075 a.id,
4076 "the raised task must sort first in the list an operator reads, \
4077 not only in next_runnable's own ordering"
4078 );
4079 }
4080
4081 #[test]
4082 fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
4083 let mut t = Task::new(
4084 "old title".to_owned(),
4085 "old instruction".to_owned(),
4086 PathBuf::from("/repo"),
4087 Source::Agent {
4088 run: "20260101-000000-beef".to_owned(),
4089 node: "implement".to_owned(),
4090 },
4091 );
4092 let id = t.id.clone();
4093 let created_at = t.created_at;
4094 t.runs.push("20260101-000000-beef".to_owned());
4095
4096 t.edit("new title".to_owned(), "new instruction".to_owned())
4097 .unwrap();
4098
4099 assert_eq!(t.title, "new title");
4100 assert_eq!(t.instruction, "new instruction");
4101 assert_eq!(t.id, id, "editing must not mint a new id");
4102 assert_eq!(t.created_at, created_at);
4103 assert_eq!(
4104 t.source,
4105 Source::Agent {
4106 run: "20260101-000000-beef".to_owned(),
4107 node: "implement".to_owned(),
4108 },
4109 "editing must not turn agent attribution into human"
4110 );
4111 assert_eq!(t.runs, ["20260101-000000-beef"]);
4112 }
4113
4114 #[test]
4115 fn editing_is_refused_once_a_task_is_running_or_finished() {
4116 let mut running = task("in flight");
4117 running.start("run-1".to_owned());
4118 let err = running
4119 .edit("x".to_owned(), "y".to_owned())
4120 .unwrap_err()
4121 .to_string();
4122 assert!(err.contains("running"), "{err}");
4123
4124 let mut done = task("finished");
4125 done.succeed();
4126 let err = done
4127 .edit("x".to_owned(), "y".to_owned())
4128 .unwrap_err()
4129 .to_string();
4130 assert!(err.contains("done"), "{err}");
4131
4132 let mut queued = task("waiting");
4134 queued.edit("x".to_owned(), "y".to_owned()).unwrap();
4135 let mut held = task("parked");
4136 held.hold_machine(None);
4137 held.edit("x".to_owned(), "y".to_owned()).unwrap();
4138 }
4139
4140 #[test]
4141 fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
4142 let (_dir, q) = queue();
4143 let path = q.path_of("20260101-000000-aaaa");
4144 std::fs::create_dir_all(q.root()).unwrap();
4145 std::fs::write(
4146 &path,
4147 serde_json::json!({
4148 "schema": SCHEMA,
4149 "id": "20260101-000000-aaaa",
4150 "title": "from before hold reasons existed",
4151 "instruction": "from before hold reasons existed",
4152 "repo": ".",
4153 "source": { "kind": "human" },
4154 "status": "held",
4155 "created_at": Timestamp::now().to_string(),
4156 "updated_at": Timestamp::now().to_string(),
4157 })
4158 .to_string(),
4159 )
4160 .unwrap();
4161
4162 let task = q.get("20260101-000000-aaaa").expect("must still read");
4163 assert!(task.hold_reason.is_none());
4164 assert!(task.operator_held());
4165 }
4166
4167 #[test]
4168 fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
4169 let (_dir, q) = queue();
4170 let path = q.path_of("20260101-000000-bbbb");
4171 std::fs::create_dir_all(q.root()).unwrap();
4172 std::fs::write(
4173 &path,
4174 serde_json::json!({
4175 "schema": 2,
4176 "id": "20260101-000000-bbbb",
4177 "title": "old manual recovery",
4178 "instruction": "old manual recovery",
4179 "repo": ".",
4180 "source": { "kind": "human" },
4181 "status": "held",
4182 "hold_reason": "active manual recovery run20260912-224242-daf5",
4183 "created_at": Timestamp::now().to_string(),
4184 "updated_at": Timestamp::now().to_string(),
4185 })
4186 .to_string(),
4187 )
4188 .unwrap();
4189
4190 let task = q.get("20260101-000000-bbbb").expect("must still read");
4191 assert_eq!(task.hold_source, None);
4192 assert!(task.operator_held());
4193 }
4194
4195 #[test]
4196 fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
4197 let (_dir, q) = queue();
4198 let path = q.path_of("20260101-000000-aaaa");
4199 std::fs::create_dir_all(q.root()).unwrap();
4200 std::fs::write(
4201 &path,
4202 serde_json::json!({
4203 "schema": SCHEMA,
4204 "id": "20260101-000000-aaaa",
4205 "title": "from before diagnostics existed",
4206 "instruction": "from before diagnostics existed",
4207 "repo": ".",
4208 "source": { "kind": "human" },
4209 "status": "held",
4210 "created_at": Timestamp::now().to_string(),
4211 "updated_at": Timestamp::now().to_string(),
4212 })
4213 .to_string(),
4214 )
4215 .unwrap();
4216
4217 let task = q.get("20260101-000000-aaaa").expect("must still read");
4218 assert!(task.diagnostic.is_none());
4219 }
4220
4221 #[test]
4222 fn a_schema_1_task_with_no_blocking_fields_still_reads() {
4223 let (_dir, q) = queue();
4227 let path = q.path_of("20260101-000000-aaaa");
4228 std::fs::create_dir_all(q.root()).unwrap();
4229 std::fs::write(
4230 &path,
4231 serde_json::json!({
4232 "schema": 1,
4233 "id": "20260101-000000-aaaa",
4234 "title": "from before blocking existed",
4235 "instruction": "from before blocking existed",
4236 "repo": ".",
4237 "source": { "kind": "human" },
4238 "status": "queued",
4239 "created_at": Timestamp::now().to_string(),
4240 "updated_at": Timestamp::now().to_string(),
4241 })
4242 .to_string(),
4243 )
4244 .unwrap();
4245
4246 let task = q.get("20260101-000000-aaaa").expect("must still read");
4247 assert!(task.blocked_by.is_empty());
4248 assert!(task.block_reason.is_none());
4249 assert!(task.answers.is_empty());
4250 assert!(task.review_branch.is_none());
4251 }
4252
4253 #[test]
4254 fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
4255 let mut held = task("diagnosed");
4260 held.start("run-1".to_owned());
4261 held.fail("gate red", 1);
4262 held.diagnostic = Some("cargo test failed: ...".to_owned());
4263 assert_eq!(held.status, TaskStatus::Held);
4264
4265 held.release();
4266 assert!(held.diagnostic.is_none());
4267
4268 held.diagnostic = Some("cargo test failed: ...".to_owned());
4269 held.succeed();
4270 assert!(held.diagnostic.is_none());
4271 }
4272
4273 #[test]
4274 fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
4275 let mut t = task("retried");
4276 t.start("run-1".to_owned());
4277 t.diagnostic = Some("stale evidence from a previous hold".to_owned());
4278 t.fail("unrelated config error", 5);
4279 assert_eq!(t.status, TaskStatus::Failed);
4280 assert!(
4281 t.diagnostic.is_none(),
4282 "fail() must not let an old diagnostic outlive the run that produced it"
4283 );
4284 }
4285
4286 #[test]
4287 fn a_claim_is_exclusive_and_releases_on_drop() {
4288 let (_dir, q) = queue();
4289 let mut t = task("contended");
4290 q.put(&mut t).unwrap();
4291
4292 let held = q.claim(&t.id).unwrap();
4293 assert!(
4294 q.claim(&t.id).is_err(),
4295 "two daemons must not drive one task into two runs"
4296 );
4297 drop(held);
4298 assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
4299 }
4300
4301 #[test]
4302 fn a_round_trip_survives_disk() {
4303 let (_dir, q) = queue();
4304 let mut t = Task::new(
4305 "titled".to_owned(),
4306 "body".to_owned(),
4307 PathBuf::from("/repo"),
4308 Source::Agent {
4309 run: "20260101-000000-beef".to_owned(),
4310 node: "implement".to_owned(),
4311 },
4312 );
4313 t.priority = 3;
4314 q.put(&mut t).unwrap();
4315
4316 let back = q.get(&t.id).unwrap();
4317 assert_eq!(back.id, t.id);
4318 assert_eq!(back.priority, 3);
4319 assert_eq!(back.source.label(), "implement@beef");
4320 assert_eq!(q.get(t.short()).unwrap().id, t.id);
4322 }
4323
4324 #[test]
4325 fn an_unreadable_task_does_not_take_the_queue_down() {
4326 let (_dir, q) = queue();
4327 let mut t = task("fine");
4328 q.put(&mut t).unwrap();
4329 std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
4330
4331 let listed = q.list();
4332 assert_eq!(listed.len(), 1, "the readable task still lists");
4333 assert_eq!(listed[0].id, t.id);
4334 }
4335
4336 #[test]
4337 fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
4338 let (_dir, q) = queue();
4339 let path = q.path_of("20260101-000000-aaaa");
4340 std::fs::create_dir_all(q.root()).unwrap();
4341 std::fs::write(
4342 &path,
4343 serde_json::json!({
4344 "schema": SCHEMA,
4345 "id": "20260101-000000-aaaa",
4346 "title": "from before solo existed",
4347 "instruction": "from before solo existed",
4348 "repo": ".",
4349 "source": { "kind": "human" },
4350 "status": "queued",
4351 "created_at": Timestamp::now().to_string(),
4352 "updated_at": Timestamp::now().to_string(),
4353 })
4354 .to_string(),
4355 )
4356 .unwrap();
4357
4358 let task = q.get("20260101-000000-aaaa").expect("must still read");
4359 assert!(!task.solo, "a queue file with no `solo` field means false");
4360 }
4361
4362 #[test]
4363 fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
4364 let (_dir, q) = queue();
4365 let path = q.path_of("20260101-000000-bbbb");
4366 std::fs::create_dir_all(q.root()).unwrap();
4367 std::fs::write(
4368 &path,
4369 serde_json::json!({
4370 "schema": SCHEMA,
4371 "id": "20260101-000000-bbbb",
4372 "title": "from before urgent existed",
4373 "instruction": "from before urgent existed",
4374 "repo": ".",
4375 "source": { "kind": "human" },
4376 "status": "queued",
4377 "created_at": Timestamp::now().to_string(),
4378 "updated_at": Timestamp::now().to_string(),
4379 })
4380 .to_string(),
4381 )
4382 .unwrap();
4383
4384 let task = q.get("20260101-000000-bbbb").expect("must still read");
4385 assert!(
4386 !task.urgent,
4387 "a queue file with no `urgent` field means false, same as `solo`"
4388 );
4389 }
4390
4391 #[test]
4392 fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
4393 let (_dir, q) = queue();
4394 let mut t = task("from the future");
4395 q.put(&mut t).unwrap();
4396 let path = q.path_of(&t.id);
4397 let body = std::fs::read_to_string(&path)
4398 .unwrap()
4399 .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
4400 std::fs::write(&path, body).unwrap();
4401
4402 let err = q.get(&t.id).unwrap_err().to_string();
4403 assert!(err.contains("schema 99"), "{err}");
4404 }
4405
4406 #[test]
4407 fn revision_moves_when_the_queue_changes() {
4408 let (_dir, q) = queue();
4409 assert_eq!(q.revision(), 0, "an empty queue has no revision");
4410 let mut t = task("first");
4411 q.put(&mut t).unwrap();
4412 assert!(q.revision() > 0, "a written task moves the revision");
4413 }
4414
4415 #[test]
4416 fn revision_moves_when_deleting_an_older_task() {
4417 let (dir, q) = queue();
4418 let questions = Questions::at(dir.path().join("questions"));
4419 let mut t1 = task("older");
4420 q.put(&mut t1).unwrap();
4421 std::thread::sleep(std::time::Duration::from_millis(10));
4423 let mut t2 = task("newer");
4424 q.put(&mut t2).unwrap();
4425
4426 let rev_before = q.revision();
4427 q.remove(&t1.id, false, &questions).unwrap();
4428 let rev_after = q.revision();
4429
4430 assert_ne!(
4431 rev_before, rev_after,
4432 "deleting an older task must change the revision so other clients see the deletion"
4433 );
4434 }
4435
4436 #[test]
4437 fn removing_a_task_takes_it_out_of_the_listing() {
4438 let (dir, q) = queue();
4439 let questions = Questions::at(dir.path().join("questions"));
4440 let mut t = task("delete me");
4441 q.put(&mut t).unwrap();
4442 let removed = q.remove(t.short(), false, &questions).unwrap();
4443 assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
4444 assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
4445 assert!(q.list().is_empty());
4446 assert!(
4447 q.remove(&t.id, false, &questions).is_err(),
4448 "removing twice is an error"
4449 );
4450 }
4451
4452 #[test]
4453 fn removing_a_task_takes_its_stale_lock_with_it() {
4454 let (dir, q) = queue();
4455 let questions = Questions::at(dir.path().join("questions"));
4456 let mut t = task("interrupted");
4457 q.put(&mut t).unwrap();
4458
4459 let claim = q.claim(&t.id).unwrap();
4462 std::mem::forget(claim);
4463 assert!(
4464 q.claim(&t.id).is_err(),
4465 "the orphaned lock is what makes the task look claimed"
4466 );
4467
4468 let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
4470 assert!(err.contains("live daemon"), "{err}");
4471 assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
4472
4473 q.remove(&t.id, false, &questions).unwrap();
4475 assert!(q.list().is_empty());
4476 let mut again = task("interrupted");
4477 again.id = t.id.clone();
4478 q.put(&mut again).unwrap();
4479 assert!(
4480 q.claim(&t.id).is_ok(),
4481 "a task that comes back must be claimable, which a left-behind lock would prevent"
4482 );
4483 }
4484
4485 fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
4486 crate::notices::Notices::at(dir.join("notifications")).list()
4487 }
4488
4489 #[test]
4490 fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
4491 let (dir, q) = queue();
4492 let questions = Questions::at(dir.path().join("questions"));
4493 let mut dep = task("dependency");
4494 q.put(&mut dep).unwrap();
4495 let mut blocked = task("waiting");
4496 blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
4497 q.put(&mut blocked).unwrap();
4498
4499 let removed = q.remove(&dep.id, false, &questions).unwrap();
4500 assert_eq!(removed.released, [blocked.id.clone()]);
4501 assert!(removed.still_blocked.is_empty());
4502
4503 let after = q.get(&blocked.id).unwrap();
4504 assert_eq!(after.status, TaskStatus::Queued);
4505 assert!(after.blocked_by.is_empty());
4506 assert!(after.block_reason.is_none());
4507 let notes = notices_of(dir.path());
4508 assert_eq!(notes.len(), 1, "{notes:?}");
4509 assert_eq!(notes[0].severity, crate::notices::Severity::Info);
4510 assert_eq!(
4511 notes[0].link,
4512 Some(crate::notices::Link::Task {
4513 id: blocked.id.clone(),
4514 })
4515 );
4516 }
4517
4518 #[test]
4519 fn removing_one_of_two_dependencies_keeps_the_other() {
4520 let (dir, q) = queue();
4521 let questions = Questions::at(dir.path().join("questions"));
4522 let mut dep = task("dependency");
4523 q.put(&mut dep).unwrap();
4524 let mut other = task("other");
4525 q.put(&mut other).unwrap();
4526 let mut blocked = task("waiting");
4527 blocked.block(
4528 vec![dep.id.clone(), other.id.clone()],
4529 Some("waits on both".to_owned()),
4530 );
4531 q.put(&mut blocked).unwrap();
4532
4533 let removed = q.remove(&dep.id, false, &questions).unwrap();
4534 assert!(removed.released.is_empty());
4535 assert_eq!(removed.still_blocked, [blocked.id.clone()]);
4536
4537 let after = q.get(&blocked.id).unwrap();
4538 assert_eq!(after.status, TaskStatus::Blocked);
4539 assert_eq!(after.blocked_by, [other.id.clone()]);
4540 assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
4543 }
4544
4545 #[test]
4546 fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
4547 let (dir, q) = queue();
4548 let questions = Questions::at(dir.path().join("questions"));
4549 let mut dep = task("dependency");
4550 q.put(&mut dep).unwrap();
4551 let mut blocked = task("waiting");
4552 blocked.block(vec![dep.id.clone()], None);
4553 q.put(&mut blocked).unwrap();
4554
4555 let claim = q.claim(&blocked.id).unwrap();
4557 let removed = q.remove(&dep.id, false, &questions).unwrap();
4558 assert!(removed.released.is_empty());
4559 drop(claim);
4560
4561 let mut task = q.get(&blocked.id).unwrap();
4562 assert_eq!(task.status, TaskStatus::Blocked);
4563 assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
4564 assert_eq!(
4565 q.apply_deleted_blockers(&mut task, &questions),
4566 [dep.id.clone()]
4567 );
4568 assert_eq!(task.status, TaskStatus::Queued);
4569 }
4570
4571 #[test]
4572 fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
4573 let (dir, q) = queue();
4574 let questions = Questions::at(dir.path().join("questions"));
4575 let mut dep = task("dependency");
4576 dep.id = "20261013-193611-ca5b".into();
4577 q.put(&mut dep).unwrap();
4578 let mut waiting = task("waiting");
4579 waiting.block(vec!["ca5b".into()], None);
4580 q.put(&mut waiting).unwrap();
4581 q.remove(&dep.id, false, &questions).unwrap();
4582 assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
4583 assert_eq!(
4584 dependency(&q, &questions, "ca5b"),
4585 Dependency::Deleted(dep.id)
4586 );
4587 }
4588
4589 #[test]
4590 fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
4591 let (dir, queue) = queue();
4592 let questions = Questions::at(dir.path().join("questions"));
4593 let mut dep = task("dependency");
4594 dep.id = "20261013-193611-ca5b".into();
4595 queue.put(&mut dep).unwrap();
4596 std::fs::create_dir_all(questions.root()).unwrap();
4597 let question_id = "20261014-193611-ca5b";
4598 std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
4599 assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
4600 assert_eq!(
4601 dependency(&queue, &questions, "ca5b"),
4602 Dependency::Ambiguous
4603 );
4604 assert_eq!(
4605 dependency(&queue, &questions, &dep.id),
4606 Dependency::Task(dep.id)
4607 );
4608 assert_eq!(
4609 dependency(&queue, &questions, question_id),
4610 Dependency::Question(question_id.into())
4611 );
4612 assert_eq!(
4613 dependency(&queue, &questions, "missing"),
4614 Dependency::Missing
4615 );
4616 }
4617
4618 #[test]
4619 fn dependency_resolution_directory_errors_are_unknown_not_missing() {
4620 let dir = tempfile::tempdir().unwrap();
4621 let q = Queue::at(dir.path().join("not-a-directory"));
4622 std::fs::write(q.root(), "file").unwrap();
4623 let questions = Questions::at(dir.path().join("questions"));
4624 assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
4625 assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
4626 }
4627
4628 #[test]
4629 fn a_dependency_without_a_tombstone_is_still_missing() {
4630 let (dir, q) = queue();
4631 let questions = Questions::at(dir.path().join("questions"));
4632 let mut dep = task("dependency");
4633 q.put(&mut dep).unwrap();
4634 std::fs::remove_file(q.path_of(&dep.id)).unwrap();
4635 assert_eq!(
4636 missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
4637 [dep.id.clone()]
4638 );
4639 assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
4640 }
4641
4642 #[test]
4643 fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
4644 let mut t = task("held");
4645 t.hold_machine(Some("because".to_owned()));
4646 t.blocked_by = vec!["gone".to_owned()];
4647 assert!(!t.dependency_deleted("gone"));
4648 assert_eq!(t.status, TaskStatus::Held);
4649 }
4650
4651 #[test]
4652 fn a_failed_record_removal_takes_the_tombstone_back() {
4653 let (_dir, q) = queue();
4654 let mut t = task("stays");
4655 q.put(&mut t).unwrap();
4656 q.write_tombstone(&t.id).unwrap();
4657 let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
4658 assert!(err.is_err());
4659 assert!(!q.was_deleted(&t.id));
4662 }
4663
4664 fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
4665 let p = dir.join(name);
4666 std::fs::write(&p, body).unwrap();
4667 p
4668 }
4669
4670 #[test]
4671 fn an_attachment_copy_survives_deleting_its_source() {
4672 let (dir, q) = queue();
4673 let src = source_file(dir.path(), "shot.png", "pixels");
4674 let mut t = task("with a picture");
4675 let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
4676 q.put(&mut t).unwrap();
4677 std::fs::remove_file(&src).unwrap();
4678 assert_eq!(names, ["shot.png"]);
4679 let loaded = q.get(&t.id).unwrap();
4680 let paths = q.attachment_paths(&loaded);
4681 assert_eq!(paths.len(), 1);
4682 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
4683 }
4684
4685 #[test]
4686 fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
4687 let (dir, q) = queue();
4688 let mut t = task("bad names");
4689 for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
4690 let src = source_file(dir.path(), name, "x");
4691 assert!(
4692 q.attach(&mut t, &[src]).is_err(),
4693 "`{name}` must be refused"
4694 );
4695 }
4696 assert!(!crate::ask::valid_asset_name("C:foo.png"));
4700 #[cfg(not(windows))]
4701 {
4702 let src = source_file(dir.path(), "C:foo.png", "x");
4703 assert!(q.attach(&mut t, &[src]).is_err());
4704 }
4705 let long = format!("{}.png", "a".repeat(70));
4706 let src = source_file(dir.path(), &long, "x");
4707 assert!(q.attach(&mut t, &[src]).is_err());
4708 assert!(t.attachments.is_empty());
4709 assert!(!q.attachments_dir(&t.id).exists());
4710 }
4711
4712 #[test]
4713 fn a_taken_attachment_name_is_numbered_not_overwritten() {
4714 let (dir, q) = queue();
4715 let a = source_file(dir.path(), "shot.png", "one");
4716 let sub = dir.path().join("other");
4717 std::fs::create_dir_all(&sub).unwrap();
4718 let b = source_file(&sub, "shot.png", "two");
4719 let mut t = task("collision");
4720 q.attach(&mut t, &[a]).unwrap();
4721 q.attach(&mut t, &[b]).unwrap();
4722 assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
4723 let paths = q.attachment_paths(&t);
4724 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
4725 assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
4726 }
4727
4728 #[test]
4729 fn a_renumbered_name_stays_inside_the_length_bound() {
4730 let (dir, q) = queue();
4731 let name = format!("{}.png", "a".repeat(60));
4732 assert_eq!(name.len(), 64);
4733 let a = source_file(dir.path(), &name, "one");
4734 let sub = dir.path().join("other");
4735 std::fs::create_dir_all(&sub).unwrap();
4736 let b = source_file(&sub, &name, "two");
4737 let mut t = task("long");
4738 q.attach(&mut t, &[a, b]).unwrap();
4739 assert_eq!(t.attachments.len(), 2);
4740 assert!(
4741 t.attachments
4742 .iter()
4743 .all(|n| crate::ask::valid_asset_name(n))
4744 );
4745 assert!(t.attachments[1].ends_with("-2.png"));
4746 }
4747
4748 #[test]
4749 fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
4750 let (dir, q) = queue();
4751 let good = source_file(dir.path(), "good.png", "ok");
4752 let mut t = task("partial");
4753 q.attach(&mut t, &[good]).unwrap();
4754 let more = source_file(dir.path(), "more.png", "ok");
4755 let missing = dir.path().join("missing.png");
4756 assert!(q.attach(&mut t, &[more, missing]).is_err());
4757 assert_eq!(t.attachments, ["good.png"]);
4758 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4759 .unwrap()
4760 .flatten()
4761 .collect();
4762 assert_eq!(on_disk.len(), 1);
4763 }
4764
4765 fn block_put(q: &Queue, t: &Task) -> PathBuf {
4767 let tmp = q.path_of(&t.id).with_extension("json.tmp");
4768 std::fs::create_dir_all(&tmp).unwrap();
4769 tmp
4770 }
4771
4772 #[test]
4773 fn a_failed_put_leaves_no_new_attachment_directory() {
4774 let (dir, q) = queue();
4775 let mut t = task("fresh");
4776 let tmp = block_put(&q, &t);
4777 let src = source_file(dir.path(), "shot.png", "x");
4778 assert!(q.attach_and_put(&mut t, &[src]).is_err());
4779 assert!(t.attachments.is_empty());
4780 assert!(!q.attachments_dir(&t.id).exists());
4781 assert!(!q.path_of(&t.id).exists());
4782 std::fs::remove_dir(tmp).unwrap();
4783 }
4784
4785 #[test]
4786 fn a_failed_put_removes_only_the_copy_it_just_made() {
4787 let (dir, q) = queue();
4788 let mut t = task("edited");
4789 let first = source_file(dir.path(), "first.png", "1");
4790 q.attach_and_put(&mut t, &[first]).unwrap();
4791 block_put(&q, &t);
4792 let second = source_file(dir.path(), "second.png", "2");
4793 assert!(q.attach_and_put(&mut t, &[second]).is_err());
4794 assert_eq!(t.attachments, ["first.png"]);
4795 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4796 .unwrap()
4797 .flatten()
4798 .map(|e| e.file_name().to_string_lossy().into_owned())
4799 .collect();
4800 assert_eq!(on_disk, ["first.png"]);
4801 assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
4802 }
4803
4804 #[test]
4805 fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
4806 let (dir, q) = queue();
4807 let questions = Questions::at(dir.path().join("questions"));
4808 let gone = task("gone");
4809 let mut other = task("other");
4810 let mut live = task("live");
4811 q.put(&mut other).unwrap();
4812 q.put(&mut live).unwrap();
4813 let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
4816 std::fs::create_dir_all(&orphan).unwrap();
4817 std::fs::write(orphan.join("shot.png"), "x").unwrap();
4818 let busy = q.root.join(format!("{}.attachments.removing", live.id));
4821 std::fs::create_dir_all(&busy).unwrap();
4822
4823 q.remove(&other.id, false, &questions).unwrap();
4824 assert!(!orphan.exists(), "an orphan is swept");
4825 assert!(busy.exists(), "a removal in progress is left alone");
4826 }
4827
4828 #[test]
4829 fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
4830 let (dir, q) = queue();
4831 let questions = Questions::at(dir.path().join("questions"));
4832 let mut t = task("stuck");
4833 let src = source_file(dir.path(), "shot.png", "x");
4834 q.attach_and_put(&mut t, &[src]).unwrap();
4835 let aside = q.root.join(format!("{}.attachments.removing", t.id));
4838 std::fs::create_dir_all(&aside).unwrap();
4839 std::fs::write(aside.join("old.png"), "o").unwrap();
4840 assert!(q.remove(&t.id, false, &questions).is_err());
4841 assert!(q.path_of(&t.id).exists());
4842 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
4843 }
4844
4845 #[test]
4846 fn a_failed_record_removal_puts_the_attachments_back() {
4847 let (dir, q) = queue();
4848 let mut t = task("rollback");
4849 let src = source_file(dir.path(), "shot.png", "x");
4850 q.attach_and_put(&mut t, &[src]).unwrap();
4851 let err = q
4852 .remove_record_with_attachments(&t.id, |_| {
4853 Err(std::io::Error::other("injected failure"))
4854 })
4855 .unwrap_err();
4856 assert!(format!("{err:#}").contains("injected failure"));
4857 assert!(q.path_of(&t.id).exists());
4858 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
4859 assert!(
4860 !q.root
4861 .join(format!("{}.attachments.removing", t.id))
4862 .exists()
4863 );
4864 }
4865
4866 #[test]
4867 fn editing_a_task_keeps_its_attachments() {
4868 let (dir, q) = queue();
4869 let src = source_file(dir.path(), "shot.png", "x");
4870 let mut t = task("editable");
4871 q.attach(&mut t, &[src]).unwrap();
4872 t.edit("new".to_owned(), "new text".to_owned()).unwrap();
4873 q.put(&mut t).unwrap();
4874 assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
4875 }
4876
4877 #[test]
4878 fn removing_a_task_deletes_its_attachments() {
4879 let (dir, q) = queue();
4880 let questions = Questions::at(dir.path().join("questions"));
4881 let src = source_file(dir.path(), "shot.png", "x");
4882 let mut t = task("doomed");
4883 q.attach(&mut t, &[src]).unwrap();
4884 q.put(&mut t).unwrap();
4885 assert!(q.attachments_dir(&t.id).is_dir());
4886 q.remove(&t.id, false, &questions).unwrap();
4887 assert!(!q.attachments_dir(&t.id).exists());
4888 assert!(q.list().is_empty());
4889 }
4890
4891 #[test]
4892 fn a_task_written_before_attachments_still_reads() {
4893 let (_dir, q) = queue();
4894 let mut t = task("old");
4895 q.put(&mut t).unwrap();
4896 let path = q.path_of(&t.id);
4897 let mut v: serde_json::Value =
4898 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
4899 v.as_object_mut().unwrap().remove("attachments");
4900 std::fs::write(&path, v.to_string()).unwrap();
4901 assert!(q.get(&t.id).unwrap().attachments.is_empty());
4902 }
4903
4904 #[test]
4905 fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
4906 let q = Queue::at(PathBuf::from("relative-queue"));
4907 let mut t = task("rel");
4908 t.attachments.push("shot.png".to_owned());
4909 let paths = q.attachment_paths(&t);
4910 assert!(paths[0].is_absolute(), "{}", paths[0].display());
4911 assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
4912 }
4913
4914 #[test]
4915 fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
4916 let dir = tempfile::tempdir().unwrap();
4917 let q = Queue::at(dir.path().join("queue"));
4918 let mut t = task("resume");
4919 t.start("A".into());
4920 t.record_pass("A", 0, 1, None);
4921 t.start("B".into());
4922 q.put(&mut t).unwrap();
4923 assert_eq!(t.runs, vec!["A", "B"]);
4924 t.start("A".into());
4925 t.record_pass("A", 1, 3, None);
4926 q.put(&mut t).unwrap();
4927 let saved = q.get(&t.id).unwrap();
4928 assert_eq!(saved.runs, vec!["B", "A"]);
4929 assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
4930 assert_eq!(saved.attempts, 3);
4931 assert_eq!(saved.passes.len(), 2);
4932 assert_eq!(saved.passes[0].run, "A");
4933 assert_eq!(saved.passes[1].run, "A");
4934 }
4935
4936 #[test]
4937 fn repeated_start_records_one_run_and_separate_passes() {
4938 let mut t = task("resume");
4939 t.start("same-run".into());
4940 t.record_pass("same-run", 0, 1, None);
4941 t.start("same-run".into());
4942 t.record_pass("same-run", 1, 2, None);
4943 assert_eq!(t.runs, vec!["same-run"]);
4944 assert_eq!(t.attempts, 2);
4945 assert_eq!(t.passes.len(), 2);
4946 }
4947
4948 #[test]
4949 fn link_run_adds_a_run_once_and_touches_nothing_else() {
4950 let dir = tempfile::tempdir().unwrap();
4951 let queue = Queue::at(dir.path().join("queue"));
4952 let mut t = Task::new(
4953 "t".to_owned(),
4954 "do it".to_owned(),
4955 PathBuf::from("."),
4956 Source::Human,
4957 );
4958 queue.put(&mut t).unwrap();
4959 let before = queue.get(&t.id).unwrap();
4960
4961 let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
4962 assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
4963 assert_eq!(linked.status, before.status);
4964 assert_eq!(linked.attempts, before.attempts);
4965 assert_eq!(linked.interrupt, before.interrupt);
4966
4967 let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
4968 assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
4969 assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
4970 assert!(queue.link_run("no-such-task", "r").is_err());
4971 }
4972
4973 #[test]
4974 fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
4975 let dir = tempfile::tempdir().unwrap();
4976 let queue = Queue::at(dir.path().join("queue"));
4977 let mut t = Task::new(
4978 "t".to_owned(),
4979 "do it".to_owned(),
4980 PathBuf::from("."),
4981 Source::Human,
4982 );
4983 queue.put(&mut t).unwrap();
4984 let mut snapshot = queue.get(&t.id).unwrap();
4986 queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
4987
4988 snapshot.start("20260930-000000-aaaa".to_owned());
4989 queue.put(&mut snapshot).unwrap();
4990
4991 let stored = queue.get(&t.id).unwrap();
4992 assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
4993 assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
4994 assert_eq!(stored.attempts, 1);
4995 }
4996
4997 #[test]
4998 fn concurrent_links_and_daemon_saves_lose_nothing() {
4999 let dir = tempfile::tempdir().unwrap();
5000 let queue = Queue::at(dir.path().join("queue"));
5001 let mut t = Task::new(
5002 "t".to_owned(),
5003 "do it".to_owned(),
5004 PathBuf::from("."),
5005 Source::Human,
5006 );
5007 queue.put(&mut t).unwrap();
5008 let id = t.id.clone();
5009
5010 let linkers: Vec<_> = (0..4)
5011 .map(|n| {
5012 let (queue, id) = (queue.clone(), id.clone());
5013 std::thread::spawn(move || {
5014 for k in 0..10 {
5015 queue
5016 .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
5017 .unwrap();
5018 }
5019 })
5020 })
5021 .collect();
5022 let mut mine = queue.get(&id).unwrap();
5025 for k in 0..10 {
5026 mine.start(format!("20260930-000009-d{k:03}"));
5027 queue.put(&mut mine).unwrap();
5028 }
5029 for l in linkers {
5030 l.join().unwrap();
5031 }
5032
5033 let stored = queue.get(&id).unwrap();
5034 assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
5035 assert_eq!(
5036 stored.attempts, 10,
5037 "linking never rewinds the daemon's work"
5038 );
5039 }
5040
5041 fn age_lock(path: &Path) {
5042 let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
5043 f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
5044 .unwrap();
5045 }
5046
5047 #[test]
5048 fn concurrent_stale_takeover_yields_one_holder() {
5049 use std::sync::atomic::{AtomicUsize, Ordering};
5050 use std::sync::{Arc, Barrier};
5051 for _ in 0..5 {
5052 let dir = tempfile::tempdir().unwrap();
5053 let q = Queue::at(dir.path().to_path_buf());
5054 let lock = dir.path().join("t.write-lock");
5055 std::fs::write(&lock, "dead-0000").unwrap();
5056 age_lock(&lock);
5057 let n = 6;
5058 let barrier = Arc::new(Barrier::new(n));
5059 let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
5060 let handles: Vec<_> = (0..n)
5061 .map(|_| {
5062 let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
5063 std::thread::spawn(move || {
5064 b.wait();
5065 let g = q.lock_task("t").unwrap();
5066 let held = now.fetch_add(1, Ordering::SeqCst) + 1;
5067 max.fetch_max(held, Ordering::SeqCst);
5068 std::thread::sleep(std::time::Duration::from_millis(20));
5069 now.fetch_sub(1, Ordering::SeqCst);
5070 drop(g);
5071 })
5072 })
5073 .collect();
5074 for h in handles {
5075 h.join().unwrap();
5076 }
5077 assert_eq!(max.load(Ordering::SeqCst), 1);
5078 assert!(!lock.exists());
5079 }
5080 }
5081
5082 #[test]
5083 fn dropping_a_stolen_lock_leaves_the_new_holders_lock() {
5084 let dir = tempfile::tempdir().unwrap();
5085 let q = Queue::at(dir.path().to_path_buf());
5086 let lock = dir.path().join("t.write-lock");
5087 let a = q.lock_task("t").unwrap();
5088 age_lock(&lock);
5089 let b = q.lock_task("t").unwrap();
5090 assert_ne!(a.token, b.token);
5091 drop(a);
5092 assert_eq!(std::fs::read_to_string(&lock).unwrap(), b.token);
5093 drop(b);
5094 assert!(!lock.exists());
5095 }
5096
5097 #[test]
5098 fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
5099 let dir = tempfile::tempdir().unwrap();
5100 let lock = dir.path().join("t.write-lock");
5101 std::fs::write(&lock, "new-token").unwrap();
5102 assert!(!break_stale(&lock, "old-token"));
5103 assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
5104 assert!(!break_stale(&lock, "new-token"));
5106 assert!(lock.exists());
5107 age_lock(&lock);
5108 assert!(break_stale(&lock, "new-token"));
5109 assert!(!lock.exists());
5110 }
5111
5112 #[test]
5113 fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
5114 let dir = tempfile::tempdir().unwrap();
5115 let lock = dir.path().join("t.write-lock");
5116 std::fs::write(&lock, "dead-1").unwrap();
5117 age_lock(&lock);
5118 let marker = break_marker(&lock, "dead-1");
5119 std::fs::write(&marker, "crashed-remover").unwrap();
5120 assert!(!break_stale(&lock, "dead-1"));
5122 assert!(lock.exists());
5123 age_lock(&marker);
5125 assert!(break_stale(&lock, "dead-1"));
5126 assert!(!lock.exists());
5127 assert!(!marker.exists());
5128 }
5129
5130 fn chat_task() -> Task {
5131 Task::new(
5132 "t".to_owned(),
5133 "do t".to_owned(),
5134 PathBuf::from("."),
5135 Source::Agent {
5136 run: "20260902-000000-beef".to_owned(),
5137 node: CHAT_NODE.to_owned(),
5138 },
5139 )
5140 }
5141
5142 #[test]
5143 fn a_blocked_hold_keeps_its_anchor_but_a_new_run_starts_a_new_hold() {
5144 let mut t = chat_task();
5145 t.hold_machine(Some("waiting".into()));
5146 let old: Timestamp = "2020-01-01T00:00:00Z".parse().unwrap();
5147 t.held_at = Some(old);
5148 t.block(vec!["decision".into()], None);
5149 t.hold_machine(Some("same waiting decision".into()));
5150 assert_eq!(t.held_at, Some(old));
5151 t.start("real-new-run".into());
5152 t.fail("new failure", 1);
5153 assert_ne!(
5154 t.held_at,
5155 Some(old),
5156 "old answers cannot approve a new hold"
5157 );
5158 }
5159
5160 #[test]
5161 fn legacy_delivery_survives_upgrade_and_a_no_progress_run_link() {
5162 let (_dir, queue) = queue();
5163 let mut t = chat_task();
5164 t.start("old-run".into());
5165 t.hold_machine(Some("operator is considering recovery".into()));
5166 t.chat_notice = Some(ChatNotice {
5167 key: format!("held#{}", t.held_at.unwrap()),
5168 at: Timestamp::now(),
5169 outcome: ChatNoticeOutcome::Posted,
5170 });
5171 let mut raw = serde_json::to_value(&t).unwrap();
5172 raw.as_object_mut().unwrap().remove("chat_notices");
5173 raw.as_object_mut().unwrap().remove("hold_notice_cause");
5174 std::fs::create_dir_all(queue.path_of(&t.id).parent().unwrap()).unwrap();
5175 std::fs::write(queue.path_of(&t.id), serde_json::to_string(&raw).unwrap()).unwrap();
5176 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5177 queue
5178 .modify(&t.id, |task| {
5179 task.requeue();
5180 task.link_run("unprepared-run");
5181 task.startup_notice_cause("startup-source");
5182 task.fail("same unresolved preflight", 0);
5183 true
5184 })
5185 .unwrap();
5186 assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5187 queue
5188 .modify(&t.id, |task| {
5189 task.succeed();
5190 true
5191 })
5192 .unwrap();
5193 assert_eq!(
5194 queue.get(&t.id).unwrap().chat_notice_due().as_deref(),
5195 Some("done")
5196 );
5197 }
5198
5199 #[test]
5200 fn a_dependency_wait_does_not_change_when_released_and_blocked_again() {
5201 let mut t = chat_task();
5202 t.block(vec!["a".into(), "b".into()], Some("waiting".into()));
5203 let key = t.chat_notice_key().unwrap();
5204 t.chat_notices.push(key);
5205 t.release();
5206 t.block(
5207 vec!["b".into(), "a".into(), "a".into()],
5208 Some("still waiting on prerequisites".into()),
5209 );
5210 assert_eq!(t.chat_notice_due(), None);
5211 t.block(
5212 vec!["new-question".into()],
5213 Some("different decision".into()),
5214 );
5215 assert!(t.chat_notice_due().is_some());
5216 }
5217
5218 #[test]
5219 fn running_startup_failures_preserve_stage_and_report_distinct_blockers() {
5220 let mut t = chat_task();
5221 t.start("existing-run".into());
5222 t.fail("implementation failed", 1);
5223 let failed = t.chat_notice_key().unwrap();
5224 t.chat_notices.push(failed.clone());
5225 t.release_operator("test", None);
5226 t.start("existing-run".into());
5227 t.fail_startup("startup-source", "same unresolved recovery source", 1);
5228 assert_eq!(
5229 t.chat_notice_due(),
5230 None,
5231 "no-progress recovery stays quiet"
5232 );
5233 t.release_operator("test", None);
5234 t.start("existing-run".into());
5235 t.fail_startup("startup-attachments", "attachment missing", 1);
5236 let attachments = t.chat_notice_due().expect("new attachment blocker");
5237 assert_ne!(attachments, failed);
5238 assert_eq!(
5239 t.hold_notice_cause.as_ref().unwrap().stage,
5240 "startup-attachments"
5241 );
5242 t.chat_notices.push(attachments);
5243 t.release_operator("test", None);
5244 t.start("existing-run".into());
5245 t.fail_startup(
5246 "startup-attachments",
5247 "same missing attachment, reworded",
5248 1,
5249 );
5250 assert_eq!(
5251 t.chat_notice_due(),
5252 None,
5253 "same stage after attempt counter reset"
5254 );
5255 t.release_operator("test", None);
5256 t.start("existing-run".into());
5257 t.fail_startup("startup-config", "different preflight stage", 1);
5258 assert!(t.chat_notice_due().is_some());
5259 assert_eq!(
5260 t.hold_notice_cause.as_ref().unwrap().stage,
5261 "startup-config"
5262 );
5263 }
5264
5265 #[test]
5266 fn structured_causes_preserve_no_progress_retries_and_report_new_passes() {
5267 let mut t = chat_task();
5268 t.hold_machine(Some("startup source missing".into()));
5269 t.startup_notice_cause("startup-source");
5270 let first = t.chat_notice_key().unwrap();
5271 t.chat_notices.push(first.clone());
5272 t.release_operator("test", None);
5273 t.startup_notice_cause("startup-source");
5274 t.hold_machine(Some("same source cannot be read".into()));
5275 assert_eq!(t.chat_notice_due(), None);
5276 t.startup_notice_cause("startup-config");
5277 assert!(t.chat_notice_due().is_some(), "different startup stage");
5278 t.notice_cause("failed", Some("existing-run"), Some(0));
5279 let run_failure = t.chat_notice_key().unwrap();
5280 t.chat_notices.push(run_failure.clone());
5281 t.release_operator("test", None);
5282 t.startup_notice_cause("startup-source");
5283 t.hold_machine(Some("no new work happened".into()));
5284 assert_eq!(t.chat_notice_due(), None);
5285 t.notice_cause("failed", Some("existing-run"), Some(1));
5286 assert!(t.chat_notice_due().is_some(), "a real resumed pass failed");
5287 assert_ne!(t.chat_notice_key().unwrap(), run_failure);
5288 }
5289
5290 #[test]
5291 fn legacy_hold_baseline_does_not_replay_or_hide_new_failures_and_completion() {
5292 let mut t = chat_task();
5293 t.hold_machine(Some("still waiting".into()));
5294 t.chat_notice = Some(ChatNotice {
5295 key: "held#2026-10-14T00:00:00Z".into(),
5296 at: Timestamp::now(),
5297 outcome: ChatNoticeOutcome::Posted,
5298 });
5299 assert_eq!(t.chat_notice_due(), None, "old hold baseline");
5300 t.import_chat_notice();
5301 t.notice_cause("failed", Some("new-run"), Some(0));
5302 assert!(t.chat_notice_due().is_some(), "new producer evidence");
5303 t.succeed();
5304 assert_eq!(t.chat_notice_due().as_deref(), Some("done"));
5305 }
5306
5307 #[test]
5308 fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
5309 let mut t = chat_task();
5310 assert_eq!(t.chat_notice_due(), None, "queued");
5311 t.start("r1".to_owned());
5312 assert_eq!(t.chat_notice_due(), None, "running");
5313 t.fail("boom", 3);
5314 assert_eq!(t.status, TaskStatus::Failed);
5315 assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
5316 t.start("r2".to_owned());
5317 t.park("upgrade");
5318 assert_eq!(t.chat_notice_due(), None, "parked");
5319 t.fail("boom", 1);
5320 assert_eq!(t.status, TaskStatus::Held);
5321 let key = t
5322 .chat_notice_due()
5323 .expect("held on the last attempt is final");
5324 t.chat_notice = Some(ChatNotice {
5325 key,
5326 at: Timestamp::now(),
5327 outcome: ChatNoticeOutcome::Posted,
5328 });
5329 assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
5330 t.release();
5331 assert_eq!(t.chat_notice_due(), None, "requeued after a release");
5332 t.start("r3".to_owned());
5333 t.succeed();
5334 assert!(t.chat_notice_due().is_some(), "done is a new final state");
5335 }
5336
5337 #[test]
5338 fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
5339 for key in ["done", "done#0", "done#1", "done#12"] {
5340 let mut t = chat_task();
5341 t.start("r1".to_owned());
5342 t.succeed();
5343 t.chat_notice = Some(ChatNotice {
5344 key: key.to_owned(),
5345 at: Timestamp::now(),
5346 outcome: ChatNoticeOutcome::Posted,
5347 });
5348 t.requeue();
5349 t.start("r2".to_owned());
5350 t.succeed();
5351 assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
5352 assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
5353 }
5354 }
5355
5356 #[test]
5357 fn a_release_without_progress_keeps_the_unresolved_notice_cause() {
5358 let mut t = chat_task();
5359 t.hold_manual(Some("wait".to_owned()));
5360 let first = t.chat_notice_due().expect("held");
5361 t.chat_notice = Some(ChatNotice {
5362 key: first.clone(),
5363 at: Timestamp::now(),
5364 outcome: ChatNoticeOutcome::Posted,
5365 });
5366 t.hold_manual(Some("still waiting".to_owned()));
5367 assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
5368 t.release();
5369 std::thread::sleep(std::time::Duration::from_millis(5));
5370 t.hold_manual(None);
5371 assert_eq!(t.chat_notice_due(), None, "no progress, same cause");
5372 }
5373
5374 #[test]
5375 fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
5376 let mut human = task("h");
5377 human.succeed();
5378 assert_eq!(human.chat_notice_due(), None);
5379 let mut inherited = task("i");
5381 inherited.origin_chat = Some("20260902-000000-beef".to_owned());
5382 inherited.succeed();
5383 assert_eq!(inherited.chat_notice_due(), None);
5384 let mut old = chat_task();
5385 old.schema = 12;
5386 old.status = TaskStatus::Done;
5387 assert_eq!(old.chat_notice_due(), None, "history is not replayed");
5388 let mut live = chat_task();
5389 live.schema = 12;
5390 live.start("r1".to_owned());
5391 live.succeed();
5392 assert!(live.chat_notice_due().is_some(), "started by this build");
5393 let mut held = chat_task();
5394 held.schema = 12;
5395 held.hold_machine(Some("config".to_owned()));
5396 assert!(held.chat_notice_due().is_some(), "held before any start");
5397 }
5398
5399 #[test]
5400 fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
5401 let tmp = tempfile::tempdir().expect("tempdir");
5402 let q = Queue::at(tmp.path().join("queue"));
5403 let mut t = chat_task();
5404 t.succeed();
5405 q.put(&mut t).expect("put");
5406 let mut stale = q.get(&t.id).expect("get");
5407 let key = t.chat_notice_due().expect("due");
5408 q.modify(&t.id, |x| {
5409 x.chat_notice = Some(ChatNotice {
5410 key: key.clone(),
5411 at: Timestamp::now(),
5412 outcome: ChatNoticeOutcome::Posted,
5413 });
5414 true
5415 })
5416 .expect("modify");
5417 q.put(&mut stale).expect("stale put");
5418 assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
5419 }
5420}