1use crate::backends::{BlockedByFetch, ExternalBackend, ExternalId, PullRequestCompletion};
8use crate::channels::notification::TaskNotification;
9use crate::channels::transport::Transport;
10use crate::store::{
11 blocker_failure_kind, blocker_is_satisfied, dep_label, BlockerSnapshot, DepSource,
12 NewDependency, ObservedPullRequest, TaskStore,
13};
14use std::collections::{HashMap, HashSet};
15
16const DEP_SYNC_TTL_SECS: i64 = 45;
17
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub struct DepRef {
20 pub repo: String,
21 pub number: i64,
22}
23
24#[derive(Debug, Clone)]
25pub struct SyncOutcome {
26 pub cache: bool,
27}
28
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct DepNotice {
31 pub repo: String,
32 pub issue_number: i64,
33 pub comment: String,
34 pub notify: bool,
35 pub summary: String,
36 pub task_id: i64,
37 pub title: String,
38}
39
40pub fn parse_body_dependencies(body: &str, default_repo: &str) -> Vec<DepRef> {
45 let mut out = Vec::new();
46 let mut seen = HashSet::new();
47 let mut in_fence = false;
48 for line in body.lines() {
49 let trimmed = line.trim();
50 if trimmed.starts_with("```") {
51 in_fence = !in_fence;
52 continue;
53 }
54 if in_fence {
55 continue;
56 }
57 let Some(rest) = dependency_line_rest(trimmed) else {
58 continue;
59 };
60 for dep in extract_refs(rest, default_repo) {
61 if seen.insert((dep.repo.clone(), dep.number)) {
62 out.push(dep);
63 }
64 }
65 }
66 out
67}
68
69fn dependency_line_rest(trimmed: &str) -> Option<&str> {
70 let lower = trimmed.to_ascii_lowercase();
71 let prefix = if lower.starts_with("depends on") {
72 "depends on".len()
73 } else if lower.starts_with("blocked by") {
74 "blocked by".len()
75 } else {
76 return None;
77 };
78 let rest = trimmed[prefix..]
79 .trim_start()
80 .trim_start_matches(':')
81 .trim();
82 if rest.is_empty() {
83 None
84 } else {
85 Some(rest)
86 }
87}
88
89pub fn parse_depends_on_token(token: &str, default_repo: &str) -> Option<DepRef> {
91 let trimmed = token.trim().trim_matches(',');
92 if trimmed.is_empty() {
93 return None;
94 }
95 if let Ok(number) = trimmed.parse::<i64>() {
96 if number > 0 && !default_repo.is_empty() {
97 return Some(DepRef {
98 repo: default_repo.to_string(),
99 number,
100 });
101 }
102 }
103 extract_refs(trimmed, default_repo).into_iter().next()
104}
105
106fn extract_refs(text: &str, default_repo: &str) -> Vec<DepRef> {
107 let mut out = Vec::new();
108 let bytes = text.as_bytes();
109 let mut index = 0;
110 while index < bytes.len() {
111 if bytes[index] != b'#' {
112 index += 1;
113 continue;
114 }
115 let digits: String = text[index + 1..]
116 .chars()
117 .take_while(|ch| ch.is_ascii_digit())
118 .collect();
119 if digits.is_empty() {
120 index += 1;
121 continue;
122 }
123 let number: i64 = match digits.parse() {
124 Ok(number) if number > 0 => number,
125 _ => {
126 index += 1 + digits.len();
127 continue;
128 }
129 };
130 let repo = repo_before(text, index).unwrap_or(default_repo).to_string();
131 if !repo.is_empty() {
132 out.push(DepRef { repo, number });
133 }
134 index += 1 + digits.len();
135 }
136 out
137}
138
139fn repo_before(text: &str, hash_at: usize) -> Option<&str> {
140 let before = &text[..hash_at];
141 let mut start = hash_at;
142 let bytes = before.as_bytes();
143 while start > 0 {
144 let ch = bytes[start - 1] as char;
145 if ch.is_ascii_alphanumeric() || matches!(ch, '_' | '.' | '-' | '/') {
146 start -= 1;
147 } else {
148 break;
149 }
150 }
151 let candidate = &before[start..];
152 let mut parts = candidate.split('/');
153 let owner = parts.next().unwrap_or("");
154 let name = parts.next().unwrap_or("");
155 if parts.next().is_some() || owner.is_empty() || name.is_empty() {
156 return None;
157 }
158 if !owner
159 .chars()
160 .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '.' | '-'))
161 || !name
162 .chars()
163 .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '.' | '-'))
164 {
165 return None;
166 }
167 Some(candidate)
168}
169
170pub async fn sync_issue_dependencies(
172 backend: &dyn ExternalBackend,
173 store: &TaskStore,
174 task_id: i64,
175 repo: &str,
176 issue_number: Option<i64>,
177 body: &str,
178) -> anyhow::Result<SyncOutcome> {
179 let Some(issue_number) = issue_number.filter(|number| *number > 0) else {
180 return Ok(SyncOutcome { cache: false });
181 };
182 let fetch = match backend.blocked_by(repo, issue_number).await {
183 Ok(fetch) => fetch,
184 Err(err) => BlockedByFetch::Failed(err.to_string()),
185 };
186 match fetch {
187 BlockedByFetch::Native(list) => {
188 for item in &list {
189 if item.state.is_empty() {
190 continue;
191 }
192 store
193 .note_blocker_issue_state(
194 &item.repo,
195 item.number,
196 &item.state,
197 item.state_reason.as_deref(),
198 )
199 .await?;
200 }
201 let rows = list
202 .iter()
203 .map(|item| NewDependency {
204 repo: item.repo.clone(),
205 number: item.number,
206 source: DepSource::Github,
207 })
208 .collect::<Vec<_>>();
209 store.replace_dependencies(task_id, &rows).await?;
210 store.link_known_task(task_id, repo, issue_number).await?;
211 Ok(SyncOutcome { cache: true })
212 }
213 BlockedByFetch::Unavailable => {
214 let parsed = parse_body_dependencies(body, repo);
215 if parsed.is_empty() {
216 store.link_known_task(task_id, repo, issue_number).await?;
217 return Ok(SyncOutcome { cache: true });
218 }
219 for dep in &parsed {
220 if store
221 .get_by_external_id(&dep.repo, &dep.number.to_string())
222 .await?
223 .is_none()
224 {
225 if let Ok(Some((state, reason))) =
226 backend.issue_close_state(&dep.repo, dep.number).await
227 {
228 store
229 .note_blocker_issue_state(
230 &dep.repo,
231 dep.number,
232 &state,
233 reason.as_deref(),
234 )
235 .await?;
236 }
237 }
238 }
239 let rows = parsed
240 .into_iter()
241 .map(|dep| NewDependency {
242 repo: dep.repo,
243 number: dep.number,
244 source: DepSource::Body,
245 })
246 .collect::<Vec<_>>();
247 store.replace_dependencies(task_id, &rows).await?;
248 store.link_known_task(task_id, repo, issue_number).await?;
249 Ok(SyncOutcome { cache: true })
250 }
251 BlockedByFetch::Failed(message) => {
252 tracing::warn!(
253 task_id,
254 repo,
255 issue_number,
256 error = %message,
257 "dependency read failed, keeping stored rows"
258 );
259 Ok(SyncOutcome { cache: false })
260 }
261 }
262}
263
264pub async fn apply_claim_gates(store: &TaskStore) -> anyhow::Result<()> {
267 let tasks = store.list_schedulable().await?;
268 for task in tasks {
269 store.block_claim_if_waiting(task.id).await?;
270 }
271 Ok(())
272}
273
274pub async fn reconcile_dependencies(
280 backend: &dyn ExternalBackend,
281 store: &TaskStore,
282 transport: Option<&Transport>,
283) -> anyhow::Result<()> {
284 let tasks = store.list_schedulable().await?;
285 let now = chrono::Utc::now().timestamp();
286 for task in &tasks {
287 let Some(number) = task
288 .external_id
289 .as_deref()
290 .and_then(|id| id.parse::<i64>().ok())
291 .filter(|number| *number > 0)
292 else {
293 continue;
294 };
295 let key = format!("dep_synced_at:{}", task.id);
296 if let Ok(Some(prev)) = store.kv_get(&key).await {
297 if let Ok(prev) = prev.parse::<i64>() {
298 if now.saturating_sub(prev) < DEP_SYNC_TTL_SECS {
299 continue;
300 }
301 }
302 }
303 match sync_issue_dependencies(
304 backend,
305 store,
306 task.id,
307 &task.repo,
308 Some(number),
309 &task.body,
310 )
311 .await
312 {
313 Ok(outcome) if outcome.cache => {
314 if let Err(err) = store.kv_set(&key, &now.to_string()).await {
315 tracing::warn!(task_id = task.id, err = %err, "failed to cache dependency sync time");
316 }
317 }
318 Ok(_) => {}
319 Err(err) => {
320 tracing::warn!(task_id = task.id, repo = %task.repo, err = %err, "dependency sync failed");
321 }
322 }
323 }
324 if let Err(err) = refresh_blocker_snapshots(backend, store).await {
325 tracing::warn!(err = %err, "blocker snapshot refresh failed");
326 }
327 apply_claim_gates(store).await?;
328 emit_dependency_notices(backend, store, transport).await?;
329 Ok(())
330}
331
332pub async fn refresh_blocker_snapshots(
338 backend: &dyn ExternalBackend,
339 store: &TaskStore,
340) -> anyhow::Result<()> {
341 let edges = store.list_dependency_edges().await?;
342 let mut seen = HashSet::new();
343 for edge in edges {
344 if !seen.insert((edge.depends_on_repo.clone(), edge.depends_on_number)) {
345 continue;
346 }
347 if let Err(err) = refresh_one_blocker(
348 backend,
349 store,
350 &edge.depends_on_repo,
351 edge.depends_on_number,
352 )
353 .await
354 {
355 tracing::warn!(
356 repo = %edge.depends_on_repo,
357 number = edge.depends_on_number,
358 err = %err,
359 "blocker snapshot refresh failed"
360 );
361 }
362 }
363 Ok(())
364}
365
366async fn refresh_one_blocker(
367 backend: &dyn ExternalBackend,
368 store: &TaskStore,
369 repo: &str,
370 number: i64,
371) -> anyhow::Result<()> {
372 let close = match backend.issue_close_state(repo, number).await {
373 Ok(state) => state,
374 Err(err) => {
375 tracing::warn!(repo, number, err = %err, "issue close state read failed");
376 None
377 }
378 };
379 let task = store.get_by_external_id(repo, &number.to_string()).await?;
380 let pr_number = task
381 .as_ref()
382 .and_then(|row| row.pr_number)
383 .filter(|n| *n > 0);
384 if close.is_none() && pr_number.is_none() {
385 return Ok(());
386 }
387 let (issue_state, reason) = match close {
388 Some((state, reason)) => (state, reason),
389 None => match store.blocker_snapshot(repo, number).await? {
390 Some(snap) => (snap.issue_state, snap.state_reason),
391 None => ("open".to_string(), None),
392 },
393 };
394 let observed = if let Some(pr) = pr_number {
395 let completion = match backend.pull_request_completion(repo, i64::from(pr)).await {
396 Ok(value) => value,
397 Err(err) => {
398 tracing::warn!(repo, number, pr, err = %err, "pull request completion read failed");
399 None
400 }
401 };
402 Some(observe_completion(completion))
403 } else {
404 None
405 };
406 let snap = BlockerSnapshot::observe(&issue_state, reason.as_deref(), observed.as_ref());
407 store.save_blocker_snapshot(repo, number, &snap).await
408}
409
410fn observe_completion(completion: Option<PullRequestCompletion>) -> ObservedPullRequest {
411 match completion {
412 Some(done) => ObservedPullRequest {
413 merged: done.merged,
414 base_is_default: !done.default_branch.is_empty()
415 && done.base_ref == done.default_branch,
416 approved: done.approved,
417 closed_unmerged: done.closed && !done.merged,
418 },
419 None => ObservedPullRequest {
420 merged: false,
421 base_is_default: false,
422 approved: false,
423 closed_unmerged: false,
424 },
425 }
426}
427
428pub async fn dispatch_is_waiting(store: &TaskStore, repo: &str, external_id: &str) -> bool {
430 let id = match store.resolve_id_by_external(repo, external_id).await {
431 Ok(Some(id)) => id,
432 Ok(None) => return false,
433 Err(err) => {
434 tracing::warn!(
435 task_id = external_id,
436 err = %err,
437 "dependency check failed, skipping dispatch"
438 );
439 return true;
440 }
441 };
442 match store.block_claim_if_waiting(id).await {
443 Ok(Some(blockers)) => {
444 tracing::info!(
445 task_id = external_id,
446 blockers = %blockers.join(", "),
447 "skipping dispatch, waiting on a dependency"
448 );
449 true
450 }
451 Ok(None) => false,
452 Err(err) => {
453 tracing::warn!(
454 task_id = external_id,
455 err = %err,
456 "dependency check failed, skipping dispatch"
457 );
458 true
459 }
460 }
461}
462
463pub async fn refuse_if_waiting(store: &TaskStore, task_id: i64) -> anyhow::Result<()> {
465 if let Some(blockers) = store.block_claim_if_waiting(task_id).await? {
466 anyhow::bail!("task {task_id} is waiting on {}", blockers.join(", "));
467 }
468 Ok(())
469}
470
471pub async fn emit_dependency_notices(
472 backend: &dyn ExternalBackend,
473 store: &TaskStore,
474 transport: Option<&Transport>,
475) -> anyhow::Result<Vec<DepNotice>> {
476 let mut notices = Vec::new();
477 notices.extend(cycle_notices(store).await?);
478 notices.extend(failure_notices(store).await?);
479 for notice in ¬ices {
480 if notice.issue_number > 0 && !notice.comment.is_empty() {
481 let id = ExternalId(notice.issue_number.to_string());
482 if let Err(err) = backend
483 .post_comment_in_repo(¬ice.repo, &id, ¬ice.comment)
484 .await
485 {
486 tracing::warn!(
487 repo = %notice.repo,
488 issue = notice.issue_number,
489 err = %err,
490 "failed to comment on dependency notice"
491 );
492 }
493 }
494 if notice.notify {
495 if let Some(transport) = transport {
496 transport.push_notification(TaskNotification {
497 task_id: if notice.issue_number > 0 {
498 notice.issue_number.to_string()
499 } else {
500 notice.task_id.to_string()
501 },
502 title: notice.title.clone(),
503 status: "failed".to_string(),
504 agent: "orch".to_string(),
505 duration_seconds: 0.0,
506 summary: notice.summary.clone(),
507 repo: Some(notice.repo.clone()),
508 notify_target: None,
509 pr_number: None,
510 });
511 }
512 }
513 }
514 Ok(notices)
515}
516
517async fn cycle_notices(store: &TaskStore) -> anyhow::Result<Vec<DepNotice>> {
518 let edges = store.list_dependency_edges().await?;
519 let mut snap_cache: HashMap<(String, i64), Option<BlockerSnapshot>> = HashMap::new();
520 let mut adj: HashMap<String, Vec<String>> = HashMap::new();
521 let mut issue_for_node: HashMap<String, (String, i64, i64, String)> = HashMap::new();
522
523 for edge in &edges {
524 let waiter = match edge
525 .task_external_id
526 .as_deref()
527 .and_then(|id| id.parse::<i64>().ok())
528 {
529 Some(number) if number > 0 => format!("{}#{}", edge.task_repo, number),
530 _ => format!("task:{}", edge.task_id),
531 };
532 if let Some(number) = edge
533 .task_external_id
534 .as_deref()
535 .and_then(|id| id.parse::<i64>().ok())
536 .filter(|number| *number > 0)
537 {
538 issue_for_node.insert(
539 waiter.clone(),
540 (
541 edge.task_repo.clone(),
542 number,
543 edge.task_id,
544 edge.task_title.clone(),
545 ),
546 );
547 }
548 let blocker_node = format!("{}#{}", edge.depends_on_repo, edge.depends_on_number);
549 let snap = cached_snapshot(
550 store,
551 &mut snap_cache,
552 &edge.depends_on_repo,
553 edge.depends_on_number,
554 )
555 .await?;
556 if blocker_is_satisfied(snap.as_ref()) {
557 continue;
558 }
559 adj.entry(waiter).or_default().push(blocker_node);
560 }
561
562 let mut notices = Vec::new();
563 for cycle in find_cycles(&adj) {
564 let signature = {
565 let mut nodes = cycle.clone();
566 nodes.sort();
567 nodes.join(",")
568 };
569 let mut named = cycle.clone();
570 if let Some(first) = named.first().cloned() {
571 named.push(first);
572 }
573 let path = named.join(" -> ");
574 let comment = format!(
575 "Dependency cycle: {path}. Orch will not run these tasks until the cycle is removed.{}",
576 crate::engine::orch_footer()
577 );
578 for node in &cycle {
579 let Some((repo, number, task_id, title)) = issue_for_node.get(node) else {
580 continue;
581 };
582 let key = format!("dep_cycle_comment:{repo}#{number}:{signature}");
583 if store.kv_get(&key).await?.is_some() {
584 continue;
585 }
586 store.kv_set(&key, "1").await?;
587 notices.push(DepNotice {
588 repo: repo.clone(),
589 issue_number: *number,
590 comment: comment.clone(),
591 notify: false,
592 summary: comment.clone(),
593 task_id: *task_id,
594 title: title.clone(),
595 });
596 }
597 }
598 Ok(notices)
599}
600
601async fn failure_notices(store: &TaskStore) -> anyhow::Result<Vec<DepNotice>> {
602 let edges = store.list_dependency_edges().await?;
603 let mut status_cache: HashMap<(String, i64), Option<String>> = HashMap::new();
604 let mut snap_cache: HashMap<(String, i64), Option<BlockerSnapshot>> = HashMap::new();
605 let mut notices = Vec::new();
606 for edge in edges {
607 let status = blocker_status(
608 store,
609 &mut status_cache,
610 &edge.depends_on_repo,
611 edge.depends_on_number,
612 )
613 .await?;
614 let snap = cached_snapshot(
615 store,
616 &mut snap_cache,
617 &edge.depends_on_repo,
618 edge.depends_on_number,
619 )
620 .await?;
621 let Some(kind) = blocker_failure_kind(status.as_deref(), snap.as_ref()) else {
622 continue;
623 };
624 let key = format!(
625 "dep_blocker_notice:{}:{}#{}:{kind}",
626 edge.task_id, edge.depends_on_repo, edge.depends_on_number
627 );
628 if store.kv_get(&key).await?.is_some() {
629 continue;
630 }
631 store.kv_set(&key, "1").await?;
632 let label = dep_label(
633 &edge.task_repo,
634 &edge.depends_on_repo,
635 edge.depends_on_number,
636 );
637 let (comment, summary) = match kind {
638 "not_planned" => {
639 let text = format!(
640 "Blocked by {label}, which was closed as not planned. This task stays waiting."
641 );
642 (format!("{text}{}", crate::engine::orch_footer()), text)
643 }
644 _ => {
645 let text = format!(
646 "Blocked by {label}, which is blocked and needs a person. This task stays waiting."
647 );
648 (format!("{text}{}", crate::engine::orch_footer()), text)
649 }
650 };
651 let issue_number = edge
652 .task_external_id
653 .as_deref()
654 .and_then(|id| id.parse::<i64>().ok())
655 .filter(|number| *number > 0)
656 .unwrap_or(0);
657 notices.push(DepNotice {
658 repo: edge.task_repo,
659 issue_number,
660 comment,
661 notify: true,
662 summary,
663 task_id: edge.task_id,
664 title: edge.task_title,
665 });
666 }
667 Ok(notices)
668}
669
670async fn cached_snapshot(
671 store: &TaskStore,
672 cache: &mut HashMap<(String, i64), Option<BlockerSnapshot>>,
673 repo: &str,
674 number: i64,
675) -> anyhow::Result<Option<BlockerSnapshot>> {
676 if let Some(found) = cache.get(&(repo.to_string(), number)) {
677 return Ok(found.clone());
678 }
679 let snap = store.blocker_snapshot(repo, number).await?;
680 cache.insert((repo.to_string(), number), snap.clone());
681 Ok(snap)
682}
683
684async fn blocker_status(
685 store: &TaskStore,
686 cache: &mut HashMap<(String, i64), Option<String>>,
687 repo: &str,
688 number: i64,
689) -> anyhow::Result<Option<String>> {
690 if let Some(status) = cache.get(&(repo.to_string(), number)) {
691 return Ok(status.clone());
692 }
693 let status = store
694 .get_by_external_id(repo, &number.to_string())
695 .await?
696 .map(|task| task.status.as_str().to_string());
697 cache.insert((repo.to_string(), number), status.clone());
698 Ok(status)
699}
700
701fn find_cycles(adj: &HashMap<String, Vec<String>>) -> Vec<Vec<String>> {
702 let mut nodes: Vec<String> = adj.keys().cloned().collect();
703 for targets in adj.values() {
704 for target in targets {
705 if !nodes.iter().any(|node| node == target) {
706 nodes.push(target.clone());
707 }
708 }
709 }
710 nodes.sort();
711 let mut color: HashMap<String, u8> = HashMap::new();
712 let mut stack = Vec::new();
713 let mut cycles = Vec::new();
714 let mut seen = HashSet::new();
715 for node in nodes {
716 if color.get(&node).copied().unwrap_or(0) == 0 {
717 visit(&node, adj, &mut color, &mut stack, &mut cycles, &mut seen);
718 }
719 }
720 cycles
721}
722
723fn visit(
724 node: &str,
725 adj: &HashMap<String, Vec<String>>,
726 color: &mut HashMap<String, u8>,
727 stack: &mut Vec<String>,
728 cycles: &mut Vec<Vec<String>>,
729 seen: &mut HashSet<String>,
730) {
731 color.insert(node.to_string(), 1);
732 stack.push(node.to_string());
733 if let Some(nexts) = adj.get(node) {
734 for next in nexts {
735 match color.get(next).copied().unwrap_or(0) {
736 1 => {
737 if let Some(start) = stack.iter().position(|item| item == next) {
738 let mut cycle = stack[start..].to_vec();
739 if let Some(min_at) = cycle
740 .iter()
741 .enumerate()
742 .min_by(|left, right| left.1.cmp(right.1))
743 .map(|(index, _)| index)
744 {
745 cycle.rotate_left(min_at);
746 }
747 let signature = {
748 let mut sorted = cycle.clone();
749 sorted.sort();
750 sorted.join(",")
751 };
752 if seen.insert(signature) {
753 cycles.push(cycle);
754 }
755 }
756 }
757 0 => visit(next, adj, color, stack, cycles, seen),
758 _ => {}
759 }
760 }
761 }
762 stack.pop();
763 color.insert(node.to_string(), 2);
764}
765
766#[cfg(test)]
767mod tests {
768 use super::*;
769 use crate::backends::{BlockedByIssue, ExternalTask, Status};
770 use crate::store::{NewTask, TaskStatus};
771 use async_trait::async_trait;
772 use std::sync::{Arc, Mutex};
773
774 #[test]
775 fn body_parser_reads_the_three_forms_and_skips_fences() {
776 let body = "\
777Depends on #12
778Blocked by #12, #13
779Depends on owner/repo#12
780```
781Depends on #99
782```
783See #4 in the notes
784";
785 let deps = parse_body_dependencies(body, "acme/app");
786 assert_eq!(
787 deps,
788 vec![
789 DepRef {
790 repo: "acme/app".into(),
791 number: 12
792 },
793 DepRef {
794 repo: "acme/app".into(),
795 number: 13
796 },
797 DepRef {
798 repo: "owner/repo".into(),
799 number: 12
800 },
801 ]
802 );
803 }
804
805 #[derive(Clone)]
806 struct ScriptedReviewComment {
807 login: String,
808 body: String,
809 created_at: String,
810 }
811
812 struct ScriptedBackend {
813 blocked: Mutex<HashMap<i64, BlockedByFetch>>,
814 comments: Mutex<Vec<(String, String)>>,
815 close_state: Mutex<HashMap<i64, (String, Option<String>)>>,
816 pulls: Mutex<HashMap<i64, Option<crate::backends::PullRequestCompletion>>>,
817 review_comments: Mutex<HashMap<i64, Vec<ScriptedReviewComment>>>,
818 reviewer: Mutex<Option<String>>,
819 }
820
821 impl ScriptedBackend {
822 fn new(blocked: HashMap<i64, BlockedByFetch>) -> Self {
823 Self {
824 blocked: Mutex::new(blocked),
825 comments: Mutex::new(Vec::new()),
826 close_state: Mutex::new(HashMap::new()),
827 pulls: Mutex::new(HashMap::new()),
828 review_comments: Mutex::new(HashMap::new()),
829 reviewer: Mutex::new(None),
830 }
831 }
832 }
833
834 #[async_trait]
835 impl ExternalBackend for ScriptedBackend {
836 fn name(&self) -> &str {
837 "scripted"
838 }
839
840 async fn create_task(
841 &self,
842 _title: &str,
843 _body: &str,
844 _labels: &[String],
845 ) -> anyhow::Result<ExternalId> {
846 Ok(ExternalId("1".into()))
847 }
848
849 async fn get_task(&self, id: &ExternalId) -> anyhow::Result<ExternalTask> {
850 Ok(ExternalTask {
851 id: id.clone(),
852 title: "t".into(),
853 body: String::new(),
854 state: "open".into(),
855 labels: vec![],
856 author: "bot".into(),
857 created_at: "2026-01-01T00:00:00Z".into(),
858 updated_at: "2026-01-01T00:00:00Z".into(),
859 url: String::new(),
860 })
861 }
862
863 async fn list_by_status(&self, _status: Status) -> anyhow::Result<Vec<ExternalTask>> {
864 Ok(vec![])
865 }
866
867 async fn post_comment(&self, id: &ExternalId, body: &str) -> anyhow::Result<()> {
868 self.comments
869 .lock()
870 .unwrap()
871 .push((id.0.clone(), body.to_string()));
872 Ok(())
873 }
874
875 async fn set_labels(&self, _id: &ExternalId, _labels: &[String]) -> anyhow::Result<()> {
876 Ok(())
877 }
878
879 async fn remove_label(&self, _id: &ExternalId, _label: &str) -> anyhow::Result<()> {
880 Ok(())
881 }
882
883 async fn get_sub_issues(&self, _id: &ExternalId) -> anyhow::Result<Vec<ExternalId>> {
884 Ok(vec![])
885 }
886
887 async fn health_check(&self) -> anyhow::Result<()> {
888 Ok(())
889 }
890
891 async fn blocked_by(
892 &self,
893 _repo: &str,
894 issue_number: i64,
895 ) -> anyhow::Result<BlockedByFetch> {
896 Ok(self
897 .blocked
898 .lock()
899 .unwrap()
900 .get(&issue_number)
901 .cloned()
902 .unwrap_or(BlockedByFetch::Unavailable))
903 }
904
905 async fn issue_close_state(
906 &self,
907 _repo: &str,
908 issue_number: i64,
909 ) -> anyhow::Result<Option<(String, Option<String>)>> {
910 Ok(self.close_state.lock().unwrap().get(&issue_number).cloned())
911 }
912
913 async fn pull_request_completion(
914 &self,
915 _repo: &str,
916 pr_number: i64,
917 ) -> anyhow::Result<Option<crate::backends::PullRequestCompletion>> {
918 let mut completion = self
919 .pulls
920 .lock()
921 .unwrap()
922 .get(&pr_number)
923 .cloned()
924 .unwrap_or(None);
925 if let Some(ref mut done) = completion {
926 if !done.approved {
927 let comments = self
928 .review_comments
929 .lock()
930 .unwrap()
931 .get(&pr_number)
932 .cloned()
933 .unwrap_or_default();
934 let reviewer = self.reviewer.lock().unwrap().clone().unwrap_or_default();
935 let mapped: Vec<crate::backends::github::AutomatedReviewComment<'_>> = comments
936 .iter()
937 .map(|comment| crate::backends::github::AutomatedReviewComment {
938 login: &comment.login,
939 body: &comment.body,
940 created_at: &comment.created_at,
941 })
942 .collect();
943 done.approved = crate::backends::github::latest_automated_review_approves(
944 &mapped, &reviewer,
945 );
946 }
947 }
948 Ok(completion)
949 }
950 }
951
952 async fn insert(store: &TaskStore, repo: &str, number: &str, title: &str, body: &str) -> i64 {
953 store
954 .create(&NewTask {
955 external_id: Some(number.to_string()),
956 repo: repo.to_string(),
957 origin: "github".to_string(),
958 title: title.to_string(),
959 body: body.to_string(),
960 source: "manual".to_string(),
961 source_id: String::new(),
962 author: String::new(),
963 url: String::new(),
964 labels: vec![],
965 parent_id: None,
966 })
967 .await
968 .unwrap()
969 }
970
971 fn native(repo: &str, number: i64, state: &str, reason: Option<&str>) -> BlockedByIssue {
972 BlockedByIssue {
973 github_id: number,
974 repo: repo.to_string(),
975 number,
976 state: state.to_string(),
977 state_reason: reason.map(str::to_string),
978 }
979 }
980
981 #[tokio::test]
982 async fn native_list_replaces_rows_and_empty_native_clears_them() {
983 let store = TaskStore::open_memory().await.unwrap();
984 let b = insert(&store, "owner/repo", "2", "B", "Depends on #1").await;
985 let mut blocked = HashMap::new();
986 blocked.insert(
987 2,
988 BlockedByFetch::Native(vec![native("owner/repo", 1, "open", None)]),
989 );
990 let backend = ScriptedBackend::new(blocked);
991 sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "Depends on #1")
992 .await
993 .unwrap();
994 assert_eq!(store.list_dependencies(b).await.unwrap().len(), 1);
995
996 backend
997 .blocked
998 .lock()
999 .unwrap()
1000 .insert(2, BlockedByFetch::Native(vec![]));
1001 sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "Depends on #1")
1002 .await
1003 .unwrap();
1004 assert!(store.list_dependencies(b).await.unwrap().is_empty());
1005 apply_claim_gates(&store).await.unwrap();
1006 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::New);
1007 }
1008
1009 #[tokio::test]
1010 async fn unavailable_keeps_rows_when_the_body_has_no_deps_and_failed_keeps_rows() {
1011 let store = TaskStore::open_memory().await.unwrap();
1012 let b = insert(&store, "owner/repo", "2", "B", "").await;
1013 store
1014 .replace_dependencies(
1015 b,
1016 &[NewDependency {
1017 repo: "owner/repo".into(),
1018 number: 1,
1019 source: DepSource::Cli,
1020 }],
1021 )
1022 .await
1023 .unwrap();
1024 let backend = ScriptedBackend::new(HashMap::new());
1025 sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "")
1026 .await
1027 .unwrap();
1028 assert_eq!(store.list_dependencies(b).await.unwrap()[0].source, "cli");
1029
1030 backend
1031 .blocked
1032 .lock()
1033 .unwrap()
1034 .insert(2, BlockedByFetch::Failed("boom".into()));
1035 sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "Depends on #9")
1036 .await
1037 .unwrap();
1038 assert_eq!(
1039 store.list_dependencies(b).await.unwrap()[0].depends_on_number,
1040 1
1041 );
1042 }
1043
1044 #[tokio::test]
1045 async fn unavailable_body_replaces_rows() {
1046 let store = TaskStore::open_memory().await.unwrap();
1047 let b = insert(
1048 &store,
1049 "acme/app",
1050 "8",
1051 "B",
1052 "Blocked by #12, #13\nDepends on owner/repo#12",
1053 )
1054 .await;
1055 let backend = ScriptedBackend::new(HashMap::new());
1056 sync_issue_dependencies(
1057 &backend,
1058 &store,
1059 b,
1060 "acme/app",
1061 Some(8),
1062 "Blocked by #12, #13\nDepends on owner/repo#12",
1063 )
1064 .await
1065 .unwrap();
1066 let deps = store.list_dependencies(b).await.unwrap();
1067 let labels: Vec<_> = deps
1068 .iter()
1069 .map(|dep| format!("{}#{}", dep.depends_on_repo, dep.depends_on_number))
1070 .collect();
1071 assert_eq!(
1072 labels,
1073 vec![
1074 "acme/app#12".to_string(),
1075 "acme/app#13".to_string(),
1076 "owner/repo#12".to_string()
1077 ]
1078 );
1079 assert!(deps.iter().all(|dep| dep.source == "body"));
1080 }
1081
1082 #[tokio::test]
1083 async fn cycle_comments_once_and_neither_task_is_claimable() {
1084 let store = TaskStore::open_memory().await.unwrap();
1085 let a = insert(&store, "owner/repo", "1", "A", "").await;
1086 let b = insert(&store, "owner/repo", "2", "B", "").await;
1087 let mut blocked = HashMap::new();
1088 blocked.insert(
1089 1,
1090 BlockedByFetch::Native(vec![native("owner/repo", 2, "open", None)]),
1091 );
1092 blocked.insert(
1093 2,
1094 BlockedByFetch::Native(vec![native("owner/repo", 1, "open", None)]),
1095 );
1096 let backend = ScriptedBackend::new(blocked);
1097 sync_issue_dependencies(&backend, &store, a, "owner/repo", Some(1), "")
1098 .await
1099 .unwrap();
1100 sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "")
1101 .await
1102 .unwrap();
1103 apply_claim_gates(&store).await.unwrap();
1104 assert_eq!(store.get(a).await.unwrap().status, TaskStatus::Waiting);
1105 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1106 let first = emit_dependency_notices(&backend, &store, None)
1107 .await
1108 .unwrap();
1109 assert_eq!(first.len(), 2);
1110 assert!(first
1111 .iter()
1112 .all(|notice| notice.comment.contains("owner/repo#1")));
1113 assert!(first
1114 .iter()
1115 .all(|notice| notice.comment.contains("owner/repo#2")));
1116 let second = emit_dependency_notices(&backend, &store, None)
1117 .await
1118 .unwrap();
1119 assert!(second.is_empty());
1120 assert_eq!(backend.comments.lock().unwrap().len(), 2);
1121 }
1122
1123 #[tokio::test]
1124 async fn not_planned_keeps_waiting_and_notifies_once() {
1125 let store = TaskStore::open_memory().await.unwrap();
1126 let b = insert(&store, "owner/repo", "2", "B", "").await;
1127 let mut blocked = HashMap::new();
1128 blocked.insert(
1129 2,
1130 BlockedByFetch::Native(vec![native("owner/repo", 1, "closed", Some("not_planned"))]),
1131 );
1132 let backend = ScriptedBackend::new(blocked);
1133 let transport = Transport::new();
1134 let mut rx = transport.subscribe_notifications();
1135 sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "")
1136 .await
1137 .unwrap();
1138 apply_claim_gates(&store).await.unwrap();
1139 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1140 let notices = emit_dependency_notices(&backend, &store, Some(&transport))
1141 .await
1142 .unwrap();
1143 assert_eq!(notices.len(), 1);
1144 assert!(notices[0].notify);
1145 assert!(notices[0].summary.contains("not planned"));
1146 let again = emit_dependency_notices(&backend, &store, Some(&transport))
1147 .await
1148 .unwrap();
1149 assert!(again.is_empty());
1150 let note = rx.try_recv().unwrap();
1151 assert_eq!(note.status, "failed");
1152 assert!(rx.try_recv().is_err());
1153 assert_eq!(backend.comments.lock().unwrap().len(), 1);
1154 }
1155
1156 #[tokio::test]
1157 async fn waiting_task_is_not_routed_until_the_blocker_is_done() {
1158 use crate::engine::router::{Router, RouterConfig};
1159 use crate::engine::tasks::TaskManager;
1160 use crate::engine::tick::{tick_route_tasks, RoutingTickQuota};
1161
1162 crate::engine::cooldown::reset_global_state().await;
1163 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1164 let a = insert(&store, "owner/repo", "1", "A", "").await;
1165 let b = insert(&store, "owner/repo", "2", "B", "").await;
1166 store
1167 .replace_dependencies(
1168 b,
1169 &[NewDependency {
1170 repo: "owner/repo".into(),
1171 number: 1,
1172 source: DepSource::Github,
1173 }],
1174 )
1175 .await
1176 .unwrap();
1177
1178 let backend: Arc<dyn ExternalBackend> =
1179 Arc::new(crate::backends::test_helpers::NoopBackend);
1180 let task_manager = Arc::new(TaskManager::with_store(
1181 backend.clone(),
1182 store.clone(),
1183 "owner/repo".into(),
1184 ));
1185 let mut tier = HashMap::new();
1186 tier.insert("claude".to_string(), vec!["haiku".to_string()]);
1187 let mut model_map = HashMap::new();
1188 model_map.insert("simple".into(), tier.clone());
1189 model_map.insert("medium".into(), tier.clone());
1190 model_map.insert("complex".into(), tier);
1191 let round_robin = RouterConfig {
1192 mode: "round_robin".into(),
1193 agents: vec!["claude".into()],
1194 model_map: model_map.clone(),
1195 ..Default::default()
1196 };
1197 let mut router = Router::new_for_test(round_robin, vec!["claude".into()]);
1198 let mut quota = RoutingTickQuota::new(10);
1199 tick_route_tasks(
1200 &backend,
1201 &task_manager,
1202 &mut router,
1203 &store,
1204 "owner/repo",
1205 &mut quota,
1206 )
1207 .await
1208 .unwrap();
1209 let waiting = store.get(b).await.unwrap();
1210 assert_eq!(waiting.status, TaskStatus::Waiting);
1211 assert!(waiting.agent.is_none());
1212
1213 let llm = RouterConfig {
1214 mode: "llm".into(),
1215 max_route_attempts: 0,
1216 agents: vec!["claude".into()],
1217 model_map,
1218 ..Default::default()
1219 };
1220 let mut llm_router = Router::new_for_test(llm, vec!["claude".into()]);
1221 let ext = crate::engine::tasks::store_task_to_external(&store.get(b).await.unwrap());
1222 store.update_status(b, TaskStatus::New).await.unwrap();
1223 let err = llm_router
1224 .route(&ext, &store, "owner/repo")
1225 .await
1226 .unwrap_err();
1227 assert!(err
1228 .downcast_ref::<crate::store::WaitingOnDependency>()
1229 .is_some());
1230 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1231 assert_eq!(store.get(b).await.unwrap().route_attempts, 0);
1232
1233 store.update_status(a, TaskStatus::Done).await.unwrap();
1234 let mut quota = RoutingTickQuota::new(10);
1235 tick_route_tasks(
1236 &backend,
1237 &task_manager,
1238 &mut router,
1239 &store,
1240 "owner/repo",
1241 &mut quota,
1242 )
1243 .await
1244 .unwrap();
1245 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1246
1247 store
1248 .save_blocker_snapshot(
1249 "owner/repo",
1250 1,
1251 &BlockerSnapshot::observe(
1252 "closed",
1253 Some("completed"),
1254 Some(&ObservedPullRequest {
1255 merged: true,
1256 base_is_default: true,
1257 approved: true,
1258 closed_unmerged: false,
1259 }),
1260 ),
1261 )
1262 .await
1263 .unwrap();
1264 let mut quota = RoutingTickQuota::new(10);
1265 tick_route_tasks(
1266 &backend,
1267 &task_manager,
1268 &mut router,
1269 &store,
1270 "owner/repo",
1271 &mut quota,
1272 )
1273 .await
1274 .unwrap();
1275 let routed = store.get(b).await.unwrap();
1276 assert_eq!(routed.status, TaskStatus::Routed);
1277 assert!(routed.agent.is_some());
1278 }
1279
1280 #[tokio::test]
1281 async fn round_robin_route_returns_waiting_on_dependency() {
1282 use crate::engine::router::{Router, RouterConfig};
1283 use crate::engine::tasks::store_task_to_external;
1284
1285 crate::engine::cooldown::reset_global_state().await;
1286 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1287 let b = insert(&store, "owner/repo", "2", "B", "").await;
1288 store
1289 .replace_dependencies(
1290 b,
1291 &[NewDependency {
1292 repo: "owner/repo".into(),
1293 number: 1,
1294 source: DepSource::Github,
1295 }],
1296 )
1297 .await
1298 .unwrap();
1299 let ext = store_task_to_external(&store.get(b).await.unwrap());
1300 let mut tier = HashMap::new();
1301 tier.insert("claude".to_string(), vec!["haiku".to_string()]);
1302 let mut model_map = HashMap::new();
1303 model_map.insert("medium".into(), tier);
1304 let config = RouterConfig {
1305 mode: "round_robin".into(),
1306 agents: vec!["claude".into()],
1307 model_map,
1308 ..Default::default()
1309 };
1310 let mut router = Router::new_for_test(config, vec!["claude".into()]);
1311 let err = router.route(&ext, &store, "owner/repo").await.unwrap_err();
1312 assert!(err
1313 .downcast_ref::<crate::store::WaitingOnDependency>()
1314 .is_some());
1315 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1316 assert_eq!(store.get(b).await.unwrap().route_attempts, 0);
1317 }
1318
1319 #[tokio::test]
1320 async fn refresh_holds_until_the_pr_is_approved_and_merged_into_the_default_branch() {
1321 use crate::backends::PullRequestCompletion;
1322
1323 let store = TaskStore::open_memory().await.unwrap();
1324 let a = insert(&store, "owner/repo", "1", "A", "").await;
1325 let b = insert(&store, "owner/repo", "2", "B", "").await;
1326 store
1327 .set_fields(a, &[("pr_number", serde_json::json!(10))])
1328 .await
1329 .unwrap();
1330 store
1331 .replace_dependencies(
1332 b,
1333 &[NewDependency {
1334 repo: "owner/repo".into(),
1335 number: 1,
1336 source: DepSource::Github,
1337 }],
1338 )
1339 .await
1340 .unwrap();
1341 let backend = ScriptedBackend::new(HashMap::new());
1342 backend
1343 .close_state
1344 .lock()
1345 .unwrap()
1346 .insert(1, ("closed".into(), Some("completed".into())));
1347
1348 let open_pr = PullRequestCompletion {
1349 merged: false,
1350 base_ref: "main".into(),
1351 default_branch: "main".into(),
1352 approved: false,
1353 closed: false,
1354 };
1355 backend.pulls.lock().unwrap().insert(10, Some(open_pr));
1356 refresh_blocker_snapshots(&backend, &store).await.unwrap();
1357 assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1358
1359 backend.pulls.lock().unwrap().insert(
1360 10,
1361 Some(PullRequestCompletion {
1362 merged: false,
1363 base_ref: "main".into(),
1364 default_branch: "main".into(),
1365 approved: true,
1366 closed: false,
1367 }),
1368 );
1369 refresh_blocker_snapshots(&backend, &store).await.unwrap();
1370 assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1371
1372 backend.pulls.lock().unwrap().insert(
1373 10,
1374 Some(PullRequestCompletion {
1375 merged: true,
1376 base_ref: "feature".into(),
1377 default_branch: "main".into(),
1378 approved: true,
1379 closed: true,
1380 }),
1381 );
1382 refresh_blocker_snapshots(&backend, &store).await.unwrap();
1383 assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1384
1385 backend.pulls.lock().unwrap().insert(
1386 10,
1387 Some(PullRequestCompletion {
1388 merged: false,
1389 base_ref: "main".into(),
1390 default_branch: "main".into(),
1391 approved: true,
1392 closed: true,
1393 }),
1394 );
1395 refresh_blocker_snapshots(&backend, &store).await.unwrap();
1396 assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1397
1398 backend.pulls.lock().unwrap().insert(
1399 10,
1400 Some(PullRequestCompletion {
1401 merged: true,
1402 base_ref: "main".into(),
1403 default_branch: "main".into(),
1404 approved: true,
1405 closed: true,
1406 }),
1407 );
1408 refresh_blocker_snapshots(&backend, &store).await.unwrap();
1409 assert_eq!(store.block_claim_if_waiting(b).await.unwrap(), None);
1410 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::New);
1411 }
1412
1413 #[tokio::test]
1414 async fn comment_approval_from_reviewer_releases_a_merged_blocker() {
1415 use crate::backends::PullRequestCompletion;
1416
1417 let store = TaskStore::open_memory().await.unwrap();
1418 let a = insert(&store, "owner/repo", "1", "A", "").await;
1419 let b = insert(&store, "owner/repo", "2", "B", "").await;
1420 store
1421 .set_fields(a, &[("pr_number", serde_json::json!(10))])
1422 .await
1423 .unwrap();
1424 store
1425 .replace_dependencies(
1426 b,
1427 &[NewDependency {
1428 repo: "owner/repo".into(),
1429 number: 1,
1430 source: DepSource::Github,
1431 }],
1432 )
1433 .await
1434 .unwrap();
1435 let backend = ScriptedBackend::new(HashMap::new());
1436 backend
1437 .close_state
1438 .lock()
1439 .unwrap()
1440 .insert(1, ("closed".into(), Some("completed".into())));
1441 backend.pulls.lock().unwrap().insert(
1442 10,
1443 Some(PullRequestCompletion {
1444 merged: true,
1445 base_ref: "main".into(),
1446 default_branch: "main".into(),
1447 approved: false,
1448 closed: true,
1449 }),
1450 );
1451 *backend.reviewer.lock().unwrap() = Some("orch-bot".into());
1452 backend.review_comments.lock().unwrap().insert(
1453 10,
1454 vec![ScriptedReviewComment {
1455 login: "mallory".into(),
1456 body: "## Automated Review \u{2014} Approve\n\nspoof".into(),
1457 created_at: "2026-01-03T00:00:00Z".into(),
1458 }],
1459 );
1460 refresh_blocker_snapshots(&backend, &store).await.unwrap();
1461 assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1462
1463 backend.review_comments.lock().unwrap().insert(
1464 10,
1465 vec![ScriptedReviewComment {
1466 login: "orch-bot".into(),
1467 body: "## Automated Review \u{2014} Approve\n\nLooks good.".into(),
1468 created_at: "2026-01-02T00:00:00Z".into(),
1469 }],
1470 );
1471 refresh_blocker_snapshots(&backend, &store).await.unwrap();
1472 assert_eq!(store.block_claim_if_waiting(b).await.unwrap(), None);
1473 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::New);
1474 }
1475
1476 #[tokio::test]
1477 async fn blocker_closed_before_import_releases_on_refresh() {
1478 let store = TaskStore::open_memory().await.unwrap();
1479 let waiter = insert(&store, "owner/repo", "24", "later", "").await;
1480 let mut blocked = HashMap::new();
1481 blocked.insert(
1482 24,
1483 BlockedByFetch::Native(vec![native("owner/repo", 14, "open", None)]),
1484 );
1485 let backend = ScriptedBackend::new(blocked);
1486 backend
1487 .close_state
1488 .lock()
1489 .unwrap()
1490 .insert(14, ("closed".into(), Some("completed".into())));
1491 reconcile_dependencies(&backend, &store, None)
1492 .await
1493 .unwrap();
1494 assert_eq!(store.get(waiter).await.unwrap().status, TaskStatus::New);
1495 let snap = store
1496 .blocker_snapshot("owner/repo", 14)
1497 .await
1498 .unwrap()
1499 .unwrap();
1500 assert!(snap.is_satisfied());
1501 assert_eq!(snap.issue_state, "closed");
1502 }
1503
1504 #[test]
1505 fn depends_on_token_accepts_a_bare_number_and_a_cross_repo_ref() {
1506 assert_eq!(
1507 parse_depends_on_token("12", "acme/app"),
1508 Some(DepRef {
1509 repo: "acme/app".into(),
1510 number: 12
1511 })
1512 );
1513 assert_eq!(
1514 parse_depends_on_token("owner/repo#12", "acme/app"),
1515 Some(DepRef {
1516 repo: "owner/repo".into(),
1517 number: 12
1518 })
1519 );
1520 }
1521
1522 #[tokio::test]
1523 async fn two_route_calls_cannot_claim_a_waiting_task() {
1524 use crate::engine::router::{Router, RouterConfig};
1525 use crate::engine::tasks::store_task_to_external;
1526
1527 crate::engine::cooldown::reset_global_state().await;
1528 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1529 let b = insert(&store, "owner/repo", "2", "B", "").await;
1530 store
1531 .replace_dependencies(
1532 b,
1533 &[NewDependency {
1534 repo: "owner/repo".into(),
1535 number: 1,
1536 source: DepSource::Github,
1537 }],
1538 )
1539 .await
1540 .unwrap();
1541 let ext = store_task_to_external(&store.get(b).await.unwrap());
1542 let mut tier = HashMap::new();
1543 tier.insert("claude".to_string(), vec!["haiku".to_string()]);
1544 let mut model_map = HashMap::new();
1545 model_map.insert("medium".into(), tier);
1546 let config = RouterConfig {
1547 mode: "round_robin".into(),
1548 agents: vec!["claude".into()],
1549 model_map,
1550 ..Default::default()
1551 };
1552 let left_store = store.clone();
1553 let right_store = store.clone();
1554 let left_task = ext.clone();
1555 let right_task = ext;
1556 let right_config = config.clone();
1557 let left = tokio::spawn(async move {
1558 let mut router = Router::new_for_test(config, vec!["claude".into()]);
1559 router.route(&left_task, &left_store, "owner/repo").await
1560 });
1561 let right = tokio::spawn(async move {
1562 let mut router = Router::new_for_test(right_config, vec!["claude".into()]);
1563 router.route(&right_task, &right_store, "owner/repo").await
1564 });
1565 assert!(left.await.unwrap().is_err());
1566 assert!(right.await.unwrap().is_err());
1567 let task = store.get(b).await.unwrap();
1568 assert_eq!(task.status, TaskStatus::Waiting);
1569 assert!(task.agent.is_none());
1570 }
1571
1572 #[tokio::test]
1573 async fn refuse_if_waiting_blocks_manual_claim() {
1574 let store = TaskStore::open_memory().await.unwrap();
1575 let b = insert(&store, "owner/repo", "2", "B", "").await;
1576 store
1577 .replace_dependencies(
1578 b,
1579 &[NewDependency {
1580 repo: "owner/repo".into(),
1581 number: 1,
1582 source: DepSource::Cli,
1583 }],
1584 )
1585 .await
1586 .unwrap();
1587 let err = refuse_if_waiting(&store, b).await.unwrap_err();
1588 assert!(err.to_string().contains("#1"));
1589 assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1590 }
1591}