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)]
393 pub answers: Vec<AnsweredQuestion>,
394 #[serde(default)]
400 pub triage_applied: Vec<String>,
401 #[serde(default)]
407 pub actions_applied: Vec<String>,
408 #[serde(default)]
413 pub resume_override: Option<OperatorResume>,
414 #[serde(default)]
417 pub resumption: Option<Resumption>,
418 #[serde(default)]
420 pub resumption_history: Vec<Resumption>,
421 #[serde(default)]
423 pub operator_decided_at: Option<Timestamp>,
424 #[serde(default)]
436 pub review_branch: Option<String>,
437 #[serde(default)]
439 pub review_instruction: Option<String>,
440 #[serde(default)]
442 pub pr_recovery: Option<crate::pr_recovery::Selection>,
443 #[serde(default)]
446 pub fresh_start: bool,
447 #[serde(default)]
458 pub interrupt: bool,
459 #[serde(default)]
482 pub urgent: bool,
483 #[serde(default)]
489 pub attachments: Vec<String>,
490 #[serde(default)]
494 pub followup: Option<FollowUp>,
495 #[serde(default)]
501 pub origin_chat: Option<String>,
502 #[serde(default)]
509 pub overrides: Option<RunOverrides>,
510 #[serde(default)]
517 pub review_of: Option<String>,
518 #[serde(default)]
523 pub held_at: Option<Timestamp>,
524 #[serde(default)]
528 pub park_reason: Option<String>,
529 #[serde(default)]
532 pub chat_notice: Option<ChatNotice>,
533 #[serde(default)]
538 pub passes: Vec<TaskPass>,
539 pub created_at: Timestamp,
541 pub updated_at: Timestamp,
543}
544
545pub(crate) fn is_done_notice_key(key: &str) -> bool {
547 key == "done"
548 || key
549 .strip_prefix("done#")
550 .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
551}
552
553#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
555#[serde(rename_all = "snake_case")]
556pub enum PassCost {
557 Spent,
559 Refunded,
561}
562
563#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
565pub struct TaskPass {
566 pub run: String,
568 pub pass_index: usize,
570 pub cost: PassCost,
572 pub attempts_after: usize,
574 pub settled_at: Timestamp,
576 #[serde(default)]
578 pub parked: Option<crate::run::ParkCause>,
579}
580
581#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
583#[serde(rename_all = "lowercase")]
584pub enum ChatNoticeOutcome {
585 Posted,
587 Skipped,
589}
590
591#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
593pub struct ChatNotice {
594 pub key: String,
596 pub at: Timestamp,
598 pub outcome: ChatNoticeOutcome,
600}
601
602#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
604pub struct RunOverrides {
605 #[serde(default)]
608 pub inherit_daemon_options: bool,
609 #[serde(default)]
611 pub merge: Option<String>,
612 #[serde(default)]
614 pub candidates: Option<usize>,
615 #[serde(default)]
617 pub judges: Option<usize>,
618 #[serde(default)]
620 pub reviewers: Option<usize>,
621 #[serde(default)]
623 pub review_rounds: Option<usize>,
624 #[serde(default)]
626 pub seed: Option<u64>,
627 #[serde(default)]
629 pub config: Option<PathBuf>,
630}
631
632impl RunOverrides {
633 pub fn merge_over(&mut self, other: &RunOverrides) {
635 macro_rules! take {
636 ($($f:ident),*) => {$(
637 if other.$f.is_some() {
638 self.$f.clone_from(&other.$f);
639 }
640 )*};
641 }
642 take!(
643 merge,
644 candidates,
645 judges,
646 reviewers,
647 review_rounds,
648 seed,
649 config
650 );
651 }
652
653 pub fn apply(&self, config: &mut crate::config::Config) {
655 if let Some(n) = self.candidates {
656 config.graph.implementers = n;
657 }
658 if let Some(n) = self.judges {
659 config.graph.judges = n;
660 }
661 if let Some(n) = self.reviewers {
662 config.graph.reviewers = n;
663 }
664 if let Some(n) = self.review_rounds {
665 config.graph.review_rounds = n;
666 }
667 if let Some(m) = &self.merge
668 && let Ok(mode) = crate::daemon::merge_mode(m)
669 {
670 config.merge.mode = mode;
671 }
672 if let Some(s) = self.seed {
673 config.blind.seed = Some(s);
674 }
675 }
676}
677
678#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
680pub struct FollowUp {
681 pub run: String,
683 #[serde(default)]
685 pub origin_task: Option<String>,
686 pub pr: String,
688 pub findings: Vec<String>,
690 pub generation: u32,
693}
694
695#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
697pub struct Resumption {
698 pub generation: u64,
700 pub source: String,
702 pub question_id: Option<String>,
704 pub at: Timestamp,
706 pub instruction: String,
708 pub answers_through: usize,
710 pub runs_through: usize,
713 pub cleared_reason: Option<String>,
715 #[serde(default)]
717 pub blocking_evidence: Vec<String>,
718}
719
720#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
723pub struct OperatorResume {
724 pub question_id: String,
726 pub at: Timestamp,
728 #[serde(default)]
731 pub conductor_rehold: Option<String>,
732 #[serde(default)]
735 pub forced: bool,
736 #[serde(default)]
743 pub pinned_run: Option<String>,
744}
745
746#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
749pub struct AnsweredQuestion {
750 pub question: String,
752 pub answer: String,
754}
755
756impl Task {
757 pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
759 let now = Timestamp::now();
760 let origin_chat = match &source {
761 Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
762 _ => None,
763 };
764 Self {
765 schema: SCHEMA,
766 id: new_id(),
767 title,
768 instruction,
769 repo,
770 source,
771 priority: 0,
772 solo: false,
773 status: TaskStatus::Queued,
774 attempts: 0,
775 attempt_limit: None,
776 runs: Vec::new(),
777 last_error: None,
778 hold_reason: None,
779 hold_source: None,
780 diagnostic: None,
781 blocked_by: Vec::new(),
782 block_reason: None,
783 blocked_from: None,
784 answers: Vec::new(),
785 triage_applied: Vec::new(),
786 actions_applied: Vec::new(),
787 resume_override: None,
788 resumption: None,
789 resumption_history: Vec::new(),
790 operator_decided_at: None,
791 review_branch: None,
792 review_instruction: None,
793 pr_recovery: None,
794 fresh_start: false,
795 interrupt: false,
796 urgent: false,
797 attachments: Vec::new(),
798 followup: None,
799 origin_chat,
800 overrides: None,
801 review_of: None,
802 held_at: None,
803 park_reason: None,
804 chat_notice: None,
805 passes: Vec::new(),
806 created_at: now,
807 updated_at: now,
808 }
809 }
810
811 pub fn chat_talk(&self) -> Option<&str> {
814 if let Some(id) = &self.origin_chat {
815 return Some(id);
816 }
817 match &self.source {
818 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
819 _ => None,
820 }
821 }
822
823 pub fn filed_by_chat(&self) -> Option<&str> {
827 match &self.source {
828 Source::Agent { run, node } if node == CHAT_NODE => Some(run),
829 _ => None,
830 }
831 }
832
833 pub fn chat_notice_key(&self) -> Option<String> {
839 match self.status {
840 TaskStatus::Done => Some("done".to_owned()),
841 TaskStatus::Held => Some(format!(
842 "held#{}",
843 self.held_at.map(|t| t.to_string()).unwrap_or_default()
844 )),
845 TaskStatus::Blocked => Some(format!(
846 "blocked#{}#{}",
847 self.blocked_by.join(","),
848 self.block_reason.as_deref().unwrap_or_default()
849 )),
850 _ => None,
851 }
852 }
853
854 pub fn chat_notice_due(&self) -> Option<String> {
858 self.filed_by_chat()?;
859 if self.schema < 13 {
860 return None;
861 }
862 let key = self.chat_notice_key()?;
863 match &self.chat_notice {
864 Some(n) if n.key == key || (key == "done" && is_done_notice_key(&n.key)) => None,
865 _ => Some(key),
866 }
867 }
868
869 pub fn short(&self) -> &str {
871 short(&self.id)
872 }
873
874 pub fn mark_triage_applied(&mut self, question_id: &str) {
876 if !self.triage_applied(question_id) {
877 self.triage_applied.push(question_id.to_owned());
878 }
879 }
880
881 pub fn action_applied(&self, question_id: &str) -> bool {
884 self.actions_applied.iter().any(|id| id == question_id)
885 }
886
887 pub fn mark_action_applied(&mut self, question_id: &str) {
889 if !self.action_applied(question_id) {
890 self.actions_applied.push(question_id.to_owned());
891 }
892 }
893
894 pub fn triage_applied(&self, question_id: &str) -> bool {
896 self.triage_applied.iter().any(|id| id == question_id)
897 }
898
899 pub fn start(&mut self, run: String) {
912 self.stamp_schema();
913 self.status = TaskStatus::Running;
914 self.attempts += 1;
915 self.runs.retain(|id| id != &run);
916 self.runs.push(run);
917 self.last_error = None;
918 self.park_reason = None;
919 self.fresh_start = false;
920 self.interrupt = false;
921 self.resume_override = None;
923 }
924
925 pub fn link_run(&mut self, run: &str) -> bool {
930 if self.runs.iter().any(|r| r == run) {
931 return false;
932 }
933 self.runs.push(run.to_owned());
934 true
935 }
936
937 pub fn succeed(&mut self) {
947 self.stamp_schema();
948 self.status = TaskStatus::Done;
949 self.resume_override = None;
950 self.last_error = None;
951 self.park_reason = None;
952 self.hold_reason = None;
953 self.hold_source = None;
954 self.diagnostic = None;
955 self.blocked_by.clear();
956 self.block_reason = None;
957 self.blocked_from = None;
958 }
959
960 pub fn already_landed(&mut self, note: impl Into<String>) {
968 self.succeed();
969 self.attempts = self.attempts.saturating_sub(1);
970 self.last_error = Some(note.into());
971 }
972
973 pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
1003 if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
1004 return &[];
1005 }
1006 &self.runs[..self.runs.len() - 1]
1007 }
1008
1009 pub fn successor_of(&self, run: &str) -> Option<&String> {
1017 let pos = self.runs.iter().rposition(|r| r == run)?;
1018 self.runs.get(pos + 1)
1019 }
1020
1021 pub fn earlier_attempts(&self) -> &[String] {
1028 &self.runs
1029 }
1030
1031 fn stamp_schema(&mut self) {
1038 self.schema = self.schema.max(SCHEMA);
1039 }
1040
1041 fn note_held(&mut self) {
1042 self.stamp_schema();
1043 if self.status != TaskStatus::Held || self.held_at.is_none() {
1044 self.held_at = Some(Timestamp::now());
1045 }
1046 }
1047
1048 pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
1064 let why = why.into();
1065 self.diagnostic = None;
1066 self.park_reason = None;
1067 self.status = if self.attempts >= self.effective_attempt_limit(max_attempts) {
1068 self.note_held();
1069 self.hold_source = Some(HoldSource::Machine);
1070 self.hold_reason = Some(why.clone());
1071 TaskStatus::Held
1072 } else {
1073 TaskStatus::Failed
1074 };
1075 self.last_error = Some(why);
1076 }
1077
1078 pub fn stall(&mut self, why: impl Into<String>) {
1088 self.last_error = Some(why.into());
1089 self.diagnostic = None;
1090 self.attempts = self.attempts.saturating_sub(1);
1091 self.status = TaskStatus::Failed;
1092 }
1093
1094 pub fn park(&mut self, why: impl Into<String>) {
1100 self.diagnostic = None;
1101 self.attempts = self.attempts.saturating_sub(1);
1102 self.status = TaskStatus::Parked;
1103 self.park_reason = Some(why.into());
1104 self.schema = self.schema.max(SCHEMA);
1107 }
1108
1109 pub fn record_pass(
1112 &mut self,
1113 run: &str,
1114 pass_index: usize,
1115 attempts_before: usize,
1116 parked: Option<crate::run::ParkCause>,
1117 ) {
1118 if self
1119 .passes
1120 .iter()
1121 .any(|p| p.run == run && p.pass_index == pass_index)
1122 {
1123 return;
1124 }
1125 let cost = if self.attempts < attempts_before {
1126 PassCost::Refunded
1127 } else {
1128 PassCost::Spent
1129 };
1130 self.passes.push(TaskPass {
1131 run: run.to_owned(),
1132 pass_index,
1133 cost,
1134 attempts_after: self.attempts,
1135 settled_at: Timestamp::now(),
1136 parked,
1137 });
1138 self.schema = self.schema.max(SCHEMA);
1139 }
1140
1141 pub fn operator_held(&self) -> bool {
1147 self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
1148 }
1149
1150 pub fn hold_manual(&mut self, reason: Option<String>) {
1160 self.operator_decided_at = Some(Timestamp::now());
1162 self.hold_unrequested_retry(reason);
1163 }
1164
1165 pub(crate) fn hold_unrequested_retry(&mut self, reason: Option<String>) {
1168 self.park_reason = None;
1169 self.note_held();
1170 self.status = TaskStatus::Held;
1171 if reason.is_some() {
1172 self.hold_reason = reason;
1173 }
1174 self.hold_source = Some(HoldSource::Manual);
1175 self.blocked_by.clear();
1176 self.block_reason = None;
1177 self.blocked_from = None;
1178 }
1179
1180 pub fn hold_machine(&mut self, reason: Option<String>) {
1185 self.park_reason = None;
1186 self.note_held();
1187 self.status = TaskStatus::Held;
1188 if reason.is_some() {
1189 self.hold_reason = reason;
1190 }
1191 self.hold_source = Some(HoldSource::Machine);
1192 self.blocked_by.clear();
1193 self.block_reason = None;
1194 self.blocked_from = None;
1195 }
1196
1197 pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
1207 self.stamp_schema();
1208 if self.status != TaskStatus::Blocked {
1209 self.blocked_from = Some(self.status);
1210 }
1211 self.status = TaskStatus::Blocked;
1212 self.blocked_by = blocked_by;
1213 self.block_reason = reason;
1214 }
1215
1216 pub fn unblock(&mut self, resolved_id: &str) {
1239 if self.status != TaskStatus::Blocked {
1240 return;
1241 }
1242 self.blocked_by.retain(|id| id != resolved_id);
1243 self.restore_if_unblocked();
1244 }
1245
1246 pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
1257 if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
1258 return false;
1259 }
1260 self.blocked_by.retain(|id| id != deleted_id);
1261 self.restore_if_unblocked();
1262 true
1263 }
1264
1265 fn restore_if_unblocked(&mut self) {
1269 if self.blocked_by.is_empty() {
1270 self.status = match self.blocked_from {
1271 Some(TaskStatus::Running) => TaskStatus::Queued,
1272 Some(other) => other,
1273 None if self.hold_reason.is_some() || self.hold_source.is_some() => {
1274 TaskStatus::Held
1275 }
1276 None => TaskStatus::Queued,
1277 };
1278 self.block_reason = None;
1279 self.blocked_from = None;
1280 }
1281 }
1282
1283 pub fn record_answer(&mut self, question: String, answer: String) {
1288 self.answers.push(AnsweredQuestion { question, answer });
1289 }
1290
1291 pub fn request_review(&mut self, branch: String) {
1295 self.release();
1296 self.review_branch = Some(branch);
1297 self.review_instruction = Some(self.instruction.clone());
1298 }
1299
1300 pub fn requeue(&mut self) {
1303 self.release();
1304 self.review_branch = None;
1306 self.review_instruction = None;
1307 self.fresh_start = true;
1308 }
1309
1310 pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
1319 if branch.is_some() {
1320 self.review_branch = branch;
1321 self.review_instruction = Some(self.instruction.clone());
1322 }
1323 self.hold_machine(Some(reason));
1324 }
1325
1326 pub fn set_urgent(&mut self, urgent: bool) -> Result<()> {
1328 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1329 bail!(
1330 "task {} is {}; urgent can only change while queued or held; \
1331 wait for the active attempt to finish and requeue or hold the task first",
1332 self.short(),
1333 self.status.as_str()
1334 );
1335 }
1336 self.urgent = urgent;
1337 Ok(())
1338 }
1339
1340 pub fn set_priority(&mut self, priority: i32) -> Result<()> {
1349 if self.status == TaskStatus::Running {
1350 bail!(
1351 "task {} is running; its priority cannot be changed until \
1352 this attempt finishes",
1353 self.short()
1354 );
1355 }
1356 self.priority = priority;
1357 Ok(())
1358 }
1359
1360 pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
1375 if interrupt && !self.status.runnable() {
1376 bail!(
1377 "task {} is {}; only a queued or failed task can be marked \
1378 to interrupt",
1379 self.short(),
1380 self.status.as_str()
1381 );
1382 }
1383 self.interrupt = interrupt;
1384 Ok(())
1385 }
1386
1387 pub fn effective_attempt_limit(&self, daemon_limit: usize) -> usize {
1389 self.attempt_limit.map_or(daemon_limit, NonZeroUsize::get)
1390 }
1391
1392 pub fn change_attempt_limit(&mut self, limit: Option<NonZeroUsize>) -> Result<()> {
1395 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1396 bail!("only queued or held tasks can change the attempt limit");
1397 }
1398 self.stamp_schema();
1399 self.attempt_limit = limit;
1400 Ok(())
1401 }
1402
1403 pub fn implementers(&self) -> Option<usize> {
1405 if self.solo {
1406 Some(1)
1407 } else {
1408 self.overrides.as_ref().and_then(|o| o.candidates)
1409 }
1410 }
1411
1412 pub fn change_implementers(&mut self, count: u8) -> Result<()> {
1415 self.check_implementers_editable()?;
1416 let explicit = self.overrides.as_ref().and_then(|o| o.config.as_deref());
1417 let (mut config, _) = crate::config::Config::discover(&self.repo, explicit)?;
1418 if let Some(o) = &self.overrides {
1419 o.apply(&mut config);
1420 }
1421 self.set_implementers(count, &config)
1422 }
1423
1424 fn check_implementers_editable(&self) -> Result<()> {
1425 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1426 bail!(
1427 "task {} is {}; only queued or held tasks can change implementers",
1428 self.short(),
1429 self.status.as_str()
1430 );
1431 }
1432 Ok(())
1433 }
1434
1435 pub fn set_implementers(&mut self, count: u8, config: &crate::config::Config) -> Result<()> {
1437 self.check_implementers_editable()?;
1438 crate::talk::check_implementers(count, config).map_err(anyhow::Error::msg)?;
1439 self.stamp_schema();
1440 self.solo = count == 1;
1441 let mut overrides = self.overrides.clone().unwrap_or_else(|| RunOverrides {
1442 inherit_daemon_options: true,
1443 ..Default::default()
1444 });
1445 overrides.candidates = (count > 1).then_some(usize::from(count));
1446 let mut settings = overrides.clone();
1447 settings.inherit_daemon_options = false;
1448 self.overrides = if overrides.inherit_daemon_options && settings == RunOverrides::default()
1450 {
1451 None
1452 } else {
1453 Some(overrides)
1454 };
1455 Ok(())
1456 }
1457
1458 pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
1470 if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1471 bail!(
1472 "task {} is {}; only a queued or held task's instruction can \
1473 be edited",
1474 self.short(),
1475 self.status.as_str()
1476 );
1477 }
1478 if self.review_branch.is_some() && self.review_instruction.is_none() {
1481 self.review_instruction = Some(self.instruction.clone());
1482 }
1483 if self.instruction != instruction {
1484 self.resume_override = None;
1487 }
1488 self.title = title;
1489 self.instruction = instruction;
1490 Ok(())
1491 }
1492
1493 pub fn handed_off(&mut self, why: impl Into<String>) {
1513 let why = why.into();
1514 self.diagnostic = None;
1515 self.note_held();
1516 self.status = TaskStatus::Held;
1517 self.hold_source = Some(HoldSource::Machine);
1518 self.hold_reason = Some(why.clone());
1519 self.last_error = Some(why);
1520 }
1521
1522 pub fn release_operator(&mut self, source: &str, question_id: Option<&str>) {
1524 if self
1525 .resumption
1526 .as_ref()
1527 .is_some_and(|r| r.instruction != self.instruction)
1528 {
1529 self.resume_override = None;
1532 }
1533 let approval = Resumption {
1534 generation: self.resumption.as_ref().map_or(1, |r| r.generation + 1),
1535 source: source.to_owned(),
1536 question_id: question_id.map(str::to_owned),
1537 at: Timestamp::now(),
1538 instruction: self.instruction.clone(),
1539 answers_through: self.answers.len(),
1540 runs_through: self.runs.len(),
1541 blocking_evidence: Vec::new(),
1542 cleared_reason: self
1543 .hold_reason
1544 .clone()
1545 .or_else(|| self.block_reason.clone()),
1546 };
1547 self.release();
1548 self.operator_decided_at = Some(approval.at);
1549 if let Some(previous) = self.resumption.replace(approval) {
1550 self.resumption_history.push(previous);
1551 }
1552 }
1553
1554 pub fn release(&mut self) {
1558 let refused_handover = self.status == TaskStatus::Held
1559 && self.hold_source == Some(HoldSource::Machine)
1560 && self.review_branch.is_some()
1561 && self
1562 .review_instruction
1563 .as_ref()
1564 .is_none_or(|i| *i == self.instruction);
1565 self.status = TaskStatus::Queued;
1566 self.held_at = None;
1567 self.attempts = 0;
1568 self.last_error = None;
1569 self.park_reason = None;
1570 self.hold_reason = None;
1573 self.hold_source = None;
1574 self.diagnostic = None;
1575 self.blocked_by.clear();
1580 self.block_reason = None;
1581 self.blocked_from = None;
1582 if !refused_handover {
1586 self.review_branch = None;
1587 self.review_instruction = None;
1588 }
1589 self.fresh_start = false;
1590 }
1591}
1592
1593fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
1597 use std::io::ErrorKind;
1598 let (stem, ext) = match name.rfind('.') {
1599 Some(i) if i > 0 => (&name[..i], &name[i..]),
1600 _ => (name, ""),
1601 };
1602 for n in 1u32.. {
1603 let candidate = if n == 1 {
1604 name.to_owned()
1605 } else {
1606 let suffix = format!("-{n}");
1607 let room = 64usize.saturating_sub(suffix.len() + ext.len());
1608 let stem: String = stem.chars().take(room).collect();
1609 format!("{stem}{suffix}{ext}")
1610 };
1611 if !crate::ask::valid_asset_name(&candidate) {
1612 bail!("no valid attachment name is left for `{name}`");
1613 }
1614 let path = dir.join(&candidate);
1615 match std::fs::OpenOptions::new()
1616 .write(true)
1617 .create_new(true)
1618 .open(&path)
1619 {
1620 Ok(mut out) => {
1621 let copied = std::fs::File::open(src)
1622 .and_then(|mut input| std::io::copy(&mut input, &mut out));
1623 if let Err(e) = copied {
1624 drop(out);
1625 let _ = std::fs::remove_file(&path);
1626 return Err(e).with_context(|| format!("copy {}", src.display()));
1627 }
1628 return Ok((candidate, path));
1629 }
1630 Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
1631 Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
1632 }
1633 }
1634 unreachable!("the counter never runs out")
1635}
1636
1637const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
1640
1641struct TaskLock {
1644 path: PathBuf,
1645 token: String,
1646}
1647
1648fn owner_token() -> String {
1651 static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
1652 let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1653 let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
1654 format!("{}-{:016x}", std::process::id(), r.next_u64())
1655}
1656
1657fn break_marker(path: &Path, token: &str) -> PathBuf {
1660 let mut h: u64 = 0xcbf29ce484222325;
1661 for b in token.bytes() {
1662 h ^= u64::from(b);
1663 h = h.wrapping_mul(0x100000001b3);
1664 }
1665 let mut name = path.as_os_str().to_owned();
1666 name.push(format!(".break-{h:016x}"));
1667 PathBuf::from(name)
1668}
1669
1670fn older_than_stale(path: &Path) -> bool {
1671 std::fs::metadata(path)
1672 .and_then(|m| m.modified())
1673 .ok()
1674 .and_then(|t| t.elapsed().ok())
1675 .is_some_and(|age| age > TASK_LOCK_STALE)
1676}
1677
1678const MAX_MARKER_DEPTH: u8 = 3;
1682
1683fn take_marker(path: &Path, token: &str, depth: u8) -> Option<(PathBuf, String)> {
1689 use std::io::Write;
1690 let marker = break_marker(path, token);
1691 for _ in 0..2 {
1692 match std::fs::OpenOptions::new()
1693 .write(true)
1694 .create_new(true)
1695 .open(&marker)
1696 {
1697 Ok(mut f) => {
1698 let mine = owner_token();
1699 if f.write_all(mine.as_bytes()).is_err() {
1700 drop(f);
1701 let _ = std::fs::remove_file(&marker);
1702 return None;
1703 }
1704 return Some((marker, mine));
1705 }
1706 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
1707 let judged = std::fs::read_to_string(&marker).ok();
1708 match judged.filter(|_| older_than_stale(&marker)) {
1709 Some(judged) if depth < MAX_MARKER_DEPTH => {
1710 remove_lock_if(&marker, &judged, true, depth + 1)?;
1711 }
1712 _ => return None,
1713 }
1714 }
1715 Err(_) => return None,
1716 }
1717 }
1718 None
1719}
1720
1721fn remove_lock_if(path: &Path, token: &str, require_stale: bool, depth: u8) -> Option<bool> {
1726 let (marker, mine) = take_marker(path, token, depth)?;
1727 let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
1728 && (!require_stale || older_than_stale(path));
1729 if still {
1730 let _ = std::fs::remove_file(path);
1731 }
1732 if depth >= MAX_MARKER_DEPTH {
1734 if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
1735 let _ = std::fs::remove_file(&marker);
1736 }
1737 } else {
1738 let _ = remove_lock_if(&marker, &mine, false, depth + 1);
1739 }
1740 Some(still)
1741}
1742
1743fn break_stale(path: &Path, judged: &str) -> bool {
1746 remove_lock_if(path, judged, true, 0).unwrap_or(false)
1747}
1748
1749impl Drop for TaskLock {
1750 fn drop(&mut self) {
1751 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
1752 loop {
1753 match remove_lock_if(&self.path, &self.token, false, 0) {
1754 Some(_) => return,
1755 None if std::time::Instant::now() > deadline => return,
1756 None => std::thread::sleep(std::time::Duration::from_millis(5)),
1757 }
1758 }
1759 }
1760}
1761
1762#[derive(Debug, Clone)]
1764pub struct Queue {
1765 root: PathBuf,
1766}
1767
1768impl Queue {
1769 pub fn open() -> Self {
1771 Self::at(crate::run::home().join("queue"))
1772 }
1773
1774 pub fn at(root: PathBuf) -> Self {
1777 Self { root }
1778 }
1779
1780 pub fn root(&self) -> &Path {
1782 &self.root
1783 }
1784
1785 pub fn path_of(&self, id: &str) -> PathBuf {
1787 self.root.join(format!("{id}.json"))
1788 }
1789
1790 pub fn attachments_dir(&self, id: &str) -> PathBuf {
1793 self.root.join(format!("{id}.attachments"))
1794 }
1795
1796 pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
1807 let mut wanted = Vec::new();
1808 for src in sources {
1809 let name = src
1810 .file_name()
1811 .and_then(|n| n.to_str())
1812 .with_context(|| format!("`{}` has no usable file name", src.display()))?;
1813 if !crate::ask::valid_asset_name(name) {
1814 bail!(
1815 "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
1816 with no `..`; rename the file and try again"
1817 );
1818 }
1819 if !src.is_file() {
1820 bail!("attachment `{}` is not a file", src.display());
1821 }
1822 wanted.push((src, name));
1823 }
1824 let dir = self.attachments_dir(&task.id);
1825 let existed = dir.is_dir();
1826 std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
1827 let mut created: Vec<PathBuf> = Vec::new();
1828 let mut names = Vec::new();
1829 let mut copy_all = || -> Result<()> {
1830 for (src, name) in &wanted {
1831 let (stored, path) = copy_new(&dir, src, name)?;
1832 created.push(path);
1833 names.push(stored);
1834 }
1835 Ok(())
1836 };
1837 if let Err(e) = copy_all() {
1838 for path in &created {
1839 let _ = std::fs::remove_file(path);
1840 }
1841 if !existed {
1842 let _ = std::fs::remove_dir(&dir);
1843 }
1844 return Err(e);
1845 }
1846 task.attachments.extend(names.iter().cloned());
1847 Ok(names)
1848 }
1849
1850 pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
1861 let dir = self.attachments_dir(&task.id);
1862 let existed = dir.is_dir();
1863 let before = task.attachments.len();
1864 let names = self.attach(task, sources)?;
1865 if let Err(e) = self.put(task) {
1866 for name in &names {
1867 let _ = std::fs::remove_file(dir.join(name));
1868 }
1869 if !existed {
1870 let _ = std::fs::remove_dir(&dir);
1871 }
1872 task.attachments.truncate(before);
1873 return Err(e);
1874 }
1875 Ok(names)
1876 }
1877
1878 pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
1882 let dir = self.attachments_dir(&task.id);
1883 task.attachments
1884 .iter()
1885 .map(|n| {
1886 let p = dir.join(n);
1887 std::path::absolute(&p).unwrap_or(p)
1888 })
1889 .collect()
1890 }
1891
1892 pub fn put(&self, task: &mut Task) -> Result<()> {
1901 let _lock = self.lock_task(&task.id)?;
1902 self.put_unlocked(task)
1903 }
1904
1905 pub fn create_new(&self, task: &mut Task) -> Result<bool> {
1910 let _lock = self.lock_task(&task.id)?;
1911 if self.path_of(&task.id).exists() {
1912 return Ok(false);
1913 }
1914 self.put_unlocked(task)?;
1915 Ok(true)
1916 }
1917
1918 fn put_unlocked(&self, task: &mut Task) -> Result<()> {
1920 if let Ok(stored) = read_path(&self.path_of(&task.id)) {
1921 for run in stored.runs {
1922 if !task.runs.contains(&run) {
1923 task.runs.push(run);
1924 }
1925 }
1926 for p in stored.passes {
1927 if !task
1928 .passes
1929 .iter()
1930 .any(|q| q.run == p.run && q.pass_index == p.pass_index)
1931 {
1932 task.passes.push(p);
1933 }
1934 }
1935 if let Some(n) = stored.chat_notice
1938 && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
1939 {
1940 task.chat_notice = Some(n);
1941 }
1942 }
1943 task.updated_at = Timestamp::now();
1944 std::fs::create_dir_all(&self.root)
1945 .with_context(|| format!("create {}", self.root.display()))?;
1946 let body = serde_json::to_string_pretty(task).context("serialize task")?;
1947 let path = self.path_of(&task.id);
1948 let tmp = path.with_extension("json.tmp");
1949 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
1950 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
1951 if let (Some(notice), Some(home)) = (
1956 crate::notices::task_held(task),
1957 self.root.parent().filter(|p| !p.as_os_str().is_empty()),
1958 ) {
1959 crate::notices::raise_in(home, notice);
1960 }
1961 Ok(())
1962 }
1963
1964 pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
1969 let id = self.resolve_id(id)?;
1970 let _lock = self.lock_task(&id)?;
1971 let mut task = self.get(&id)?;
1972 if !f(&mut task) {
1973 return Ok(false);
1974 }
1975 self.put_unlocked(&mut task)?;
1976 Ok(true)
1977 }
1978
1979 pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
1984 let id = self.resolve_id(id)?;
1985 let _lock = self.lock_task(&id)?;
1988 let mut task = self.get(&id)?;
1989 if task.link_run(run) {
1990 self.put_unlocked(&mut task)?;
1991 }
1992 Ok(task)
1993 }
1994
1995 fn lock_task(&self, id: &str) -> Result<TaskLock> {
2003 std::fs::create_dir_all(&self.root)
2004 .with_context(|| format!("create {}", self.root.display()))?;
2005 let path = self.root.join(format!("{id}.write-lock"));
2006 let started = std::time::Instant::now();
2007 loop {
2008 match std::fs::OpenOptions::new()
2009 .write(true)
2010 .create_new(true)
2011 .open(&path)
2012 {
2013 Ok(mut file) => {
2014 use std::io::Write;
2015 let token = owner_token();
2016 if let Err(e) = file.write_all(token.as_bytes()) {
2017 drop(file);
2018 let _ = std::fs::remove_file(&path);
2019 return Err(e).with_context(|| format!("lock {}", path.display()));
2020 }
2021 return Ok(TaskLock { path, token });
2022 }
2023 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2024 let judged = std::fs::read_to_string(&path).ok();
2027 let broken = judged
2028 .filter(|_| older_than_stale(&path))
2029 .is_some_and(|judged| break_stale(&path, &judged));
2030 if !broken {
2034 if started.elapsed() > TASK_LOCK_STALE {
2035 bail!("could not lock task {id}");
2036 }
2037 std::thread::sleep(std::time::Duration::from_millis(15));
2038 }
2039 }
2040 Err(e)
2044 if e.kind() == std::io::ErrorKind::PermissionDenied
2045 && started.elapsed() <= TASK_LOCK_STALE =>
2046 {
2047 std::thread::sleep(std::time::Duration::from_millis(15));
2048 }
2049 Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
2050 }
2051 }
2052 }
2053
2054 pub fn get(&self, id: &str) -> Result<Task> {
2056 let resolved = self.resolve_id(id)?;
2057 read_path(&self.path_of(&resolved))
2058 }
2059
2060 pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
2092 let resolved = self.resolve_id(id)?;
2093 if in_flight {
2094 bail!("task {resolved} is being run by a live daemon right now");
2095 }
2096 self.write_tombstone(&resolved)?;
2097 if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
2098 {
2099 if self.path_of(&resolved).exists() {
2103 let _ = std::fs::remove_file(self.tombstone_path(&resolved));
2104 }
2105 return Err(e);
2106 }
2107 let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
2108 Ok(Removal {
2109 id: resolved,
2110 released,
2111 still_blocked,
2112 })
2113 }
2114
2115 fn tombstone_path(&self, id: &str) -> PathBuf {
2118 self.root.join(format!("{id}.removed"))
2119 }
2120
2121 fn write_tombstone(&self, id: &str) -> Result<()> {
2122 let path = self.tombstone_path(id);
2123 let tmp = self.root.join(format!("{id}.removed.tmp"));
2124 std::fs::write(&tmp, b"")
2125 .and_then(|()| std::fs::rename(&tmp, &path))
2126 .with_context(|| format!("write {}", path.display()))
2127 }
2128
2129 fn was_deleted(&self, id: &str) -> bool {
2133 self.tombstone_path(id).is_file() && !self.path_of(id).exists()
2134 }
2135
2136 pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
2140 let deleted = deleted_blockers(self, questions, &task.blocked_by);
2141 deleted
2142 .into_iter()
2143 .filter(|id| task.dependency_deleted(id))
2144 .collect()
2145 }
2146
2147 pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
2151 let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
2152 return;
2153 };
2154 let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
2155 let waiting = dependent.status == TaskStatus::Blocked;
2156 let message = match (ja, waiting) {
2157 (true, false) => format!(
2158 "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
2159 dependent.short(),
2160 short(deleted)
2161 ),
2162 (true, true) => format!(
2163 "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
2164 dependent.short(),
2165 short(deleted)
2166 ),
2167 (false, false) => format!(
2168 "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
2169 dependent.short(),
2170 short(deleted)
2171 ),
2172 (false, true) => format!(
2173 "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
2174 dependent.short(),
2175 short(deleted)
2176 ),
2177 };
2178 crate::notices::raise_in(
2179 home,
2180 crate::notices::Notice::info(
2181 &format!("unblocked:{}:{}", dependent.id, deleted),
2182 message,
2183 )
2184 .link(crate::notices::Link::Task {
2185 id: dependent.id.clone(),
2186 }),
2187 );
2188 }
2189
2190 fn release_dependents_of(
2195 &self,
2196 dependency: &str,
2197 questions: &Questions,
2198 ) -> (Vec<String>, Vec<String>) {
2199 let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
2200 for listed in self.list() {
2201 if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
2202 matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
2203 }) {
2204 continue;
2205 }
2206 let Ok(_claim) = self.claim(&listed.id) else {
2207 continue;
2208 };
2209 let Ok(mut task) = self.get(&listed.id) else {
2210 continue;
2211 };
2212 let references: Vec<String> = task.blocked_by.iter().filter(|id| {
2213 matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
2214 }).cloned().collect();
2215 if references.is_empty() {
2216 continue;
2217 }
2218 for reference in references {
2219 task.dependency_deleted(&reference);
2220 }
2221 if self.put(&mut task).is_err() {
2222 continue;
2223 }
2224 self.note_dependency_deleted(&task, dependency);
2225 if task.status == TaskStatus::Blocked {
2226 still_blocked.push(task.id.clone());
2227 } else {
2228 released.push(task.id.clone());
2229 }
2230 }
2231 (released, still_blocked)
2232 }
2233
2234 fn remove_record_with_attachments(
2238 &self,
2239 resolved: &str,
2240 remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
2241 ) -> Result<()> {
2242 self.sweep_removed_attachments();
2249 let attachments = self.attachments_dir(resolved);
2250 let aside = self.root.join(format!("{resolved}.attachments.removing"));
2251 let moved = match std::fs::rename(&attachments, &aside) {
2252 Ok(()) => true,
2253 Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
2254 Err(e) => {
2255 return Err(e).with_context(|| format!("remove {}", attachments.display()));
2256 }
2257 };
2258 let path = self.path_of(resolved);
2259 if let Err(e) = remove_record(&path) {
2260 if moved {
2261 let _ = std::fs::rename(&aside, &attachments);
2262 }
2263 return Err(e).with_context(|| format!("remove {}", path.display()));
2264 }
2265 if moved {
2266 if let Err(e) = std::fs::remove_dir_all(&aside) {
2267 tracing::warn!("leftover attachments {}: {e}", aside.display());
2268 }
2269 }
2270 let lock = self.lock_path(resolved);
2271 if let Err(e) = std::fs::remove_file(&lock) {
2272 if e.kind() != std::io::ErrorKind::NotFound {
2273 return Err(e).with_context(|| format!("remove {}", lock.display()));
2274 }
2275 }
2276 Ok(())
2277 }
2278
2279 fn sweep_removed_attachments(&self) {
2285 let Ok(entries) = std::fs::read_dir(&self.root) else {
2286 return;
2287 };
2288 for entry in entries.flatten() {
2289 let name = entry.file_name();
2290 let name = name.to_string_lossy();
2291 let Some(id) = name.strip_suffix(".attachments.removing") else {
2292 continue;
2293 };
2294 if !self.path_of(id).exists() {
2297 if let Err(e) = std::fs::remove_dir_all(entry.path()) {
2298 tracing::warn!("leftover attachments {}: {e}", entry.path().display());
2299 }
2300 }
2301 }
2302 self.sweep_tombstones();
2303 }
2304
2305 fn sweep_tombstones(&self) {
2314 let Ok(entries) = std::fs::read_dir(&self.root) else {
2315 return;
2316 };
2317 let tasks = self.list();
2318 for entry in entries.flatten() {
2319 let name = entry.file_name();
2320 let name = name.to_string_lossy();
2321 let Some(id) = name.strip_suffix(".removed") else {
2322 continue;
2323 };
2324 if self.path_of(id).exists()
2325 || tasks.iter().any(|t| {
2326 t.blocked_by
2327 .iter()
2328 .any(|b| id.starts_with(b) || id.ends_with(b))
2329 })
2330 {
2331 continue;
2332 }
2333 let old = entry
2334 .metadata()
2335 .and_then(|m| m.modified())
2336 .ok()
2337 .and_then(|m| m.elapsed().ok())
2338 .is_some_and(|age| age >= TOMBSTONE_GRACE);
2339 if old {
2340 let _ = std::fs::remove_file(entry.path());
2341 }
2342 }
2343 }
2344
2345 fn lock_path(&self, id: &str) -> PathBuf {
2348 self.root.join(format!("{id}.lock"))
2349 }
2350
2351 pub fn list(&self) -> Vec<Task> {
2364 let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
2365 .into_iter()
2366 .flatten()
2367 .flatten()
2368 .map(|e| e.path())
2369 .filter(|p| p.extension().is_some_and(|x| x == "json"))
2370 .filter_map(|p| read_path(&p).ok())
2371 .collect();
2372 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
2373 tasks
2374 }
2375
2376 pub fn superseded(&self) -> HashMap<String, String> {
2389 let mut by = HashMap::new();
2390 for task in self.list() {
2391 for earlier in &task.runs {
2392 if let Some(later) = task.successor_of(earlier) {
2393 by.insert(earlier.clone(), later.clone());
2394 }
2395 }
2396 }
2397 by
2398 }
2399
2400 pub fn superseded_by(&self, run: &str) -> Option<String> {
2408 for task in self.list() {
2409 if task.runs.iter().any(|r| r == run) {
2410 return task.successor_of(run).cloned();
2411 }
2412 }
2413 None
2414 }
2415
2416 pub fn latest_attempt(&self, run: &str) -> Option<String> {
2427 for task in self.list() {
2428 if task.runs.iter().any(|r| r == run) {
2429 return task.runs.last().filter(|last| **last != run).cloned();
2430 }
2431 }
2432 None
2433 }
2434
2435 pub fn next_runnable(&self) -> Option<Task> {
2440 let mut runnable: Vec<Task> = self
2441 .list()
2442 .into_iter()
2443 .filter(|t| t.status.runnable())
2444 .collect();
2445 runnable.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
2446 runnable.into_iter().next()
2447 }
2448
2449 pub fn claim(&self, id: &str) -> Result<Claim> {
2456 std::fs::create_dir_all(&self.root)
2457 .with_context(|| format!("create {}", self.root.display()))?;
2458 let path = self.lock_path(id);
2459 match std::fs::OpenOptions::new()
2460 .write(true)
2461 .create_new(true)
2462 .open(&path)
2463 {
2464 Ok(mut f) => {
2465 use std::io::Write as _;
2466 let _ = writeln!(f, "{}", std::process::id());
2468 Ok(Claim { path })
2469 }
2470 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2471 bail!("task {id} is already claimed ({} exists)", path.display())
2472 }
2473 Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
2474 }
2475 }
2476
2477 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
2479 let ids = record_ids(&self.root, "json")?;
2480 match resolve_reference(prefix, ids.iter().map(String::as_str)) {
2481 Reference::Found(id) => Ok(id),
2482 Reference::Missing => bail!("no task matches `{prefix}`"),
2483 Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
2484 }
2485 }
2486
2487 pub fn revision(&self) -> u64 {
2494 self.revision_excluding(&std::collections::BTreeSet::new())
2495 }
2496
2497 pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
2500 use std::hash::{Hash as _, Hasher as _};
2501
2502 let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
2503 .into_iter()
2504 .flatten()
2505 .flatten()
2506 .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
2507 .filter(|e| {
2508 let path = e.path();
2509 !path
2510 .file_stem()
2511 .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
2512 })
2513 .filter_map(|e| {
2514 let name = e.file_name().to_string_lossy().into_owned();
2515 let mtime = e
2516 .metadata()
2517 .ok()?
2518 .modified()
2519 .ok()?
2520 .duration_since(std::time::UNIX_EPOCH)
2521 .ok()?
2522 .as_millis() as u64;
2523 Some((name, mtime))
2524 })
2525 .collect();
2526
2527 if entries.is_empty() {
2528 return 0;
2529 }
2530
2531 entries.sort_unstable();
2532 let mut hasher = std::hash::DefaultHasher::new();
2533 for (name, mtime) in &entries {
2534 name.hash(&mut hasher);
2535 mtime.hash(&mut hasher);
2536 }
2537 let h = hasher.finish();
2538 if h == 0 { 1 } else { h }
2539 }
2540}
2541
2542const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
2545
2546#[derive(Debug, Clone)]
2548pub struct Removal {
2549 pub id: String,
2551 pub released: Vec<String>,
2555 pub still_blocked: Vec<String>,
2558}
2559
2560#[derive(Debug)]
2562pub struct Claim {
2563 path: PathBuf,
2564}
2565
2566impl Drop for Claim {
2567 fn drop(&mut self) {
2568 let _ = std::fs::remove_file(&self.path);
2569 }
2570}
2571
2572pub fn title_from(instruction: &str, max: usize) -> String {
2575 let Some(line) = first_line(instruction) else {
2576 return "(empty task)".to_owned();
2577 };
2578 if line.chars().count() <= max {
2579 return line.to_owned();
2580 }
2581 let head: String = line.chars().take(max.saturating_sub(1)).collect();
2582 format!("{head}…")
2583}
2584
2585pub(crate) fn first_line(instruction: &str) -> Option<&str> {
2591 let line = instruction
2592 .lines()
2593 .map(str::trim)
2594 .find(|l| !l.is_empty())?
2595 .trim_start_matches(['#', '-', '*', '>', ' '])
2596 .trim();
2597 (!line.is_empty()).then_some(line)
2598}
2599
2600fn read_path(path: &Path) -> Result<Task> {
2601 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
2602 let task: Task =
2603 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2604 if task.schema > SCHEMA {
2610 bail!(
2611 "task {} was written by a different magi (schema {}, this build \
2612 speaks {SCHEMA})",
2613 task.id,
2614 task.schema
2615 );
2616 }
2617 Ok(task)
2618}
2619
2620#[derive(Debug, PartialEq, Eq)]
2623pub(crate) enum Reference {
2624 Found(String),
2625 Missing,
2626 Ambiguous,
2627}
2628
2629pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
2630 let mut hits = Vec::new();
2631 for candidate in ids {
2632 if candidate == id {
2633 return Reference::Found(candidate.to_owned());
2634 }
2635 if candidate.starts_with(id) || candidate.ends_with(id) {
2636 hits.push(candidate);
2637 }
2638 }
2639 match hits.as_slice() {
2640 [] => Reference::Missing,
2641 [one] => Reference::Found((*one).to_owned()),
2642 _ => Reference::Ambiguous,
2643 }
2644}
2645
2646pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
2649 let entries = match std::fs::read_dir(root) {
2650 Ok(entries) => entries,
2651 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
2652 Err(e) => return Err(e.into()),
2653 };
2654 let mut ids = Vec::new();
2655 for entry in entries {
2656 let path = entry?.path();
2657 if path.extension().is_some_and(|e| e == extension)
2658 && let Some(id) = path.file_stem().and_then(|s| s.to_str())
2659 {
2660 ids.push(id.to_owned());
2661 }
2662 }
2663 Ok(ids)
2664}
2665
2666#[derive(Debug, PartialEq, Eq)]
2669pub enum Dependency {
2670 Task(String),
2672 Question(String),
2674 Deleted(String),
2676 Missing,
2678 Ambiguous,
2680 Unknown,
2682}
2683
2684pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
2686 let (Ok(tasks), Ok(qs), Ok(removed)) = (
2687 record_ids(queue.root(), "json"),
2688 record_ids(questions.root(), "json"),
2689 record_ids(queue.root(), "removed"),
2690 ) else {
2691 return Dependency::Unknown;
2692 };
2693 let mut ids = tasks
2696 .iter()
2697 .chain(qs.iter())
2698 .chain(removed.iter())
2699 .cloned()
2700 .collect::<Vec<_>>();
2701 ids.sort();
2702 ids.dedup();
2703 match resolve_reference(id, ids.iter().map(String::as_str)) {
2704 Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
2705 Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
2706 Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
2707 Reference::Found(_) => Dependency::Unknown,
2708 Reference::Missing => Dependency::Missing,
2709 Reference::Ambiguous => Dependency::Ambiguous,
2710 }
2711}
2712
2713pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
2715 for id in ids {
2716 if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
2717 dependency(queue, questions, id)
2718 {
2719 *id = full;
2720 }
2721 }
2722}
2723
2724pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
2727 let inventory = crate::blockers::Inventory::load(queue, questions);
2728 let mut text = String::from(
2729 "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
2730 );
2731 match record_ids(queue.root(), "json") {
2732 Ok(ids) => {
2733 for id in ids {
2734 match queue.get(&id) {
2735 Ok(task) => text.push_str(&format!(
2736 "{}: {}; blocked_by={:?}; chain={:?}\n",
2737 task.id,
2738 task.status.as_str(),
2739 task.blocked_by,
2740 inventory.waits_on(&task)
2741 )),
2742 Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
2743 }
2744 }
2745 }
2746 Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
2747 }
2748 text
2749}
2750
2751pub fn missing_blockers(
2753 queue: &Queue,
2754 questions: &Questions,
2755 blocked_by: &[String],
2756) -> Vec<String> {
2757 blocked_by
2758 .iter()
2759 .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
2760 .cloned()
2761 .collect()
2762}
2763
2764pub fn deleted_blockers(
2767 queue: &Queue,
2768 questions: &Questions,
2769 blocked_by: &[String],
2770) -> Vec<String> {
2771 blocked_by
2772 .iter()
2773 .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
2774 .cloned()
2775 .collect()
2776}
2777
2778pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
2790 missing_blocker_hold_reason_in(blocked_by, missing, "en")
2791}
2792
2793pub fn missing_blocker_hold_reason_in(
2795 blocked_by: &[String],
2796 missing: &[String],
2797 language: &str,
2798) -> String {
2799 if crate::lang::is_japanese(language) {
2800 format!(
2801 "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
2802 blocked_by.join(", "),
2803 missing.join(", "),
2804 )
2805 } else {
2806 format!(
2807 "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
2808 blocked_by.join(", "),
2809 missing.join(", "),
2810 )
2811 }
2812}
2813
2814pub fn short(id: &str) -> &str {
2816 id.split('-').next_back().unwrap_or(id)
2817}
2818
2819fn new_id() -> String {
2820 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
2821 let seed = crate::rng::entropy();
2822 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
2823}
2824
2825#[cfg(test)]
2826mod tests {
2827 #[test]
2828 fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
2829 let t = Task::new(
2830 "retry".into(),
2831 "do it".into(),
2832 PathBuf::from("."),
2833 Source::Human,
2834 );
2835 let mut value = serde_json::to_value(&t).unwrap();
2836 for field in ["resumption", "resumption_history", "operator_decided_at"] {
2837 value.as_object_mut().unwrap().remove(field);
2838 }
2839 value["schema"] = serde_json::json!(15);
2840 let mut legacy: Task = serde_json::from_value(value).unwrap();
2841 legacy.hold_machine(Some("automatic retry".into()));
2842 legacy.release();
2843 assert!(legacy.resumption.is_none());
2844 assert!(legacy.resumption_history.is_empty());
2845 assert!(legacy.operator_decided_at.is_none());
2846 legacy.release_operator("cli", None);
2847 legacy.start("run".into());
2848 let reloaded: Task =
2849 serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
2850 assert!(reloaded.resume_override.is_none());
2851 assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
2852 }
2853
2854 #[test]
2855 fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
2856 let mut base = RunOverrides {
2857 merge: Some("pr".to_owned()),
2858 candidates: Some(3),
2859 ..RunOverrides::default()
2860 };
2861 base.merge_over(&RunOverrides {
2862 merge: Some("none".to_owned()),
2863 seed: Some(7),
2864 ..RunOverrides::default()
2865 });
2866 assert_eq!(base.merge.as_deref(), Some("none"));
2867 assert_eq!(base.candidates, Some(3));
2868 assert_eq!(base.seed, Some(7));
2869 }
2870
2871 #[test]
2872 fn an_already_landed_task_is_done_with_its_attempt_refunded() {
2873 let mut t = task("relanded");
2874 t.attempts = 1;
2875 t.status = TaskStatus::Running;
2876 t.already_landed("already in main as 0e368de");
2877 assert_eq!(t.status, TaskStatus::Done);
2878 assert_eq!(t.attempts, 0);
2879 assert!(t.hold_reason.is_none());
2880 assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
2881 }
2882
2883 #[test]
2884 fn missing_blocker_reason_follows_the_language() {
2885 let b = vec!["a".to_owned()];
2886 let en = missing_blocker_hold_reason_in(&b, &b, "en");
2887 assert_eq!(en, missing_blocker_hold_reason(&b, &b));
2888 assert!(en.starts_with("blocked on a"));
2889 assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
2890 assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
2891 }
2892
2893 use super::*;
2894
2895 #[test]
2896 fn triage_applied_survives_release_and_old_records_read_as_empty() {
2897 let mut t = Task::new(
2898 "t".to_owned(),
2899 "i".to_owned(),
2900 PathBuf::from("r"),
2901 Source::Human,
2902 );
2903 t.mark_triage_applied("q1");
2904 t.mark_triage_applied("q1");
2905 t.hold_machine(Some("x".to_owned()));
2906 t.release();
2907 assert_eq!(t.triage_applied, ["q1"]);
2908 assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
2909
2910 let mut v = serde_json::to_value(&t).unwrap();
2911 v.as_object_mut().unwrap().remove("triage_applied");
2912 let old: Task = serde_json::from_value(v).unwrap();
2913 assert!(old.triage_applied.is_empty());
2914 }
2915
2916 #[test]
2917 fn task_counts_of_empty_is_all_zero() {
2918 assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
2919 }
2920
2921 #[test]
2922 fn task_counts_of_tallies_every_status() {
2923 let mut queued = Task::new(
2924 "q".to_owned(),
2925 "i".to_owned(),
2926 PathBuf::from("."),
2927 Source::Human,
2928 );
2929 queued.status = TaskStatus::Queued;
2930 let mut running = queued.clone();
2931 running.status = TaskStatus::Running;
2932 let mut done = queued.clone();
2933 done.status = TaskStatus::Done;
2934 let mut failed = queued.clone();
2935 failed.status = TaskStatus::Failed;
2936 let mut held = queued.clone();
2937 held.status = TaskStatus::Held;
2938 let mut blocked = queued.clone();
2939 blocked.status = TaskStatus::Blocked;
2940
2941 let mut parked = queued.clone();
2942 parked.status = TaskStatus::Parked;
2943
2944 let counts = TaskCounts::of(&[
2945 queued,
2946 running,
2947 done.clone(),
2948 done,
2949 failed,
2950 held,
2951 blocked,
2952 parked,
2953 ]);
2954 assert_eq!(
2955 counts,
2956 TaskCounts {
2957 queued: 1,
2958 running: 1,
2959 done: 2,
2960 failed: 1,
2961 held: 1,
2962 blocked: 1,
2963 parked: 1,
2964 }
2965 );
2966 }
2967
2968 #[test]
2969 fn task_implementers_persist_normalized_without_losing_history() {
2970 let (_dir, q) = queue();
2971 let mut t = task("keep me");
2972 t.hold_manual(Some("waiting".into()));
2973 t.runs.push("earlier-run".into());
2974 t.attachments.push("reference.txt".into());
2975 let cfg: crate::config::Config =
2976 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
2977 .unwrap();
2978 q.put(&mut t).unwrap();
2979 let original = serde_json::to_value(&t).unwrap();
2980 for n in [2, 1, 2] {
2981 t.set_implementers(n, &cfg).unwrap();
2982 q.put(&mut t).unwrap();
2983 t = q.get(&t.id).unwrap();
2984 assert_eq!(t.implementers(), Some(usize::from(n)));
2985 assert_eq!(t.solo, n == 1);
2986 if n == 1 {
2987 assert!(t.overrides.is_none());
2988 }
2989 let mut actual = serde_json::to_value(&t).unwrap();
2990 for field in ["solo", "overrides", "updated_at"] {
2991 actual[field] = original[field].clone();
2992 }
2993 assert_eq!(actual, original);
2994 }
2995 t.overrides.as_mut().unwrap().judges = Some(2);
2996 t.set_implementers(1, &cfg).unwrap();
2997 assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
2998 assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
2999 }
3000
3001 #[test]
3002 fn task_implementers_old_records_keep_inheritance_until_changed() {
3003 let mut value = serde_json::to_value(task("old task")).unwrap();
3004 value.as_object_mut().unwrap().remove("solo");
3005 value.as_object_mut().unwrap().remove("overrides");
3006 let mut old: Task = serde_json::from_value(value).unwrap();
3007 assert_eq!(old.implementers(), None);
3008 old.set_implementers(1, &crate::config::Config::default())
3009 .unwrap();
3010 assert!(old.solo);
3011 assert!(old.overrides.is_none());
3012 }
3013
3014 #[test]
3015 fn task_implementers_reject_invalid_counts_and_states() {
3016 let cfg = crate::config::Config::default();
3017 let mut t = task("unchanged");
3018 for n in [0, 4, 255] {
3019 assert!(t.set_implementers(n, &cfg).is_err());
3020 }
3021 for status in [
3022 TaskStatus::Running,
3023 TaskStatus::Done,
3024 TaskStatus::Failed,
3025 TaskStatus::Blocked,
3026 TaskStatus::Parked,
3027 ] {
3028 t.status = status;
3029 let before = serde_json::to_value(&t).unwrap();
3030 assert!(t.set_implementers(1, &cfg).is_err());
3031 assert_eq!(serde_json::to_value(&t).unwrap(), before);
3032 }
3033 }
3034
3035 fn queue() -> (tempfile::TempDir, Queue) {
3038 let dir = tempfile::tempdir().unwrap();
3039 let q = Queue::at(dir.path().join("queue"));
3040 (dir, q)
3041 }
3042
3043 #[test]
3044 fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
3045 let dir = tempfile::tempdir().unwrap();
3046 let q = Queue::at(dir.path().join("queue"));
3047 let mut t = task("held");
3048 q.put(&mut t).unwrap();
3049 assert_eq!(
3050 crate::notices::Notices::at(dir.path().join("notifications"))
3051 .list()
3052 .len(),
3053 0
3054 );
3055 t.hold_machine(Some("out of attempts".to_owned()));
3056 q.put(&mut t).unwrap();
3057 let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
3058 assert_eq!(listed.len(), 1);
3059 assert!(listed[0].message.contains("out of attempts"));
3060 }
3061
3062 fn task(title: &str) -> Task {
3063 Task::new(
3064 title.to_owned(),
3065 format!("do {title}"),
3066 PathBuf::from("."),
3067 Source::Human,
3068 )
3069 }
3070
3071 #[test]
3072 fn attempt_limit_inherits_old_records_and_controls_failure() {
3073 let mut t = task("budget");
3074 let mut old = serde_json::to_value(&t).unwrap();
3075 old.as_object_mut().unwrap().remove("attempt_limit");
3076 old["schema"] = serde_json::json!(15);
3077 let legacy: Task = serde_json::from_value(old).unwrap();
3078 assert_eq!(legacy.attempt_limit, None);
3079 assert_eq!(legacy.effective_attempt_limit(2), 2);
3080 t.attempts = 2;
3081 t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
3082 t.fail("retry", 1);
3083 assert_eq!(t.status, TaskStatus::Failed);
3084 t.attempts = 3;
3085 t.fail("stop", 10);
3086 assert_eq!(t.status, TaskStatus::Held);
3087 t.change_attempt_limit(None).unwrap();
3088 assert_eq!(t.status, TaskStatus::Held);
3089 assert_eq!(t.attempts, 3);
3090 t.fail("inherit", 4);
3091 assert_eq!(t.status, TaskStatus::Failed);
3092 for status in [
3093 TaskStatus::Running,
3094 TaskStatus::Done,
3095 TaskStatus::Failed,
3096 TaskStatus::Blocked,
3097 TaskStatus::Parked,
3098 ] {
3099 t.status = status;
3100 assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
3101 assert_eq!(t.attempt_limit, None);
3102 }
3103 }
3104
3105 #[test]
3106 fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
3107 let mut t = task("retried");
3108 assert!(
3109 t.earlier_attempts().is_empty(),
3110 "a first attempt takes nothing over"
3111 );
3112 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3113 assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
3114 assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
3116 assert_eq!(t.successor_of("bbbb"), None);
3117 assert_eq!(t.successor_of("zzzz"), None);
3118 }
3119
3120 #[test]
3121 fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
3122 let (_dir, q) = queue();
3123 let mut t = task("resumed");
3124 for (runs, successor) in [
3125 (vec!["aaaa", "aaaa"], None),
3126 (vec!["aaaa", "bbbb", "aaaa"], None),
3127 (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
3128 (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
3129 ] {
3130 t.runs = runs.into_iter().map(str::to_owned).collect();
3131 q.put(&mut t).unwrap();
3132
3133 assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
3134 assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
3135 assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
3136 }
3137 }
3138
3139 #[test]
3140 fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
3141 let (_dir, q) = queue();
3142 let mut t = task("retried");
3143 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3144 q.put(&mut t).unwrap();
3145
3146 assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
3147 assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
3148 assert_eq!(
3149 q.superseded_by("cccc"),
3150 None,
3151 "the latest attempt replaces nothing"
3152 );
3153 assert_eq!(
3154 q.superseded_by("never-heard-of-it"),
3155 None,
3156 "a run belonging to no task on this queue is not superseded"
3157 );
3158
3159 let mut by = HashMap::new();
3160 by.insert("aaaa".to_owned(), "bbbb".to_owned());
3161 by.insert("bbbb".to_owned(), "cccc".to_owned());
3162 assert_eq!(
3163 q.superseded(),
3164 by,
3165 "the whole-map and single-run forms must agree"
3166 );
3167 }
3168
3169 #[test]
3170 fn superseded_attempts_is_empty_until_the_task_is_done() {
3171 let mut t = task("retried");
3172 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3173 t.status = TaskStatus::Failed;
3174 assert_eq!(
3175 t.superseded_attempts(true),
3176 &[] as &[String],
3177 "a task still retrying has no attempt yet that a later one made moot"
3178 );
3179
3180 t.status = TaskStatus::Running;
3181 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3182 }
3183
3184 #[test]
3185 fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
3186 let mut t = task("retried");
3187 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3188 t.status = TaskStatus::Done;
3189 assert_eq!(
3190 t.superseded_attempts(true),
3191 &["aaaa".to_owned(), "bbbb".to_owned()],
3192 "cccc is the attempt whose success made the task done, and stays out"
3193 );
3194 }
3195
3196 #[test]
3197 fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
3198 let mut t = task("first try landed");
3199 t.runs = vec!["aaaa".to_owned()];
3200 t.status = TaskStatus::Done;
3201 assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3202 }
3203
3204 #[test]
3205 fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
3206 let mut t = task("closed by hand after a manual merge");
3213 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3214 t.status = TaskStatus::Done;
3215 assert_eq!(
3216 t.superseded_attempts(false),
3217 &[] as &[String],
3218 "nothing here is provably why the task is done, so nothing is superseded"
3219 );
3220 }
3221
3222 #[test]
3223 fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
3224 let (_dir, q) = queue();
3225 let mut t = task("retried twice");
3226 t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3227 q.put(&mut t).unwrap();
3228
3229 assert_eq!(
3230 q.latest_attempt("aaaa"),
3231 Some("cccc".to_owned()),
3232 "an old attempt points straight at the chain's current head, not the \
3233 next attempt in the middle of it"
3234 );
3235 assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
3236 assert_eq!(
3237 q.latest_attempt("cccc"),
3238 None,
3239 "the latest attempt is not superseded by anything"
3240 );
3241 assert_eq!(
3242 q.latest_attempt("never-heard-of-it"),
3243 None,
3244 "a run belonging to no task on this queue is not superseded"
3245 );
3246 }
3247
3248 #[test]
3249 fn a_markdown_heading_is_the_title_not_decoration() {
3250 assert_eq!(
3255 title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
3256 "Rework the config loader"
3257 );
3258 assert_eq!(title_from("- fix the thing", 40), "fix the thing");
3259 assert_eq!(title_from("> quoted task", 40), "quoted task");
3260 assert_eq!(title_from(" \n\n", 40), "(empty task)");
3262 assert_eq!(title_from("###\n", 40), "(empty task)");
3263 }
3264
3265 #[test]
3266 fn a_long_title_is_elided_by_characters_not_bytes() {
3267 let long = "課題".repeat(30);
3269 let title = title_from(&long, 10);
3270 assert_eq!(title.chars().count(), 10);
3271 assert!(title.ends_with('…'));
3272 }
3273
3274 #[test]
3275 fn priority_wins_and_ties_break_oldest_first() {
3276 let (_dir, q) = queue();
3277 let mut a = task("first");
3278 let mut b = task("second");
3279 let mut c = task("urgent");
3280 a.id = "20260101-000001-aaaa".to_owned();
3282 b.id = "20260101-000002-bbbb".to_owned();
3283 c.id = "20260101-000003-cccc".to_owned();
3284 c.priority = 5;
3285 for t in [&mut a, &mut b, &mut c] {
3286 q.put(t).unwrap();
3287 }
3288
3289 assert_eq!(q.next_runnable().unwrap().id, c.id);
3291 c.hold_machine(None);
3292 q.put(&mut c).unwrap();
3293 assert_eq!(q.next_runnable().unwrap().id, a.id);
3295 assert_eq!(q.list().len(), 3, "b is still waiting its turn");
3296 }
3297
3298 #[test]
3299 fn a_blocked_task_never_starves_another_runnable_one() {
3300 let (_dir, q) = queue();
3301 let mut blocked = task("blocked");
3302 blocked.block(vec!["something".to_owned()], None);
3303 q.put(&mut blocked).unwrap();
3304
3305 let mut runnable = task("free to go");
3306 q.put(&mut runnable).unwrap();
3307
3308 let next = q.next_runnable().expect("a runnable task is still offered");
3309 assert_eq!(next.id, runnable.id);
3310 }
3311
3312 #[test]
3313 fn a_held_task_is_never_offered_to_the_loop() {
3314 let (_dir, q) = queue();
3315 let mut t = task("held");
3316 q.put(&mut t).unwrap();
3317 assert!(q.next_runnable().is_some());
3318
3319 t.hold_machine(None);
3320 q.put(&mut t).unwrap();
3321 assert!(
3322 q.next_runnable().is_none(),
3323 "a held task must wait for a human"
3324 );
3325
3326 t.status = TaskStatus::Failed;
3328 q.put(&mut t).unwrap();
3329 assert!(q.next_runnable().is_some());
3330 }
3331
3332 #[test]
3333 fn attempts_are_capped_and_then_the_task_is_held() {
3334 let mut t = task("doomed");
3335
3336 t.start("run-1".to_owned());
3337 t.fail("gate red", 2);
3338 assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
3339
3340 t.start("run-2".to_owned());
3341 t.fail("gate red", 2);
3342 assert_eq!(
3343 t.status,
3344 TaskStatus::Held,
3345 "out of attempts: stop spending money on it"
3346 );
3347 assert_eq!(t.runs, ["run-1", "run-2"]);
3348 assert_eq!(t.last_error.as_deref(), Some("gate red"));
3349 assert_eq!(
3350 t.hold_reason.as_deref(),
3351 Some("gate red"),
3352 "the hold must say why, not leave hold_reason null next to a \
3353 populated last_error"
3354 );
3355 }
3356
3357 #[test]
3358 fn handing_off_a_task_records_a_hold_reason_too() {
3359 let mut t = task("left a pull request");
3360 t.start("run-1".to_owned());
3361 t.handed_off("run ended with a pull request open [run run-1]");
3362 assert_eq!(t.status, TaskStatus::Held);
3363 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3364 assert_eq!(
3365 t.hold_reason.as_deref(),
3366 Some("run ended with a pull request open [run run-1]")
3367 );
3368 assert_eq!(t.hold_reason, t.last_error);
3369 }
3370
3371 #[test]
3372 fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
3373 let mut t = task("stalled by quota");
3374
3375 t.start("run-1".to_owned());
3376 assert_eq!(t.attempts, 1);
3377 t.stall("judge-1, judge-2 out of quota");
3378 assert_eq!(
3379 t.attempts, 0,
3380 "a closed quota window must not spend the task's retry budget"
3381 );
3382 assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
3383 assert_eq!(
3384 t.last_error.as_deref(),
3385 Some("judge-1, judge-2 out of quota")
3386 );
3387
3388 for _ in 0..20 {
3391 t.start("run-n".to_owned());
3392 t.stall("still out of quota");
3393 }
3394 t.start("run-real".to_owned());
3395 t.fail("gate red", 2);
3396 assert_eq!(
3397 t.status,
3398 TaskStatus::Failed,
3399 "the first attempt that was really judged is attempt one"
3400 );
3401 }
3402
3403 #[test]
3404 fn releasing_a_held_task_gives_it_a_real_second_chance() {
3405 let mut t = task("retry me");
3406 t.start("run-1".to_owned());
3407 t.fail("gate red", 1);
3408 assert_eq!(t.status, TaskStatus::Held);
3409
3410 t.release();
3411 assert_eq!(t.status, TaskStatus::Queued);
3412 assert_eq!(t.attempts, 0);
3415 assert!(t.last_error.is_none());
3416 assert_eq!(
3417 t.runs.len(),
3418 1,
3419 "history is kept: attempts reset, evidence does not"
3420 );
3421 }
3422
3423 #[test]
3424 fn a_hold_reason_survives_and_a_release_clears_it() {
3425 let mut t = task("waiting on something else");
3426 t.hold_manual(Some(
3427 "waiting for 20260101-000000-aaaa to land first".to_owned(),
3428 ));
3429 assert_eq!(t.status, TaskStatus::Held);
3430 assert_eq!(
3431 t.hold_reason.as_deref(),
3432 Some("waiting for 20260101-000000-aaaa to land first")
3433 );
3434
3435 t.hold_manual(None);
3437 assert_eq!(
3438 t.hold_reason.as_deref(),
3439 Some("waiting for 20260101-000000-aaaa to land first"),
3440 "a bare re-hold keeps whatever a human already wrote down"
3441 );
3442
3443 let mut plain = task("no reason given");
3445 plain.hold_manual(None);
3446 assert_eq!(plain.status, TaskStatus::Held);
3447 assert!(plain.hold_reason.is_none());
3448
3449 t.release();
3450 assert_eq!(t.status, TaskStatus::Queued);
3451 assert!(
3452 t.hold_reason.is_none(),
3453 "a stale reason must not greet the next person who holds this task"
3454 );
3455 }
3456
3457 #[test]
3458 fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
3459 let mut t = task("landed by hand while held");
3464 t.hold_manual(Some("waiting on 3ed9".to_owned()));
3465 assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
3466
3467 t.succeed();
3468 assert_eq!(t.status, TaskStatus::Done);
3469 assert!(
3470 t.hold_reason.is_none(),
3471 "a done task cannot still be waiting on something"
3472 );
3473 }
3474
3475 #[test]
3476 fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
3477 let mut held = task("held straight out of blocked");
3484 held.block(
3485 vec!["20260101-000000-dead".to_owned()],
3486 Some("waiting on the migration script".to_owned()),
3487 );
3488 assert_eq!(held.status, TaskStatus::Blocked);
3489
3490 held.hold_manual(None);
3491 assert_eq!(held.status, TaskStatus::Held);
3492 assert!(
3493 held.blocked_by.is_empty(),
3494 "hold overrides the wait, same as release"
3495 );
3496 assert!(held.block_reason.is_none());
3497
3498 let mut done = task("closed straight out of blocked");
3499 done.block(
3500 vec!["20260101-000000-dead".to_owned()],
3501 Some("waiting on the migration script".to_owned()),
3502 );
3503 done.succeed();
3504 assert_eq!(done.status, TaskStatus::Done);
3505 assert!(
3506 done.blocked_by.is_empty(),
3507 "a done task cannot still be waiting on a dependency"
3508 );
3509 assert!(done.block_reason.is_none());
3510 }
3511
3512 #[test]
3513 fn a_blocked_task_is_never_offered_to_the_loop() {
3514 let mut t = task("blocked");
3515 assert!(t.status.runnable());
3516 t.block(
3517 vec!["dep-id".to_owned()],
3518 Some("waits on dep-id".to_owned()),
3519 );
3520 assert_eq!(t.status, TaskStatus::Blocked);
3521 assert!(!t.status.runnable());
3522 assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
3523 }
3524
3525 #[test]
3526 fn unblocking_the_last_dependency_returns_the_task_to_queued() {
3527 let mut t = task("blocked on two");
3528 t.block(
3529 vec!["a".to_owned(), "b".to_owned()],
3530 Some("waits on a and b".to_owned()),
3531 );
3532
3533 t.unblock("a");
3534 assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
3535 assert_eq!(t.blocked_by, ["b"]);
3536
3537 t.unblock("b");
3538 assert_eq!(t.status, TaskStatus::Queued);
3539 assert!(t.blocked_by.is_empty());
3540 assert!(t.block_reason.is_none());
3541 }
3542
3543 #[test]
3544 fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
3545 let mut t = task("never blocked");
3546 t.unblock("whatever");
3547 assert_eq!(t.status, TaskStatus::Queued);
3548 }
3549
3550 #[test]
3551 fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
3552 let mut t = task("held, then asked about");
3557 t.hold_machine(Some("out of attempts".to_owned()));
3558 assert_eq!(t.status, TaskStatus::Held);
3559
3560 t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
3561 assert_eq!(t.status, TaskStatus::Blocked);
3562
3563 t.record_answer("what now?".to_owned(), "leave it held".to_owned());
3564 t.unblock("q1");
3565 assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
3566 assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
3567 assert_eq!(t.hold_source, Some(HoldSource::Machine));
3568 assert!(t.blocked_from.is_none(), "consumed once restored");
3569 }
3570
3571 #[test]
3572 fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
3573 let mut t = task("manually held, then asked about");
3574 t.hold_manual(Some("waiting on a dependency".to_owned()));
3575
3576 t.block(vec!["q1".to_owned()], None);
3577 t.unblock("q1");
3578
3579 assert_eq!(t.status, TaskStatus::Held);
3580 assert_eq!(t.hold_source, Some(HoldSource::Manual));
3581 }
3582
3583 #[test]
3584 fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
3585 let mut t = task("held, blocked twice");
3589 t.hold_machine(None);
3590 t.block(vec!["q1".to_owned()], Some("first".to_owned()));
3591 t.block(
3592 vec!["q1".to_owned(), "q2".to_owned()],
3593 Some("second".to_owned()),
3594 );
3595
3596 t.unblock("q1");
3597 assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
3598 t.unblock("q2");
3599 assert_eq!(t.status, TaskStatus::Held);
3600 }
3601
3602 #[test]
3603 fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
3604 let mut t = task("blocked mid-run");
3608 t.start("run-1".to_owned());
3609 assert_eq!(t.status, TaskStatus::Running);
3610
3611 t.block(vec!["q1".to_owned()], None);
3612 t.unblock("q1");
3613 assert_eq!(t.status, TaskStatus::Queued);
3614 }
3615
3616 #[test]
3617 fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
3618 let mut t = task("legacy record, held before it was blocked");
3624 t.hold_source = Some(HoldSource::Machine);
3625 t.hold_reason = Some("legacy hold reason".to_owned());
3626 t.status = TaskStatus::Blocked;
3627 t.blocked_by = vec!["q1".to_owned()];
3628 t.blocked_from = None;
3629
3630 t.unblock("q1");
3631 assert_eq!(t.status, TaskStatus::Held);
3632 }
3633
3634 #[test]
3635 fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
3636 let mut t = task("legacy record, ordinary dependency block");
3637 t.status = TaskStatus::Blocked;
3638 t.blocked_by = vec!["dep".to_owned()];
3639 t.blocked_from = None;
3640
3641 t.unblock("dep");
3642 assert_eq!(t.status, TaskStatus::Queued);
3643 }
3644
3645 #[test]
3646 fn answering_a_question_is_recorded_and_survives_a_release() {
3647 let mut t = task("asked something");
3648 t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
3649 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
3650 t.unblock("q1");
3651 assert_eq!(t.status, TaskStatus::Queued);
3652 assert_eq!(t.answers.len(), 1);
3653 assert_eq!(t.answers[0].answer, "SQLite");
3654
3655 t.release();
3659 assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
3660 }
3661
3662 #[test]
3663 fn resumption_review_branch_requires_unchanged_authored_criteria() {
3664 for source in ["cli", "web"] {
3665 let mut t = task("review old candidate");
3666 t.request_review("candidate-A".into());
3667 t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
3668 let mut unchanged = t.clone();
3669 unchanged.release_operator(source, None);
3670 assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
3671 t.edit(
3672 "review approved head".into(),
3673 "Reuse measured PR head B; fix Clippy; review".into(),
3674 )
3675 .unwrap();
3676 t.release_operator(source, None);
3677 assert!(t.review_branch.is_none());
3678 assert!(t.review_instruction.is_none());
3679 let mut legacy = unchanged;
3681 legacy.review_instruction = None;
3682 legacy.hold_for_handover(None, "still checked out".into());
3683 legacy
3684 .edit("new plan".into(), "Reuse measured PR head B".into())
3685 .unwrap();
3686 legacy.release_operator(source, None);
3687 assert!(legacy.review_branch.is_none());
3688 }
3689 }
3690
3691 #[test]
3692 fn a_refused_handover_keeps_the_review_branch_across_release() {
3693 let mut t = task("refused takeover");
3694 t.start("run-1".to_owned());
3695 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
3696 assert_eq!(t.status, TaskStatus::Held);
3697 assert_eq!(t.attempts, 1);
3698 t.release();
3699 assert_eq!(t.status, TaskStatus::Queued);
3700 assert_eq!(t.attempts, 0);
3701 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
3702
3703 t.hold_for_handover(None, "again".to_owned());
3705 t.requeue();
3706 assert!(t.review_branch.is_none());
3707
3708 let mut m = task("manual");
3710 m.review_branch = Some("magi/x/A".to_owned());
3711 m.hold_manual(None);
3712 m.release();
3713 assert!(m.review_branch.is_none());
3714 }
3715
3716 #[test]
3717 fn requesting_review_requeues_the_task_and_remembers_the_branch() {
3718 let mut t = task("blocked run with a surviving branch");
3719 t.start("run-1".to_owned());
3720 t.fail("blocked with major findings", 5);
3721 assert_eq!(t.status, TaskStatus::Failed);
3722
3723 t.request_review("magi/eba2/A".to_owned());
3724 assert_eq!(t.status, TaskStatus::Queued);
3725 assert_eq!(t.attempts, 0);
3726 assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
3727
3728 t.release();
3730 assert!(t.review_branch.is_none());
3731 }
3732
3733 #[test]
3734 fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
3735 let mut t = task("retry");
3736 t.start("run-1".to_owned());
3737 t.requeue();
3738 assert!(t.fresh_start);
3739
3740 t.release();
3741 assert!(!t.fresh_start);
3742 }
3743
3744 #[test]
3745 fn urgent_only_changes_queued_or_held_and_preserves_settings() {
3746 for status in [
3747 TaskStatus::Queued,
3748 TaskStatus::Held,
3749 TaskStatus::Running,
3750 TaskStatus::Done,
3751 TaskStatus::Failed,
3752 TaskStatus::Blocked,
3753 TaskStatus::Parked,
3754 ] {
3755 let mut t = task("urgent settings");
3756 t.status = status;
3757 t.attempts = 2;
3758 t.hold_reason = Some("waiting".into());
3759 t.priority = 7;
3760 t.interrupt = true;
3761 let before = serde_json::to_value(&t).unwrap();
3762 for urgent in [true, true, false, false] {
3763 let result = t.set_urgent(urgent);
3764 let mut expected = before.clone();
3765 if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
3766 result.unwrap();
3767 expected["urgent"] = urgent.into();
3768 } else {
3769 let error = result.unwrap_err().to_string();
3770 assert!(error.contains(status.as_str()), "{error}");
3771 assert!(error.contains("queued or held"), "{error}");
3772 }
3773 assert_eq!(serde_json::to_value(&t).unwrap(), expected);
3774 }
3775 }
3776 }
3777
3778 #[test]
3779 fn priority_can_be_changed_while_queued_but_not_while_running() {
3780 let mut t = task("reprioritise me");
3781 t.set_priority(5).unwrap();
3782 assert_eq!(t.priority, 5);
3783
3784 t.start("run-1".to_owned());
3785 let err = t.set_priority(9).unwrap_err().to_string();
3786 assert!(err.contains("running"), "{err}");
3787 assert_eq!(t.priority, 5, "the rejected write must not partially apply");
3788 }
3789
3790 #[test]
3791 fn interrupt_can_be_marked_while_queued_but_not_while_running() {
3792 let mut t = task("interrupt me");
3793 assert!(!t.interrupt, "off unless asked, same as any other task");
3794
3795 t.set_interrupt(true).unwrap();
3796 assert!(t.interrupt);
3797
3798 t.start("run-1".to_owned());
3799 assert!(
3800 !t.interrupt,
3801 "the mark is one-shot: dispatching the task fulfils it, \
3802 whatever the run that follows ends up doing"
3803 );
3804 let err = t.set_interrupt(true).unwrap_err().to_string();
3805 assert!(err.contains("running"), "{err}");
3806 t.set_interrupt(false).unwrap();
3809 assert!(!t.interrupt);
3810 }
3811
3812 #[test]
3816 fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
3817 let mut t = task("interrupt me");
3818 t.set_interrupt(true).unwrap();
3819 t.start("run-1".to_owned());
3820 t.fail("mock failure", 5);
3821 assert_eq!(t.status, TaskStatus::Failed);
3822 assert!(
3823 !t.interrupt,
3824 "one attempt already spent the mark; a retry is an ordinary \
3825 requeue, not a fresh interrupt request"
3826 );
3827 }
3828
3829 #[test]
3830 fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
3831 let (_dir, q) = queue();
3832 let mut a = task("first filed");
3833 let mut b = task("second filed");
3834 a.id = "20260101-000001-aaaa".to_owned();
3835 b.id = "20260101-000002-bbbb".to_owned();
3836 q.put(&mut a).unwrap();
3837 q.put(&mut b).unwrap();
3838
3839 assert_eq!(
3840 q.next_runnable().unwrap().id,
3841 a.id,
3842 "with equal priority the older task goes first, so a burst of \
3843 new work cannot starve it"
3844 );
3845 assert_eq!(
3846 q.list()[0].id,
3847 b.id,
3848 "but the list an operator reads is newest first, the same as \
3849 before priority existed - a's turn to run does not make it the \
3850 newest task"
3851 );
3852
3853 let mut a = q.get(&a.id).unwrap();
3854 a.set_priority(10).unwrap();
3855 q.put(&mut a).unwrap();
3856
3857 assert_eq!(
3858 q.next_runnable().unwrap().id,
3859 a.id,
3860 "a raised priority must be reflected the moment it is saved"
3861 );
3862 assert_eq!(
3866 q.list()[0].id,
3867 a.id,
3868 "the raised task must sort first in the list an operator reads, \
3869 not only in next_runnable's own ordering"
3870 );
3871 }
3872
3873 #[test]
3874 fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
3875 let mut t = Task::new(
3876 "old title".to_owned(),
3877 "old instruction".to_owned(),
3878 PathBuf::from("/repo"),
3879 Source::Agent {
3880 run: "20260101-000000-beef".to_owned(),
3881 node: "implement".to_owned(),
3882 },
3883 );
3884 let id = t.id.clone();
3885 let created_at = t.created_at;
3886 t.runs.push("20260101-000000-beef".to_owned());
3887
3888 t.edit("new title".to_owned(), "new instruction".to_owned())
3889 .unwrap();
3890
3891 assert_eq!(t.title, "new title");
3892 assert_eq!(t.instruction, "new instruction");
3893 assert_eq!(t.id, id, "editing must not mint a new id");
3894 assert_eq!(t.created_at, created_at);
3895 assert_eq!(
3896 t.source,
3897 Source::Agent {
3898 run: "20260101-000000-beef".to_owned(),
3899 node: "implement".to_owned(),
3900 },
3901 "editing must not turn agent attribution into human"
3902 );
3903 assert_eq!(t.runs, ["20260101-000000-beef"]);
3904 }
3905
3906 #[test]
3907 fn editing_is_refused_once_a_task_is_running_or_finished() {
3908 let mut running = task("in flight");
3909 running.start("run-1".to_owned());
3910 let err = running
3911 .edit("x".to_owned(), "y".to_owned())
3912 .unwrap_err()
3913 .to_string();
3914 assert!(err.contains("running"), "{err}");
3915
3916 let mut done = task("finished");
3917 done.succeed();
3918 let err = done
3919 .edit("x".to_owned(), "y".to_owned())
3920 .unwrap_err()
3921 .to_string();
3922 assert!(err.contains("done"), "{err}");
3923
3924 let mut queued = task("waiting");
3926 queued.edit("x".to_owned(), "y".to_owned()).unwrap();
3927 let mut held = task("parked");
3928 held.hold_machine(None);
3929 held.edit("x".to_owned(), "y".to_owned()).unwrap();
3930 }
3931
3932 #[test]
3933 fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
3934 let (_dir, q) = queue();
3935 let path = q.path_of("20260101-000000-aaaa");
3936 std::fs::create_dir_all(q.root()).unwrap();
3937 std::fs::write(
3938 &path,
3939 serde_json::json!({
3940 "schema": SCHEMA,
3941 "id": "20260101-000000-aaaa",
3942 "title": "from before hold reasons existed",
3943 "instruction": "from before hold reasons existed",
3944 "repo": ".",
3945 "source": { "kind": "human" },
3946 "status": "held",
3947 "created_at": Timestamp::now().to_string(),
3948 "updated_at": Timestamp::now().to_string(),
3949 })
3950 .to_string(),
3951 )
3952 .unwrap();
3953
3954 let task = q.get("20260101-000000-aaaa").expect("must still read");
3955 assert!(task.hold_reason.is_none());
3956 assert!(task.operator_held());
3957 }
3958
3959 #[test]
3960 fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
3961 let (_dir, q) = queue();
3962 let path = q.path_of("20260101-000000-bbbb");
3963 std::fs::create_dir_all(q.root()).unwrap();
3964 std::fs::write(
3965 &path,
3966 serde_json::json!({
3967 "schema": 2,
3968 "id": "20260101-000000-bbbb",
3969 "title": "old manual recovery",
3970 "instruction": "old manual recovery",
3971 "repo": ".",
3972 "source": { "kind": "human" },
3973 "status": "held",
3974 "hold_reason": "active manual recovery run20260912-224242-daf5",
3975 "created_at": Timestamp::now().to_string(),
3976 "updated_at": Timestamp::now().to_string(),
3977 })
3978 .to_string(),
3979 )
3980 .unwrap();
3981
3982 let task = q.get("20260101-000000-bbbb").expect("must still read");
3983 assert_eq!(task.hold_source, None);
3984 assert!(task.operator_held());
3985 }
3986
3987 #[test]
3988 fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
3989 let (_dir, q) = queue();
3990 let path = q.path_of("20260101-000000-aaaa");
3991 std::fs::create_dir_all(q.root()).unwrap();
3992 std::fs::write(
3993 &path,
3994 serde_json::json!({
3995 "schema": SCHEMA,
3996 "id": "20260101-000000-aaaa",
3997 "title": "from before diagnostics existed",
3998 "instruction": "from before diagnostics existed",
3999 "repo": ".",
4000 "source": { "kind": "human" },
4001 "status": "held",
4002 "created_at": Timestamp::now().to_string(),
4003 "updated_at": Timestamp::now().to_string(),
4004 })
4005 .to_string(),
4006 )
4007 .unwrap();
4008
4009 let task = q.get("20260101-000000-aaaa").expect("must still read");
4010 assert!(task.diagnostic.is_none());
4011 }
4012
4013 #[test]
4014 fn a_schema_1_task_with_no_blocking_fields_still_reads() {
4015 let (_dir, q) = queue();
4019 let path = q.path_of("20260101-000000-aaaa");
4020 std::fs::create_dir_all(q.root()).unwrap();
4021 std::fs::write(
4022 &path,
4023 serde_json::json!({
4024 "schema": 1,
4025 "id": "20260101-000000-aaaa",
4026 "title": "from before blocking existed",
4027 "instruction": "from before blocking existed",
4028 "repo": ".",
4029 "source": { "kind": "human" },
4030 "status": "queued",
4031 "created_at": Timestamp::now().to_string(),
4032 "updated_at": Timestamp::now().to_string(),
4033 })
4034 .to_string(),
4035 )
4036 .unwrap();
4037
4038 let task = q.get("20260101-000000-aaaa").expect("must still read");
4039 assert!(task.blocked_by.is_empty());
4040 assert!(task.block_reason.is_none());
4041 assert!(task.answers.is_empty());
4042 assert!(task.review_branch.is_none());
4043 }
4044
4045 #[test]
4046 fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
4047 let mut held = task("diagnosed");
4052 held.start("run-1".to_owned());
4053 held.fail("gate red", 1);
4054 held.diagnostic = Some("cargo test failed: ...".to_owned());
4055 assert_eq!(held.status, TaskStatus::Held);
4056
4057 held.release();
4058 assert!(held.diagnostic.is_none());
4059
4060 held.diagnostic = Some("cargo test failed: ...".to_owned());
4061 held.succeed();
4062 assert!(held.diagnostic.is_none());
4063 }
4064
4065 #[test]
4066 fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
4067 let mut t = task("retried");
4068 t.start("run-1".to_owned());
4069 t.diagnostic = Some("stale evidence from a previous hold".to_owned());
4070 t.fail("unrelated config error", 5);
4071 assert_eq!(t.status, TaskStatus::Failed);
4072 assert!(
4073 t.diagnostic.is_none(),
4074 "fail() must not let an old diagnostic outlive the run that produced it"
4075 );
4076 }
4077
4078 #[test]
4079 fn a_claim_is_exclusive_and_releases_on_drop() {
4080 let (_dir, q) = queue();
4081 let mut t = task("contended");
4082 q.put(&mut t).unwrap();
4083
4084 let held = q.claim(&t.id).unwrap();
4085 assert!(
4086 q.claim(&t.id).is_err(),
4087 "two daemons must not drive one task into two runs"
4088 );
4089 drop(held);
4090 assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
4091 }
4092
4093 #[test]
4094 fn a_round_trip_survives_disk() {
4095 let (_dir, q) = queue();
4096 let mut t = Task::new(
4097 "titled".to_owned(),
4098 "body".to_owned(),
4099 PathBuf::from("/repo"),
4100 Source::Agent {
4101 run: "20260101-000000-beef".to_owned(),
4102 node: "implement".to_owned(),
4103 },
4104 );
4105 t.priority = 3;
4106 q.put(&mut t).unwrap();
4107
4108 let back = q.get(&t.id).unwrap();
4109 assert_eq!(back.id, t.id);
4110 assert_eq!(back.priority, 3);
4111 assert_eq!(back.source.label(), "implement@beef");
4112 assert_eq!(q.get(t.short()).unwrap().id, t.id);
4114 }
4115
4116 #[test]
4117 fn an_unreadable_task_does_not_take_the_queue_down() {
4118 let (_dir, q) = queue();
4119 let mut t = task("fine");
4120 q.put(&mut t).unwrap();
4121 std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
4122
4123 let listed = q.list();
4124 assert_eq!(listed.len(), 1, "the readable task still lists");
4125 assert_eq!(listed[0].id, t.id);
4126 }
4127
4128 #[test]
4129 fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
4130 let (_dir, q) = queue();
4131 let path = q.path_of("20260101-000000-aaaa");
4132 std::fs::create_dir_all(q.root()).unwrap();
4133 std::fs::write(
4134 &path,
4135 serde_json::json!({
4136 "schema": SCHEMA,
4137 "id": "20260101-000000-aaaa",
4138 "title": "from before solo existed",
4139 "instruction": "from before solo existed",
4140 "repo": ".",
4141 "source": { "kind": "human" },
4142 "status": "queued",
4143 "created_at": Timestamp::now().to_string(),
4144 "updated_at": Timestamp::now().to_string(),
4145 })
4146 .to_string(),
4147 )
4148 .unwrap();
4149
4150 let task = q.get("20260101-000000-aaaa").expect("must still read");
4151 assert!(!task.solo, "a queue file with no `solo` field means false");
4152 }
4153
4154 #[test]
4155 fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
4156 let (_dir, q) = queue();
4157 let path = q.path_of("20260101-000000-bbbb");
4158 std::fs::create_dir_all(q.root()).unwrap();
4159 std::fs::write(
4160 &path,
4161 serde_json::json!({
4162 "schema": SCHEMA,
4163 "id": "20260101-000000-bbbb",
4164 "title": "from before urgent existed",
4165 "instruction": "from before urgent existed",
4166 "repo": ".",
4167 "source": { "kind": "human" },
4168 "status": "queued",
4169 "created_at": Timestamp::now().to_string(),
4170 "updated_at": Timestamp::now().to_string(),
4171 })
4172 .to_string(),
4173 )
4174 .unwrap();
4175
4176 let task = q.get("20260101-000000-bbbb").expect("must still read");
4177 assert!(
4178 !task.urgent,
4179 "a queue file with no `urgent` field means false, same as `solo`"
4180 );
4181 }
4182
4183 #[test]
4184 fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
4185 let (_dir, q) = queue();
4186 let mut t = task("from the future");
4187 q.put(&mut t).unwrap();
4188 let path = q.path_of(&t.id);
4189 let body = std::fs::read_to_string(&path)
4190 .unwrap()
4191 .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
4192 std::fs::write(&path, body).unwrap();
4193
4194 let err = q.get(&t.id).unwrap_err().to_string();
4195 assert!(err.contains("schema 99"), "{err}");
4196 }
4197
4198 #[test]
4199 fn revision_moves_when_the_queue_changes() {
4200 let (_dir, q) = queue();
4201 assert_eq!(q.revision(), 0, "an empty queue has no revision");
4202 let mut t = task("first");
4203 q.put(&mut t).unwrap();
4204 assert!(q.revision() > 0, "a written task moves the revision");
4205 }
4206
4207 #[test]
4208 fn revision_moves_when_deleting_an_older_task() {
4209 let (dir, q) = queue();
4210 let questions = Questions::at(dir.path().join("questions"));
4211 let mut t1 = task("older");
4212 q.put(&mut t1).unwrap();
4213 std::thread::sleep(std::time::Duration::from_millis(10));
4215 let mut t2 = task("newer");
4216 q.put(&mut t2).unwrap();
4217
4218 let rev_before = q.revision();
4219 q.remove(&t1.id, false, &questions).unwrap();
4220 let rev_after = q.revision();
4221
4222 assert_ne!(
4223 rev_before, rev_after,
4224 "deleting an older task must change the revision so other clients see the deletion"
4225 );
4226 }
4227
4228 #[test]
4229 fn removing_a_task_takes_it_out_of_the_listing() {
4230 let (dir, q) = queue();
4231 let questions = Questions::at(dir.path().join("questions"));
4232 let mut t = task("delete me");
4233 q.put(&mut t).unwrap();
4234 let removed = q.remove(t.short(), false, &questions).unwrap();
4235 assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
4236 assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
4237 assert!(q.list().is_empty());
4238 assert!(
4239 q.remove(&t.id, false, &questions).is_err(),
4240 "removing twice is an error"
4241 );
4242 }
4243
4244 #[test]
4245 fn removing_a_task_takes_its_stale_lock_with_it() {
4246 let (dir, q) = queue();
4247 let questions = Questions::at(dir.path().join("questions"));
4248 let mut t = task("interrupted");
4249 q.put(&mut t).unwrap();
4250
4251 let claim = q.claim(&t.id).unwrap();
4254 std::mem::forget(claim);
4255 assert!(
4256 q.claim(&t.id).is_err(),
4257 "the orphaned lock is what makes the task look claimed"
4258 );
4259
4260 let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
4262 assert!(err.contains("live daemon"), "{err}");
4263 assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
4264
4265 q.remove(&t.id, false, &questions).unwrap();
4267 assert!(q.list().is_empty());
4268 let mut again = task("interrupted");
4269 again.id = t.id.clone();
4270 q.put(&mut again).unwrap();
4271 assert!(
4272 q.claim(&t.id).is_ok(),
4273 "a task that comes back must be claimable, which a left-behind lock would prevent"
4274 );
4275 }
4276
4277 fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
4278 crate::notices::Notices::at(dir.join("notifications")).list()
4279 }
4280
4281 #[test]
4282 fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
4283 let (dir, q) = queue();
4284 let questions = Questions::at(dir.path().join("questions"));
4285 let mut dep = task("dependency");
4286 q.put(&mut dep).unwrap();
4287 let mut blocked = task("waiting");
4288 blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
4289 q.put(&mut blocked).unwrap();
4290
4291 let removed = q.remove(&dep.id, false, &questions).unwrap();
4292 assert_eq!(removed.released, [blocked.id.clone()]);
4293 assert!(removed.still_blocked.is_empty());
4294
4295 let after = q.get(&blocked.id).unwrap();
4296 assert_eq!(after.status, TaskStatus::Queued);
4297 assert!(after.blocked_by.is_empty());
4298 assert!(after.block_reason.is_none());
4299 let notes = notices_of(dir.path());
4300 assert_eq!(notes.len(), 1, "{notes:?}");
4301 assert_eq!(notes[0].severity, crate::notices::Severity::Info);
4302 assert_eq!(
4303 notes[0].link,
4304 Some(crate::notices::Link::Task {
4305 id: blocked.id.clone(),
4306 })
4307 );
4308 }
4309
4310 #[test]
4311 fn removing_one_of_two_dependencies_keeps_the_other() {
4312 let (dir, q) = queue();
4313 let questions = Questions::at(dir.path().join("questions"));
4314 let mut dep = task("dependency");
4315 q.put(&mut dep).unwrap();
4316 let mut other = task("other");
4317 q.put(&mut other).unwrap();
4318 let mut blocked = task("waiting");
4319 blocked.block(
4320 vec![dep.id.clone(), other.id.clone()],
4321 Some("waits on both".to_owned()),
4322 );
4323 q.put(&mut blocked).unwrap();
4324
4325 let removed = q.remove(&dep.id, false, &questions).unwrap();
4326 assert!(removed.released.is_empty());
4327 assert_eq!(removed.still_blocked, [blocked.id.clone()]);
4328
4329 let after = q.get(&blocked.id).unwrap();
4330 assert_eq!(after.status, TaskStatus::Blocked);
4331 assert_eq!(after.blocked_by, [other.id.clone()]);
4332 assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
4335 }
4336
4337 #[test]
4338 fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
4339 let (dir, q) = queue();
4340 let questions = Questions::at(dir.path().join("questions"));
4341 let mut dep = task("dependency");
4342 q.put(&mut dep).unwrap();
4343 let mut blocked = task("waiting");
4344 blocked.block(vec![dep.id.clone()], None);
4345 q.put(&mut blocked).unwrap();
4346
4347 let claim = q.claim(&blocked.id).unwrap();
4349 let removed = q.remove(&dep.id, false, &questions).unwrap();
4350 assert!(removed.released.is_empty());
4351 drop(claim);
4352
4353 let mut task = q.get(&blocked.id).unwrap();
4354 assert_eq!(task.status, TaskStatus::Blocked);
4355 assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
4356 assert_eq!(
4357 q.apply_deleted_blockers(&mut task, &questions),
4358 [dep.id.clone()]
4359 );
4360 assert_eq!(task.status, TaskStatus::Queued);
4361 }
4362
4363 #[test]
4364 fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
4365 let (dir, q) = queue();
4366 let questions = Questions::at(dir.path().join("questions"));
4367 let mut dep = task("dependency");
4368 dep.id = "20261013-193611-ca5b".into();
4369 q.put(&mut dep).unwrap();
4370 let mut waiting = task("waiting");
4371 waiting.block(vec!["ca5b".into()], None);
4372 q.put(&mut waiting).unwrap();
4373 q.remove(&dep.id, false, &questions).unwrap();
4374 assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
4375 assert_eq!(
4376 dependency(&q, &questions, "ca5b"),
4377 Dependency::Deleted(dep.id)
4378 );
4379 }
4380
4381 #[test]
4382 fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
4383 let (dir, queue) = queue();
4384 let questions = Questions::at(dir.path().join("questions"));
4385 let mut dep = task("dependency");
4386 dep.id = "20261013-193611-ca5b".into();
4387 queue.put(&mut dep).unwrap();
4388 std::fs::create_dir_all(questions.root()).unwrap();
4389 let question_id = "20261014-193611-ca5b";
4390 std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
4391 assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
4392 assert_eq!(
4393 dependency(&queue, &questions, "ca5b"),
4394 Dependency::Ambiguous
4395 );
4396 assert_eq!(
4397 dependency(&queue, &questions, &dep.id),
4398 Dependency::Task(dep.id)
4399 );
4400 assert_eq!(
4401 dependency(&queue, &questions, question_id),
4402 Dependency::Question(question_id.into())
4403 );
4404 assert_eq!(
4405 dependency(&queue, &questions, "missing"),
4406 Dependency::Missing
4407 );
4408 }
4409
4410 #[test]
4411 fn dependency_resolution_directory_errors_are_unknown_not_missing() {
4412 let dir = tempfile::tempdir().unwrap();
4413 let q = Queue::at(dir.path().join("not-a-directory"));
4414 std::fs::write(q.root(), "file").unwrap();
4415 let questions = Questions::at(dir.path().join("questions"));
4416 assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
4417 assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
4418 }
4419
4420 #[test]
4421 fn a_dependency_without_a_tombstone_is_still_missing() {
4422 let (dir, q) = queue();
4423 let questions = Questions::at(dir.path().join("questions"));
4424 let mut dep = task("dependency");
4425 q.put(&mut dep).unwrap();
4426 std::fs::remove_file(q.path_of(&dep.id)).unwrap();
4427 assert_eq!(
4428 missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
4429 [dep.id.clone()]
4430 );
4431 assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
4432 }
4433
4434 #[test]
4435 fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
4436 let mut t = task("held");
4437 t.hold_machine(Some("because".to_owned()));
4438 t.blocked_by = vec!["gone".to_owned()];
4439 assert!(!t.dependency_deleted("gone"));
4440 assert_eq!(t.status, TaskStatus::Held);
4441 }
4442
4443 #[test]
4444 fn a_failed_record_removal_takes_the_tombstone_back() {
4445 let (_dir, q) = queue();
4446 let mut t = task("stays");
4447 q.put(&mut t).unwrap();
4448 q.write_tombstone(&t.id).unwrap();
4449 let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
4450 assert!(err.is_err());
4451 assert!(!q.was_deleted(&t.id));
4454 }
4455
4456 fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
4457 let p = dir.join(name);
4458 std::fs::write(&p, body).unwrap();
4459 p
4460 }
4461
4462 #[test]
4463 fn an_attachment_copy_survives_deleting_its_source() {
4464 let (dir, q) = queue();
4465 let src = source_file(dir.path(), "shot.png", "pixels");
4466 let mut t = task("with a picture");
4467 let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
4468 q.put(&mut t).unwrap();
4469 std::fs::remove_file(&src).unwrap();
4470 assert_eq!(names, ["shot.png"]);
4471 let loaded = q.get(&t.id).unwrap();
4472 let paths = q.attachment_paths(&loaded);
4473 assert_eq!(paths.len(), 1);
4474 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
4475 }
4476
4477 #[test]
4478 fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
4479 let (dir, q) = queue();
4480 let mut t = task("bad names");
4481 for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
4482 let src = source_file(dir.path(), name, "x");
4483 assert!(
4484 q.attach(&mut t, &[src]).is_err(),
4485 "`{name}` must be refused"
4486 );
4487 }
4488 assert!(!crate::ask::valid_asset_name("C:foo.png"));
4492 #[cfg(not(windows))]
4493 {
4494 let src = source_file(dir.path(), "C:foo.png", "x");
4495 assert!(q.attach(&mut t, &[src]).is_err());
4496 }
4497 let long = format!("{}.png", "a".repeat(70));
4498 let src = source_file(dir.path(), &long, "x");
4499 assert!(q.attach(&mut t, &[src]).is_err());
4500 assert!(t.attachments.is_empty());
4501 assert!(!q.attachments_dir(&t.id).exists());
4502 }
4503
4504 #[test]
4505 fn a_taken_attachment_name_is_numbered_not_overwritten() {
4506 let (dir, q) = queue();
4507 let a = source_file(dir.path(), "shot.png", "one");
4508 let sub = dir.path().join("other");
4509 std::fs::create_dir_all(&sub).unwrap();
4510 let b = source_file(&sub, "shot.png", "two");
4511 let mut t = task("collision");
4512 q.attach(&mut t, &[a]).unwrap();
4513 q.attach(&mut t, &[b]).unwrap();
4514 assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
4515 let paths = q.attachment_paths(&t);
4516 assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
4517 assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
4518 }
4519
4520 #[test]
4521 fn a_renumbered_name_stays_inside_the_length_bound() {
4522 let (dir, q) = queue();
4523 let name = format!("{}.png", "a".repeat(60));
4524 assert_eq!(name.len(), 64);
4525 let a = source_file(dir.path(), &name, "one");
4526 let sub = dir.path().join("other");
4527 std::fs::create_dir_all(&sub).unwrap();
4528 let b = source_file(&sub, &name, "two");
4529 let mut t = task("long");
4530 q.attach(&mut t, &[a, b]).unwrap();
4531 assert_eq!(t.attachments.len(), 2);
4532 assert!(
4533 t.attachments
4534 .iter()
4535 .all(|n| crate::ask::valid_asset_name(n))
4536 );
4537 assert!(t.attachments[1].ends_with("-2.png"));
4538 }
4539
4540 #[test]
4541 fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
4542 let (dir, q) = queue();
4543 let good = source_file(dir.path(), "good.png", "ok");
4544 let mut t = task("partial");
4545 q.attach(&mut t, &[good]).unwrap();
4546 let more = source_file(dir.path(), "more.png", "ok");
4547 let missing = dir.path().join("missing.png");
4548 assert!(q.attach(&mut t, &[more, missing]).is_err());
4549 assert_eq!(t.attachments, ["good.png"]);
4550 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4551 .unwrap()
4552 .flatten()
4553 .collect();
4554 assert_eq!(on_disk.len(), 1);
4555 }
4556
4557 fn block_put(q: &Queue, t: &Task) -> PathBuf {
4559 let tmp = q.path_of(&t.id).with_extension("json.tmp");
4560 std::fs::create_dir_all(&tmp).unwrap();
4561 tmp
4562 }
4563
4564 #[test]
4565 fn a_failed_put_leaves_no_new_attachment_directory() {
4566 let (dir, q) = queue();
4567 let mut t = task("fresh");
4568 let tmp = block_put(&q, &t);
4569 let src = source_file(dir.path(), "shot.png", "x");
4570 assert!(q.attach_and_put(&mut t, &[src]).is_err());
4571 assert!(t.attachments.is_empty());
4572 assert!(!q.attachments_dir(&t.id).exists());
4573 assert!(!q.path_of(&t.id).exists());
4574 std::fs::remove_dir(tmp).unwrap();
4575 }
4576
4577 #[test]
4578 fn a_failed_put_removes_only_the_copy_it_just_made() {
4579 let (dir, q) = queue();
4580 let mut t = task("edited");
4581 let first = source_file(dir.path(), "first.png", "1");
4582 q.attach_and_put(&mut t, &[first]).unwrap();
4583 block_put(&q, &t);
4584 let second = source_file(dir.path(), "second.png", "2");
4585 assert!(q.attach_and_put(&mut t, &[second]).is_err());
4586 assert_eq!(t.attachments, ["first.png"]);
4587 let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4588 .unwrap()
4589 .flatten()
4590 .map(|e| e.file_name().to_string_lossy().into_owned())
4591 .collect();
4592 assert_eq!(on_disk, ["first.png"]);
4593 assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
4594 }
4595
4596 #[test]
4597 fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
4598 let (dir, q) = queue();
4599 let questions = Questions::at(dir.path().join("questions"));
4600 let gone = task("gone");
4601 let mut other = task("other");
4602 let mut live = task("live");
4603 q.put(&mut other).unwrap();
4604 q.put(&mut live).unwrap();
4605 let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
4608 std::fs::create_dir_all(&orphan).unwrap();
4609 std::fs::write(orphan.join("shot.png"), "x").unwrap();
4610 let busy = q.root.join(format!("{}.attachments.removing", live.id));
4613 std::fs::create_dir_all(&busy).unwrap();
4614
4615 q.remove(&other.id, false, &questions).unwrap();
4616 assert!(!orphan.exists(), "an orphan is swept");
4617 assert!(busy.exists(), "a removal in progress is left alone");
4618 }
4619
4620 #[test]
4621 fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
4622 let (dir, q) = queue();
4623 let questions = Questions::at(dir.path().join("questions"));
4624 let mut t = task("stuck");
4625 let src = source_file(dir.path(), "shot.png", "x");
4626 q.attach_and_put(&mut t, &[src]).unwrap();
4627 let aside = q.root.join(format!("{}.attachments.removing", t.id));
4630 std::fs::create_dir_all(&aside).unwrap();
4631 std::fs::write(aside.join("old.png"), "o").unwrap();
4632 assert!(q.remove(&t.id, false, &questions).is_err());
4633 assert!(q.path_of(&t.id).exists());
4634 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
4635 }
4636
4637 #[test]
4638 fn a_failed_record_removal_puts_the_attachments_back() {
4639 let (dir, q) = queue();
4640 let mut t = task("rollback");
4641 let src = source_file(dir.path(), "shot.png", "x");
4642 q.attach_and_put(&mut t, &[src]).unwrap();
4643 let err = q
4644 .remove_record_with_attachments(&t.id, |_| {
4645 Err(std::io::Error::other("injected failure"))
4646 })
4647 .unwrap_err();
4648 assert!(format!("{err:#}").contains("injected failure"));
4649 assert!(q.path_of(&t.id).exists());
4650 assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
4651 assert!(
4652 !q.root
4653 .join(format!("{}.attachments.removing", t.id))
4654 .exists()
4655 );
4656 }
4657
4658 #[test]
4659 fn editing_a_task_keeps_its_attachments() {
4660 let (dir, q) = queue();
4661 let src = source_file(dir.path(), "shot.png", "x");
4662 let mut t = task("editable");
4663 q.attach(&mut t, &[src]).unwrap();
4664 t.edit("new".to_owned(), "new text".to_owned()).unwrap();
4665 q.put(&mut t).unwrap();
4666 assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
4667 }
4668
4669 #[test]
4670 fn removing_a_task_deletes_its_attachments() {
4671 let (dir, q) = queue();
4672 let questions = Questions::at(dir.path().join("questions"));
4673 let src = source_file(dir.path(), "shot.png", "x");
4674 let mut t = task("doomed");
4675 q.attach(&mut t, &[src]).unwrap();
4676 q.put(&mut t).unwrap();
4677 assert!(q.attachments_dir(&t.id).is_dir());
4678 q.remove(&t.id, false, &questions).unwrap();
4679 assert!(!q.attachments_dir(&t.id).exists());
4680 assert!(q.list().is_empty());
4681 }
4682
4683 #[test]
4684 fn a_task_written_before_attachments_still_reads() {
4685 let (_dir, q) = queue();
4686 let mut t = task("old");
4687 q.put(&mut t).unwrap();
4688 let path = q.path_of(&t.id);
4689 let mut v: serde_json::Value =
4690 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
4691 v.as_object_mut().unwrap().remove("attachments");
4692 std::fs::write(&path, v.to_string()).unwrap();
4693 assert!(q.get(&t.id).unwrap().attachments.is_empty());
4694 }
4695
4696 #[test]
4697 fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
4698 let q = Queue::at(PathBuf::from("relative-queue"));
4699 let mut t = task("rel");
4700 t.attachments.push("shot.png".to_owned());
4701 let paths = q.attachment_paths(&t);
4702 assert!(paths[0].is_absolute(), "{}", paths[0].display());
4703 assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
4704 }
4705
4706 #[test]
4707 fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
4708 let dir = tempfile::tempdir().unwrap();
4709 let q = Queue::at(dir.path().join("queue"));
4710 let mut t = task("resume");
4711 t.start("A".into());
4712 t.record_pass("A", 0, 1, None);
4713 t.start("B".into());
4714 q.put(&mut t).unwrap();
4715 assert_eq!(t.runs, vec!["A", "B"]);
4716 t.start("A".into());
4717 t.record_pass("A", 1, 3, None);
4718 q.put(&mut t).unwrap();
4719 let saved = q.get(&t.id).unwrap();
4720 assert_eq!(saved.runs, vec!["B", "A"]);
4721 assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
4722 assert_eq!(saved.attempts, 3);
4723 assert_eq!(saved.passes.len(), 2);
4724 assert_eq!(saved.passes[0].run, "A");
4725 assert_eq!(saved.passes[1].run, "A");
4726 }
4727
4728 #[test]
4729 fn repeated_start_records_one_run_and_separate_passes() {
4730 let mut t = task("resume");
4731 t.start("same-run".into());
4732 t.record_pass("same-run", 0, 1, None);
4733 t.start("same-run".into());
4734 t.record_pass("same-run", 1, 2, None);
4735 assert_eq!(t.runs, vec!["same-run"]);
4736 assert_eq!(t.attempts, 2);
4737 assert_eq!(t.passes.len(), 2);
4738 }
4739
4740 #[test]
4741 fn link_run_adds_a_run_once_and_touches_nothing_else() {
4742 let dir = tempfile::tempdir().unwrap();
4743 let queue = Queue::at(dir.path().join("queue"));
4744 let mut t = Task::new(
4745 "t".to_owned(),
4746 "do it".to_owned(),
4747 PathBuf::from("."),
4748 Source::Human,
4749 );
4750 queue.put(&mut t).unwrap();
4751 let before = queue.get(&t.id).unwrap();
4752
4753 let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
4754 assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
4755 assert_eq!(linked.status, before.status);
4756 assert_eq!(linked.attempts, before.attempts);
4757 assert_eq!(linked.interrupt, before.interrupt);
4758
4759 let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
4760 assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
4761 assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
4762 assert!(queue.link_run("no-such-task", "r").is_err());
4763 }
4764
4765 #[test]
4766 fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
4767 let dir = tempfile::tempdir().unwrap();
4768 let queue = Queue::at(dir.path().join("queue"));
4769 let mut t = Task::new(
4770 "t".to_owned(),
4771 "do it".to_owned(),
4772 PathBuf::from("."),
4773 Source::Human,
4774 );
4775 queue.put(&mut t).unwrap();
4776 let mut snapshot = queue.get(&t.id).unwrap();
4778 queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
4779
4780 snapshot.start("20260930-000000-aaaa".to_owned());
4781 queue.put(&mut snapshot).unwrap();
4782
4783 let stored = queue.get(&t.id).unwrap();
4784 assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
4785 assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
4786 assert_eq!(stored.attempts, 1);
4787 }
4788
4789 #[test]
4790 fn concurrent_links_and_daemon_saves_lose_nothing() {
4791 let dir = tempfile::tempdir().unwrap();
4792 let queue = Queue::at(dir.path().join("queue"));
4793 let mut t = Task::new(
4794 "t".to_owned(),
4795 "do it".to_owned(),
4796 PathBuf::from("."),
4797 Source::Human,
4798 );
4799 queue.put(&mut t).unwrap();
4800 let id = t.id.clone();
4801
4802 let linkers: Vec<_> = (0..4)
4803 .map(|n| {
4804 let (queue, id) = (queue.clone(), id.clone());
4805 std::thread::spawn(move || {
4806 for k in 0..10 {
4807 queue
4808 .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
4809 .unwrap();
4810 }
4811 })
4812 })
4813 .collect();
4814 let mut mine = queue.get(&id).unwrap();
4817 for k in 0..10 {
4818 mine.start(format!("20260930-000009-d{k:03}"));
4819 queue.put(&mut mine).unwrap();
4820 }
4821 for l in linkers {
4822 l.join().unwrap();
4823 }
4824
4825 let stored = queue.get(&id).unwrap();
4826 assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
4827 assert_eq!(
4828 stored.attempts, 10,
4829 "linking never rewinds the daemon's work"
4830 );
4831 }
4832
4833 fn age_lock(path: &Path) {
4834 let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
4835 f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
4836 .unwrap();
4837 }
4838
4839 #[test]
4840 fn concurrent_stale_takeover_yields_one_holder() {
4841 use std::sync::atomic::{AtomicUsize, Ordering};
4842 use std::sync::{Arc, Barrier};
4843 for _ in 0..5 {
4844 let dir = tempfile::tempdir().unwrap();
4845 let q = Queue::at(dir.path().to_path_buf());
4846 let lock = dir.path().join("t.write-lock");
4847 std::fs::write(&lock, "dead-0000").unwrap();
4848 age_lock(&lock);
4849 let n = 6;
4850 let barrier = Arc::new(Barrier::new(n));
4851 let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
4852 let handles: Vec<_> = (0..n)
4853 .map(|_| {
4854 let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
4855 std::thread::spawn(move || {
4856 b.wait();
4857 let g = q.lock_task("t").unwrap();
4858 let held = now.fetch_add(1, Ordering::SeqCst) + 1;
4859 max.fetch_max(held, Ordering::SeqCst);
4860 std::thread::sleep(std::time::Duration::from_millis(20));
4861 now.fetch_sub(1, Ordering::SeqCst);
4862 drop(g);
4863 })
4864 })
4865 .collect();
4866 for h in handles {
4867 h.join().unwrap();
4868 }
4869 assert_eq!(max.load(Ordering::SeqCst), 1);
4870 assert!(!lock.exists());
4871 }
4872 }
4873
4874 #[test]
4875 fn dropping_a_stolen_lock_leaves_the_new_holders_lock() {
4876 let dir = tempfile::tempdir().unwrap();
4877 let q = Queue::at(dir.path().to_path_buf());
4878 let lock = dir.path().join("t.write-lock");
4879 let a = q.lock_task("t").unwrap();
4880 age_lock(&lock);
4881 let b = q.lock_task("t").unwrap();
4882 assert_ne!(a.token, b.token);
4883 drop(a);
4884 assert_eq!(std::fs::read_to_string(&lock).unwrap(), b.token);
4885 drop(b);
4886 assert!(!lock.exists());
4887 }
4888
4889 #[test]
4890 fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
4891 let dir = tempfile::tempdir().unwrap();
4892 let lock = dir.path().join("t.write-lock");
4893 std::fs::write(&lock, "new-token").unwrap();
4894 assert!(!break_stale(&lock, "old-token"));
4895 assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
4896 assert!(!break_stale(&lock, "new-token"));
4898 assert!(lock.exists());
4899 age_lock(&lock);
4900 assert!(break_stale(&lock, "new-token"));
4901 assert!(!lock.exists());
4902 }
4903
4904 #[test]
4905 fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
4906 let dir = tempfile::tempdir().unwrap();
4907 let lock = dir.path().join("t.write-lock");
4908 std::fs::write(&lock, "dead-1").unwrap();
4909 age_lock(&lock);
4910 let marker = break_marker(&lock, "dead-1");
4911 std::fs::write(&marker, "crashed-remover").unwrap();
4912 assert!(!break_stale(&lock, "dead-1"));
4914 assert!(lock.exists());
4915 age_lock(&marker);
4917 assert!(break_stale(&lock, "dead-1"));
4918 assert!(!lock.exists());
4919 assert!(!marker.exists());
4920 }
4921
4922 fn chat_task() -> Task {
4923 Task::new(
4924 "t".to_owned(),
4925 "do t".to_owned(),
4926 PathBuf::from("."),
4927 Source::Agent {
4928 run: "20260902-000000-beef".to_owned(),
4929 node: CHAT_NODE.to_owned(),
4930 },
4931 )
4932 }
4933
4934 #[test]
4935 fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
4936 let mut t = chat_task();
4937 assert_eq!(t.chat_notice_due(), None, "queued");
4938 t.start("r1".to_owned());
4939 assert_eq!(t.chat_notice_due(), None, "running");
4940 t.fail("boom", 3);
4941 assert_eq!(t.status, TaskStatus::Failed);
4942 assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
4943 t.start("r2".to_owned());
4944 t.park("upgrade");
4945 assert_eq!(t.chat_notice_due(), None, "parked");
4946 t.fail("boom", 1);
4947 assert_eq!(t.status, TaskStatus::Held);
4948 let key = t
4949 .chat_notice_due()
4950 .expect("held on the last attempt is final");
4951 t.chat_notice = Some(ChatNotice {
4952 key,
4953 at: Timestamp::now(),
4954 outcome: ChatNoticeOutcome::Posted,
4955 });
4956 assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
4957 t.release();
4958 assert_eq!(t.chat_notice_due(), None, "requeued after a release");
4959 t.start("r3".to_owned());
4960 t.succeed();
4961 assert!(t.chat_notice_due().is_some(), "done is a new final state");
4962 }
4963
4964 #[test]
4965 fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
4966 for key in ["done", "done#0", "done#1", "done#12"] {
4967 let mut t = chat_task();
4968 t.start("r1".to_owned());
4969 t.succeed();
4970 t.chat_notice = Some(ChatNotice {
4971 key: key.to_owned(),
4972 at: Timestamp::now(),
4973 outcome: ChatNoticeOutcome::Posted,
4974 });
4975 t.requeue();
4976 t.start("r2".to_owned());
4977 t.succeed();
4978 assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
4979 assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
4980 }
4981 }
4982
4983 #[test]
4984 fn a_new_hold_after_a_release_is_a_new_cause() {
4985 let mut t = chat_task();
4986 t.hold_manual(Some("wait".to_owned()));
4987 let first = t.chat_notice_due().expect("held");
4988 t.chat_notice = Some(ChatNotice {
4989 key: first.clone(),
4990 at: Timestamp::now(),
4991 outcome: ChatNoticeOutcome::Posted,
4992 });
4993 t.hold_manual(Some("still waiting".to_owned()));
4994 assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
4995 t.release();
4996 std::thread::sleep(std::time::Duration::from_millis(5));
4997 t.hold_manual(None);
4998 let second = t.chat_notice_due().expect("a fresh hold");
4999 assert_ne!(first, second);
5000 }
5001
5002 #[test]
5003 fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
5004 let mut human = task("h");
5005 human.succeed();
5006 assert_eq!(human.chat_notice_due(), None);
5007 let mut inherited = task("i");
5009 inherited.origin_chat = Some("20260902-000000-beef".to_owned());
5010 inherited.succeed();
5011 assert_eq!(inherited.chat_notice_due(), None);
5012 let mut old = chat_task();
5013 old.schema = 12;
5014 old.status = TaskStatus::Done;
5015 assert_eq!(old.chat_notice_due(), None, "history is not replayed");
5016 let mut live = chat_task();
5017 live.schema = 12;
5018 live.start("r1".to_owned());
5019 live.succeed();
5020 assert!(live.chat_notice_due().is_some(), "started by this build");
5021 let mut held = chat_task();
5022 held.schema = 12;
5023 held.hold_machine(Some("config".to_owned()));
5024 assert!(held.chat_notice_due().is_some(), "held before any start");
5025 }
5026
5027 #[test]
5028 fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
5029 let tmp = tempfile::tempdir().expect("tempdir");
5030 let q = Queue::at(tmp.path().join("queue"));
5031 let mut t = chat_task();
5032 t.succeed();
5033 q.put(&mut t).expect("put");
5034 let mut stale = q.get(&t.id).expect("get");
5035 let key = t.chat_notice_due().expect("due");
5036 q.modify(&t.id, |x| {
5037 x.chat_notice = Some(ChatNotice {
5038 key: key.clone(),
5039 at: Timestamp::now(),
5040 outcome: ChatNoticeOutcome::Posted,
5041 });
5042 true
5043 })
5044 .expect("modify");
5045 q.put(&mut stale).expect("stale put");
5046 assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
5047 }
5048}