orch/engine/
dependencies.rs

1//! Sync GitHub blocked-by links into SQLite and hold tasks until those links are done.
2//!
3//! Native `blocked_by` is the source of truth when the API answers. The issue
4//! body is used only when that API is unavailable. A failed read leaves stored
5//! rows in place.
6
7use 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
40/// Parse `Depends on` and `Blocked by` lines.
41///
42/// Bare `#12` uses `default_repo`. `owner/repo#12` keeps that repository.
43/// Lines inside fenced code blocks are ignored
44pub 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
89/// One `--depends-on` token: a bare issue number, `#12`, or `owner/repo#12`
90pub 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
170/// Replace one issue's dependency rows from GitHub, or from the body when GitHub cannot answer.
171pub 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                store
190                    .note_blocker_issue_state(
191                        &item.repo,
192                        item.number,
193                        &item.state,
194                        item.state_reason.as_deref(),
195                    )
196                    .await?;
197            }
198            let rows = list
199                .iter()
200                .map(|item| NewDependency {
201                    repo: item.repo.clone(),
202                    number: item.number,
203                    source: DepSource::Github,
204                })
205                .collect::<Vec<_>>();
206            store.replace_dependencies(task_id, &rows).await?;
207            store.link_known_task(task_id, repo, issue_number).await?;
208            Ok(SyncOutcome { cache: true })
209        }
210        BlockedByFetch::Unavailable => {
211            let parsed = parse_body_dependencies(body, repo);
212            if parsed.is_empty() {
213                store.link_known_task(task_id, repo, issue_number).await?;
214                return Ok(SyncOutcome { cache: true });
215            }
216            for dep in &parsed {
217                if store
218                    .get_by_external_id(&dep.repo, &dep.number.to_string())
219                    .await?
220                    .is_none()
221                {
222                    if let Ok(Some((state, reason))) =
223                        backend.issue_close_state(&dep.repo, dep.number).await
224                    {
225                        store
226                            .note_blocker_issue_state(
227                                &dep.repo,
228                                dep.number,
229                                &state,
230                                reason.as_deref(),
231                            )
232                            .await?;
233                    }
234                }
235            }
236            let rows = parsed
237                .into_iter()
238                .map(|dep| NewDependency {
239                    repo: dep.repo,
240                    number: dep.number,
241                    source: DepSource::Body,
242                })
243                .collect::<Vec<_>>();
244            store.replace_dependencies(task_id, &rows).await?;
245            store.link_known_task(task_id, repo, issue_number).await?;
246            Ok(SyncOutcome { cache: true })
247        }
248        BlockedByFetch::Failed(message) => {
249            tracing::warn!(
250                task_id,
251                repo,
252                issue_number,
253                error = %message,
254                "dependency read failed, keeping stored rows"
255            );
256            Ok(SyncOutcome { cache: false })
257        }
258    }
259}
260
261/// Move `new` and `routed` tasks with open dependencies to `waiting`, and release
262/// `waiting` tasks whose dependencies are done
263pub async fn apply_claim_gates(store: &TaskStore) -> anyhow::Result<()> {
264    let tasks = store.list_schedulable().await?;
265    for task in tasks {
266        store.block_claim_if_waiting(task.id).await?;
267    }
268    Ok(())
269}
270
271/// Refresh blocked-by links for schedulable tasks, then hold or release them.
272///
273/// A successful read is cached for 45 seconds so a 10 second tick does not
274/// fetch every task every time. Sync from ingest calls `sync_issue_dependencies`
275/// directly and does not use this cache
276pub async fn reconcile_dependencies(
277    backend: &dyn ExternalBackend,
278    store: &TaskStore,
279    transport: Option<&Transport>,
280) -> anyhow::Result<()> {
281    let tasks = store.list_schedulable().await?;
282    let now = chrono::Utc::now().timestamp();
283    for task in &tasks {
284        let Some(number) = task
285            .external_id
286            .as_deref()
287            .and_then(|id| id.parse::<i64>().ok())
288            .filter(|number| *number > 0)
289        else {
290            continue;
291        };
292        let key = format!("dep_synced_at:{}", task.id);
293        if let Ok(Some(prev)) = store.kv_get(&key).await {
294            if let Ok(prev) = prev.parse::<i64>() {
295                if now.saturating_sub(prev) < DEP_SYNC_TTL_SECS {
296                    continue;
297                }
298            }
299        }
300        match sync_issue_dependencies(
301            backend,
302            store,
303            task.id,
304            &task.repo,
305            Some(number),
306            &task.body,
307        )
308        .await
309        {
310            Ok(outcome) if outcome.cache => {
311                if let Err(err) = store.kv_set(&key, &now.to_string()).await {
312                    tracing::warn!(task_id = task.id, err = %err, "failed to cache dependency sync time");
313                }
314            }
315            Ok(_) => {}
316            Err(err) => {
317                tracing::warn!(task_id = task.id, repo = %task.repo, err = %err, "dependency sync failed");
318            }
319        }
320    }
321    if let Err(err) = refresh_blocker_snapshots(backend, store).await {
322        tracing::warn!(err = %err, "blocker snapshot refresh failed");
323    }
324    apply_claim_gates(store).await?;
325    emit_dependency_notices(backend, store, transport).await?;
326    Ok(())
327}
328
329/// Read each blocker's issue and pull request, then store the snapshot the claim gate reads.
330///
331/// A backend that returns no issue state and no pull request number leaves the
332/// stored snapshot in place. A known pull request that cannot be read is stored
333/// as open and unmerged
334pub async fn refresh_blocker_snapshots(
335    backend: &dyn ExternalBackend,
336    store: &TaskStore,
337) -> anyhow::Result<()> {
338    let edges = store.list_dependency_edges().await?;
339    let mut seen = HashSet::new();
340    for edge in edges {
341        if !seen.insert((edge.depends_on_repo.clone(), edge.depends_on_number)) {
342            continue;
343        }
344        if let Err(err) = refresh_one_blocker(
345            backend,
346            store,
347            &edge.depends_on_repo,
348            edge.depends_on_number,
349        )
350        .await
351        {
352            tracing::warn!(
353                repo = %edge.depends_on_repo,
354                number = edge.depends_on_number,
355                err = %err,
356                "blocker snapshot refresh failed"
357            );
358        }
359    }
360    Ok(())
361}
362
363async fn refresh_one_blocker(
364    backend: &dyn ExternalBackend,
365    store: &TaskStore,
366    repo: &str,
367    number: i64,
368) -> anyhow::Result<()> {
369    let close = match backend.issue_close_state(repo, number).await {
370        Ok(state) => state,
371        Err(err) => {
372            tracing::warn!(repo, number, err = %err, "issue close state read failed");
373            None
374        }
375    };
376    let task = store.get_by_external_id(repo, &number.to_string()).await?;
377    let pr_number = task
378        .as_ref()
379        .and_then(|row| row.pr_number)
380        .filter(|n| *n > 0);
381    if close.is_none() && pr_number.is_none() {
382        return Ok(());
383    }
384    let (issue_state, reason) = match close {
385        Some((state, reason)) => (state, reason),
386        None => match store.blocker_snapshot(repo, number).await? {
387            Some(snap) => (snap.issue_state, snap.state_reason),
388            None => ("open".to_string(), None),
389        },
390    };
391    let observed = if let Some(pr) = pr_number {
392        let completion = match backend.pull_request_completion(repo, i64::from(pr)).await {
393            Ok(value) => value,
394            Err(err) => {
395                tracing::warn!(repo, number, pr, err = %err, "pull request completion read failed");
396                None
397            }
398        };
399        Some(observe_completion(completion))
400    } else {
401        None
402    };
403    let snap = BlockerSnapshot::observe(&issue_state, reason.as_deref(), observed.as_ref());
404    store.save_blocker_snapshot(repo, number, &snap).await
405}
406
407fn observe_completion(completion: Option<PullRequestCompletion>) -> ObservedPullRequest {
408    match completion {
409        Some(done) => ObservedPullRequest {
410            merged: done.merged,
411            base_is_default: !done.default_branch.is_empty()
412                && done.base_ref == done.default_branch,
413            approved: done.approved,
414            closed_unmerged: done.closed && !done.merged,
415        },
416        None => ObservedPullRequest {
417            merged: false,
418            base_is_default: false,
419            approved: false,
420            closed_unmerged: false,
421        },
422    }
423}
424
425/// True when dispatch must not start this task. A check error also skips dispatch
426pub async fn dispatch_is_waiting(store: &TaskStore, repo: &str, external_id: &str) -> bool {
427    let id = match store.resolve_id_by_external(repo, external_id).await {
428        Ok(Some(id)) => id,
429        Ok(None) => return false,
430        Err(err) => {
431            tracing::warn!(
432                task_id = external_id,
433                err = %err,
434                "dependency check failed, skipping dispatch"
435            );
436            return true;
437        }
438    };
439    match store.block_claim_if_waiting(id).await {
440        Ok(Some(blockers)) => {
441            tracing::info!(
442                task_id = external_id,
443                blockers = %blockers.join(", "),
444                "skipping dispatch, waiting on a dependency"
445            );
446            true
447        }
448        Ok(None) => false,
449        Err(err) => {
450            tracing::warn!(
451                task_id = external_id,
452                err = %err,
453                "dependency check failed, skipping dispatch"
454            );
455            true
456        }
457    }
458}
459
460/// Stop a manual route, run, or retry while dependencies are open.
461pub async fn refuse_if_waiting(store: &TaskStore, task_id: i64) -> anyhow::Result<()> {
462    if let Some(blockers) = store.block_claim_if_waiting(task_id).await? {
463        anyhow::bail!("task {task_id} is waiting on {}", blockers.join(", "));
464    }
465    Ok(())
466}
467
468pub async fn emit_dependency_notices(
469    backend: &dyn ExternalBackend,
470    store: &TaskStore,
471    transport: Option<&Transport>,
472) -> anyhow::Result<Vec<DepNotice>> {
473    let mut notices = Vec::new();
474    notices.extend(cycle_notices(store).await?);
475    notices.extend(failure_notices(store).await?);
476    for notice in &notices {
477        if notice.issue_number > 0 && !notice.comment.is_empty() {
478            let id = ExternalId(notice.issue_number.to_string());
479            if let Err(err) = backend
480                .post_comment_in_repo(&notice.repo, &id, &notice.comment)
481                .await
482            {
483                tracing::warn!(
484                    repo = %notice.repo,
485                    issue = notice.issue_number,
486                    err = %err,
487                    "failed to comment on dependency notice"
488                );
489            }
490        }
491        if notice.notify {
492            if let Some(transport) = transport {
493                transport.push_notification(TaskNotification {
494                    task_id: if notice.issue_number > 0 {
495                        notice.issue_number.to_string()
496                    } else {
497                        notice.task_id.to_string()
498                    },
499                    title: notice.title.clone(),
500                    status: "failed".to_string(),
501                    agent: "orch".to_string(),
502                    duration_seconds: 0.0,
503                    summary: notice.summary.clone(),
504                    repo: Some(notice.repo.clone()),
505                    notify_target: None,
506                    pr_number: None,
507                });
508            }
509        }
510    }
511    Ok(notices)
512}
513
514async fn cycle_notices(store: &TaskStore) -> anyhow::Result<Vec<DepNotice>> {
515    let edges = store.list_dependency_edges().await?;
516    let mut snap_cache: HashMap<(String, i64), Option<BlockerSnapshot>> = HashMap::new();
517    let mut adj: HashMap<String, Vec<String>> = HashMap::new();
518    let mut issue_for_node: HashMap<String, (String, i64, i64, String)> = HashMap::new();
519
520    for edge in &edges {
521        let waiter = match edge
522            .task_external_id
523            .as_deref()
524            .and_then(|id| id.parse::<i64>().ok())
525        {
526            Some(number) if number > 0 => format!("{}#{}", edge.task_repo, number),
527            _ => format!("task:{}", edge.task_id),
528        };
529        if let Some(number) = edge
530            .task_external_id
531            .as_deref()
532            .and_then(|id| id.parse::<i64>().ok())
533            .filter(|number| *number > 0)
534        {
535            issue_for_node.insert(
536                waiter.clone(),
537                (
538                    edge.task_repo.clone(),
539                    number,
540                    edge.task_id,
541                    edge.task_title.clone(),
542                ),
543            );
544        }
545        let blocker_node = format!("{}#{}", edge.depends_on_repo, edge.depends_on_number);
546        let snap = cached_snapshot(
547            store,
548            &mut snap_cache,
549            &edge.depends_on_repo,
550            edge.depends_on_number,
551        )
552        .await?;
553        if blocker_is_satisfied(snap.as_ref()) {
554            continue;
555        }
556        adj.entry(waiter).or_default().push(blocker_node);
557    }
558
559    let mut notices = Vec::new();
560    for cycle in find_cycles(&adj) {
561        let signature = {
562            let mut nodes = cycle.clone();
563            nodes.sort();
564            nodes.join(",")
565        };
566        let mut named = cycle.clone();
567        if let Some(first) = named.first().cloned() {
568            named.push(first);
569        }
570        let path = named.join(" -> ");
571        let comment = format!(
572            "Dependency cycle: {path}. Orch will not run these tasks until the cycle is removed.{}",
573            crate::engine::orch_footer()
574        );
575        for node in &cycle {
576            let Some((repo, number, task_id, title)) = issue_for_node.get(node) else {
577                continue;
578            };
579            let key = format!("dep_cycle_comment:{repo}#{number}:{signature}");
580            if store.kv_get(&key).await?.is_some() {
581                continue;
582            }
583            store.kv_set(&key, "1").await?;
584            notices.push(DepNotice {
585                repo: repo.clone(),
586                issue_number: *number,
587                comment: comment.clone(),
588                notify: false,
589                summary: comment.clone(),
590                task_id: *task_id,
591                title: title.clone(),
592            });
593        }
594    }
595    Ok(notices)
596}
597
598async fn failure_notices(store: &TaskStore) -> anyhow::Result<Vec<DepNotice>> {
599    let edges = store.list_dependency_edges().await?;
600    let mut status_cache: HashMap<(String, i64), Option<String>> = HashMap::new();
601    let mut snap_cache: HashMap<(String, i64), Option<BlockerSnapshot>> = HashMap::new();
602    let mut notices = Vec::new();
603    for edge in edges {
604        let status = blocker_status(
605            store,
606            &mut status_cache,
607            &edge.depends_on_repo,
608            edge.depends_on_number,
609        )
610        .await?;
611        let snap = cached_snapshot(
612            store,
613            &mut snap_cache,
614            &edge.depends_on_repo,
615            edge.depends_on_number,
616        )
617        .await?;
618        let Some(kind) = blocker_failure_kind(status.as_deref(), snap.as_ref()) else {
619            continue;
620        };
621        let key = format!(
622            "dep_blocker_notice:{}:{}#{}:{kind}",
623            edge.task_id, edge.depends_on_repo, edge.depends_on_number
624        );
625        if store.kv_get(&key).await?.is_some() {
626            continue;
627        }
628        store.kv_set(&key, "1").await?;
629        let label = dep_label(
630            &edge.task_repo,
631            &edge.depends_on_repo,
632            edge.depends_on_number,
633        );
634        let (comment, summary) = match kind {
635            "not_planned" => {
636                let text = format!(
637                    "Blocked by {label}, which was closed as not planned. This task stays waiting."
638                );
639                (format!("{text}{}", crate::engine::orch_footer()), text)
640            }
641            _ => {
642                let text = format!(
643                    "Blocked by {label}, which is blocked and needs a person. This task stays waiting."
644                );
645                (format!("{text}{}", crate::engine::orch_footer()), text)
646            }
647        };
648        let issue_number = edge
649            .task_external_id
650            .as_deref()
651            .and_then(|id| id.parse::<i64>().ok())
652            .filter(|number| *number > 0)
653            .unwrap_or(0);
654        notices.push(DepNotice {
655            repo: edge.task_repo,
656            issue_number,
657            comment,
658            notify: true,
659            summary,
660            task_id: edge.task_id,
661            title: edge.task_title,
662        });
663    }
664    Ok(notices)
665}
666
667async fn cached_snapshot(
668    store: &TaskStore,
669    cache: &mut HashMap<(String, i64), Option<BlockerSnapshot>>,
670    repo: &str,
671    number: i64,
672) -> anyhow::Result<Option<BlockerSnapshot>> {
673    if let Some(found) = cache.get(&(repo.to_string(), number)) {
674        return Ok(found.clone());
675    }
676    let snap = store.blocker_snapshot(repo, number).await?;
677    cache.insert((repo.to_string(), number), snap.clone());
678    Ok(snap)
679}
680
681async fn blocker_status(
682    store: &TaskStore,
683    cache: &mut HashMap<(String, i64), Option<String>>,
684    repo: &str,
685    number: i64,
686) -> anyhow::Result<Option<String>> {
687    if let Some(status) = cache.get(&(repo.to_string(), number)) {
688        return Ok(status.clone());
689    }
690    let status = store
691        .get_by_external_id(repo, &number.to_string())
692        .await?
693        .map(|task| task.status.as_str().to_string());
694    cache.insert((repo.to_string(), number), status.clone());
695    Ok(status)
696}
697
698fn find_cycles(adj: &HashMap<String, Vec<String>>) -> Vec<Vec<String>> {
699    let mut nodes: Vec<String> = adj.keys().cloned().collect();
700    for targets in adj.values() {
701        for target in targets {
702            if !nodes.iter().any(|node| node == target) {
703                nodes.push(target.clone());
704            }
705        }
706    }
707    nodes.sort();
708    let mut color: HashMap<String, u8> = HashMap::new();
709    let mut stack = Vec::new();
710    let mut cycles = Vec::new();
711    let mut seen = HashSet::new();
712    for node in nodes {
713        if color.get(&node).copied().unwrap_or(0) == 0 {
714            visit(&node, adj, &mut color, &mut stack, &mut cycles, &mut seen);
715        }
716    }
717    cycles
718}
719
720fn visit(
721    node: &str,
722    adj: &HashMap<String, Vec<String>>,
723    color: &mut HashMap<String, u8>,
724    stack: &mut Vec<String>,
725    cycles: &mut Vec<Vec<String>>,
726    seen: &mut HashSet<String>,
727) {
728    color.insert(node.to_string(), 1);
729    stack.push(node.to_string());
730    if let Some(nexts) = adj.get(node) {
731        for next in nexts {
732            match color.get(next).copied().unwrap_or(0) {
733                1 => {
734                    if let Some(start) = stack.iter().position(|item| item == next) {
735                        let mut cycle = stack[start..].to_vec();
736                        if let Some(min_at) = cycle
737                            .iter()
738                            .enumerate()
739                            .min_by(|left, right| left.1.cmp(right.1))
740                            .map(|(index, _)| index)
741                        {
742                            cycle.rotate_left(min_at);
743                        }
744                        let signature = {
745                            let mut sorted = cycle.clone();
746                            sorted.sort();
747                            sorted.join(",")
748                        };
749                        if seen.insert(signature) {
750                            cycles.push(cycle);
751                        }
752                    }
753                }
754                0 => visit(next, adj, color, stack, cycles, seen),
755                _ => {}
756            }
757        }
758    }
759    stack.pop();
760    color.insert(node.to_string(), 2);
761}
762
763#[cfg(test)]
764mod tests {
765    use super::*;
766    use crate::backends::{BlockedByIssue, ExternalTask, Status};
767    use crate::store::{NewTask, TaskStatus};
768    use async_trait::async_trait;
769    use std::sync::{Arc, Mutex};
770
771    #[test]
772    fn body_parser_reads_the_three_forms_and_skips_fences() {
773        let body = "\
774Depends on #12
775Blocked by #12, #13
776Depends on owner/repo#12
777```
778Depends on #99
779```
780See #4 in the notes
781";
782        let deps = parse_body_dependencies(body, "acme/app");
783        assert_eq!(
784            deps,
785            vec![
786                DepRef {
787                    repo: "acme/app".into(),
788                    number: 12
789                },
790                DepRef {
791                    repo: "acme/app".into(),
792                    number: 13
793                },
794                DepRef {
795                    repo: "owner/repo".into(),
796                    number: 12
797                },
798            ]
799        );
800    }
801
802    struct ScriptedBackend {
803        blocked: Mutex<HashMap<i64, BlockedByFetch>>,
804        comments: Mutex<Vec<(String, String)>>,
805        close_state: Mutex<HashMap<i64, (String, Option<String>)>>,
806        pulls: Mutex<HashMap<i64, Option<crate::backends::PullRequestCompletion>>>,
807    }
808
809    impl ScriptedBackend {
810        fn new(blocked: HashMap<i64, BlockedByFetch>) -> Self {
811            Self {
812                blocked: Mutex::new(blocked),
813                comments: Mutex::new(Vec::new()),
814                close_state: Mutex::new(HashMap::new()),
815                pulls: Mutex::new(HashMap::new()),
816            }
817        }
818    }
819
820    #[async_trait]
821    impl ExternalBackend for ScriptedBackend {
822        fn name(&self) -> &str {
823            "scripted"
824        }
825
826        async fn create_task(
827            &self,
828            _title: &str,
829            _body: &str,
830            _labels: &[String],
831        ) -> anyhow::Result<ExternalId> {
832            Ok(ExternalId("1".into()))
833        }
834
835        async fn get_task(&self, id: &ExternalId) -> anyhow::Result<ExternalTask> {
836            Ok(ExternalTask {
837                id: id.clone(),
838                title: "t".into(),
839                body: String::new(),
840                state: "open".into(),
841                labels: vec![],
842                author: "bot".into(),
843                created_at: "2026-01-01T00:00:00Z".into(),
844                updated_at: "2026-01-01T00:00:00Z".into(),
845                url: String::new(),
846            })
847        }
848
849        async fn list_by_status(&self, _status: Status) -> anyhow::Result<Vec<ExternalTask>> {
850            Ok(vec![])
851        }
852
853        async fn post_comment(&self, id: &ExternalId, body: &str) -> anyhow::Result<()> {
854            self.comments
855                .lock()
856                .unwrap()
857                .push((id.0.clone(), body.to_string()));
858            Ok(())
859        }
860
861        async fn set_labels(&self, _id: &ExternalId, _labels: &[String]) -> anyhow::Result<()> {
862            Ok(())
863        }
864
865        async fn remove_label(&self, _id: &ExternalId, _label: &str) -> anyhow::Result<()> {
866            Ok(())
867        }
868
869        async fn get_sub_issues(&self, _id: &ExternalId) -> anyhow::Result<Vec<ExternalId>> {
870            Ok(vec![])
871        }
872
873        async fn health_check(&self) -> anyhow::Result<()> {
874            Ok(())
875        }
876
877        async fn blocked_by(
878            &self,
879            _repo: &str,
880            issue_number: i64,
881        ) -> anyhow::Result<BlockedByFetch> {
882            Ok(self
883                .blocked
884                .lock()
885                .unwrap()
886                .get(&issue_number)
887                .cloned()
888                .unwrap_or(BlockedByFetch::Unavailable))
889        }
890
891        async fn issue_close_state(
892            &self,
893            _repo: &str,
894            issue_number: i64,
895        ) -> anyhow::Result<Option<(String, Option<String>)>> {
896            Ok(self.close_state.lock().unwrap().get(&issue_number).cloned())
897        }
898
899        async fn pull_request_completion(
900            &self,
901            _repo: &str,
902            pr_number: i64,
903        ) -> anyhow::Result<Option<crate::backends::PullRequestCompletion>> {
904            Ok(self
905                .pulls
906                .lock()
907                .unwrap()
908                .get(&pr_number)
909                .cloned()
910                .unwrap_or(None))
911        }
912    }
913
914    async fn insert(store: &TaskStore, repo: &str, number: &str, title: &str, body: &str) -> i64 {
915        store
916            .create(&NewTask {
917                external_id: Some(number.to_string()),
918                repo: repo.to_string(),
919                origin: "github".to_string(),
920                title: title.to_string(),
921                body: body.to_string(),
922                source: "manual".to_string(),
923                source_id: String::new(),
924                author: String::new(),
925                url: String::new(),
926                labels: vec![],
927                parent_id: None,
928            })
929            .await
930            .unwrap()
931    }
932
933    fn native(repo: &str, number: i64, state: &str, reason: Option<&str>) -> BlockedByIssue {
934        BlockedByIssue {
935            github_id: number,
936            repo: repo.to_string(),
937            number,
938            state: state.to_string(),
939            state_reason: reason.map(str::to_string),
940        }
941    }
942
943    #[tokio::test]
944    async fn native_list_replaces_rows_and_empty_native_clears_them() {
945        let store = TaskStore::open_memory().await.unwrap();
946        let b = insert(&store, "owner/repo", "2", "B", "Depends on #1").await;
947        let mut blocked = HashMap::new();
948        blocked.insert(
949            2,
950            BlockedByFetch::Native(vec![native("owner/repo", 1, "open", None)]),
951        );
952        let backend = ScriptedBackend::new(blocked);
953        sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "Depends on #1")
954            .await
955            .unwrap();
956        assert_eq!(store.list_dependencies(b).await.unwrap().len(), 1);
957
958        backend
959            .blocked
960            .lock()
961            .unwrap()
962            .insert(2, BlockedByFetch::Native(vec![]));
963        sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "Depends on #1")
964            .await
965            .unwrap();
966        assert!(store.list_dependencies(b).await.unwrap().is_empty());
967        apply_claim_gates(&store).await.unwrap();
968        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::New);
969    }
970
971    #[tokio::test]
972    async fn unavailable_keeps_rows_when_the_body_has_no_deps_and_failed_keeps_rows() {
973        let store = TaskStore::open_memory().await.unwrap();
974        let b = insert(&store, "owner/repo", "2", "B", "").await;
975        store
976            .replace_dependencies(
977                b,
978                &[NewDependency {
979                    repo: "owner/repo".into(),
980                    number: 1,
981                    source: DepSource::Cli,
982                }],
983            )
984            .await
985            .unwrap();
986        let backend = ScriptedBackend::new(HashMap::new());
987        sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "")
988            .await
989            .unwrap();
990        assert_eq!(store.list_dependencies(b).await.unwrap()[0].source, "cli");
991
992        backend
993            .blocked
994            .lock()
995            .unwrap()
996            .insert(2, BlockedByFetch::Failed("boom".into()));
997        sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "Depends on #9")
998            .await
999            .unwrap();
1000        assert_eq!(
1001            store.list_dependencies(b).await.unwrap()[0].depends_on_number,
1002            1
1003        );
1004    }
1005
1006    #[tokio::test]
1007    async fn unavailable_body_replaces_rows() {
1008        let store = TaskStore::open_memory().await.unwrap();
1009        let b = insert(
1010            &store,
1011            "acme/app",
1012            "8",
1013            "B",
1014            "Blocked by #12, #13\nDepends on owner/repo#12",
1015        )
1016        .await;
1017        let backend = ScriptedBackend::new(HashMap::new());
1018        sync_issue_dependencies(
1019            &backend,
1020            &store,
1021            b,
1022            "acme/app",
1023            Some(8),
1024            "Blocked by #12, #13\nDepends on owner/repo#12",
1025        )
1026        .await
1027        .unwrap();
1028        let deps = store.list_dependencies(b).await.unwrap();
1029        let labels: Vec<_> = deps
1030            .iter()
1031            .map(|dep| format!("{}#{}", dep.depends_on_repo, dep.depends_on_number))
1032            .collect();
1033        assert_eq!(
1034            labels,
1035            vec![
1036                "acme/app#12".to_string(),
1037                "acme/app#13".to_string(),
1038                "owner/repo#12".to_string()
1039            ]
1040        );
1041        assert!(deps.iter().all(|dep| dep.source == "body"));
1042    }
1043
1044    #[tokio::test]
1045    async fn cycle_comments_once_and_neither_task_is_claimable() {
1046        let store = TaskStore::open_memory().await.unwrap();
1047        let a = insert(&store, "owner/repo", "1", "A", "").await;
1048        let b = insert(&store, "owner/repo", "2", "B", "").await;
1049        let mut blocked = HashMap::new();
1050        blocked.insert(
1051            1,
1052            BlockedByFetch::Native(vec![native("owner/repo", 2, "open", None)]),
1053        );
1054        blocked.insert(
1055            2,
1056            BlockedByFetch::Native(vec![native("owner/repo", 1, "open", None)]),
1057        );
1058        let backend = ScriptedBackend::new(blocked);
1059        sync_issue_dependencies(&backend, &store, a, "owner/repo", Some(1), "")
1060            .await
1061            .unwrap();
1062        sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "")
1063            .await
1064            .unwrap();
1065        apply_claim_gates(&store).await.unwrap();
1066        assert_eq!(store.get(a).await.unwrap().status, TaskStatus::Waiting);
1067        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1068        let first = emit_dependency_notices(&backend, &store, None)
1069            .await
1070            .unwrap();
1071        assert_eq!(first.len(), 2);
1072        assert!(first
1073            .iter()
1074            .all(|notice| notice.comment.contains("owner/repo#1")));
1075        assert!(first
1076            .iter()
1077            .all(|notice| notice.comment.contains("owner/repo#2")));
1078        let second = emit_dependency_notices(&backend, &store, None)
1079            .await
1080            .unwrap();
1081        assert!(second.is_empty());
1082        assert_eq!(backend.comments.lock().unwrap().len(), 2);
1083    }
1084
1085    #[tokio::test]
1086    async fn not_planned_keeps_waiting_and_notifies_once() {
1087        let store = TaskStore::open_memory().await.unwrap();
1088        let b = insert(&store, "owner/repo", "2", "B", "").await;
1089        let mut blocked = HashMap::new();
1090        blocked.insert(
1091            2,
1092            BlockedByFetch::Native(vec![native("owner/repo", 1, "closed", Some("not_planned"))]),
1093        );
1094        let backend = ScriptedBackend::new(blocked);
1095        let transport = Transport::new();
1096        let mut rx = transport.subscribe_notifications();
1097        sync_issue_dependencies(&backend, &store, b, "owner/repo", Some(2), "")
1098            .await
1099            .unwrap();
1100        apply_claim_gates(&store).await.unwrap();
1101        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1102        let notices = emit_dependency_notices(&backend, &store, Some(&transport))
1103            .await
1104            .unwrap();
1105        assert_eq!(notices.len(), 1);
1106        assert!(notices[0].notify);
1107        assert!(notices[0].summary.contains("not planned"));
1108        let again = emit_dependency_notices(&backend, &store, Some(&transport))
1109            .await
1110            .unwrap();
1111        assert!(again.is_empty());
1112        let note = rx.try_recv().unwrap();
1113        assert_eq!(note.status, "failed");
1114        assert!(rx.try_recv().is_err());
1115        assert_eq!(backend.comments.lock().unwrap().len(), 1);
1116    }
1117
1118    #[tokio::test]
1119    async fn waiting_task_is_not_routed_until_the_blocker_is_done() {
1120        use crate::engine::router::{Router, RouterConfig};
1121        use crate::engine::tasks::TaskManager;
1122        use crate::engine::tick::{tick_route_tasks, RoutingTickQuota};
1123
1124        crate::engine::cooldown::reset_global_state().await;
1125        let store = Arc::new(TaskStore::open_memory().await.unwrap());
1126        let a = insert(&store, "owner/repo", "1", "A", "").await;
1127        let b = insert(&store, "owner/repo", "2", "B", "").await;
1128        store
1129            .replace_dependencies(
1130                b,
1131                &[NewDependency {
1132                    repo: "owner/repo".into(),
1133                    number: 1,
1134                    source: DepSource::Github,
1135                }],
1136            )
1137            .await
1138            .unwrap();
1139
1140        let backend: Arc<dyn ExternalBackend> =
1141            Arc::new(crate::backends::test_helpers::NoopBackend);
1142        let task_manager = Arc::new(TaskManager::with_store(
1143            backend.clone(),
1144            store.clone(),
1145            "owner/repo".into(),
1146        ));
1147        let mut tier = HashMap::new();
1148        tier.insert("claude".to_string(), vec!["haiku".to_string()]);
1149        let mut model_map = HashMap::new();
1150        model_map.insert("simple".into(), tier.clone());
1151        model_map.insert("medium".into(), tier.clone());
1152        model_map.insert("complex".into(), tier);
1153        let round_robin = RouterConfig {
1154            mode: "round_robin".into(),
1155            agents: vec!["claude".into()],
1156            model_map: model_map.clone(),
1157            ..Default::default()
1158        };
1159        let mut router = Router::new_for_test(round_robin, vec!["claude".into()]);
1160        let mut quota = RoutingTickQuota::new(10);
1161        tick_route_tasks(
1162            &backend,
1163            &task_manager,
1164            &mut router,
1165            &store,
1166            "owner/repo",
1167            &mut quota,
1168        )
1169        .await
1170        .unwrap();
1171        let waiting = store.get(b).await.unwrap();
1172        assert_eq!(waiting.status, TaskStatus::Waiting);
1173        assert!(waiting.agent.is_none());
1174
1175        let llm = RouterConfig {
1176            mode: "llm".into(),
1177            max_route_attempts: 0,
1178            agents: vec!["claude".into()],
1179            model_map,
1180            ..Default::default()
1181        };
1182        let mut llm_router = Router::new_for_test(llm, vec!["claude".into()]);
1183        let ext = crate::engine::tasks::store_task_to_external(&store.get(b).await.unwrap());
1184        store.update_status(b, TaskStatus::New).await.unwrap();
1185        let err = llm_router
1186            .route(&ext, &store, "owner/repo")
1187            .await
1188            .unwrap_err();
1189        assert!(err
1190            .downcast_ref::<crate::store::WaitingOnDependency>()
1191            .is_some());
1192        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1193        assert_eq!(store.get(b).await.unwrap().route_attempts, 0);
1194
1195        store.update_status(a, TaskStatus::Done).await.unwrap();
1196        let mut quota = RoutingTickQuota::new(10);
1197        tick_route_tasks(
1198            &backend,
1199            &task_manager,
1200            &mut router,
1201            &store,
1202            "owner/repo",
1203            &mut quota,
1204        )
1205        .await
1206        .unwrap();
1207        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1208
1209        store
1210            .save_blocker_snapshot(
1211                "owner/repo",
1212                1,
1213                &BlockerSnapshot::observe(
1214                    "closed",
1215                    Some("completed"),
1216                    Some(&ObservedPullRequest {
1217                        merged: true,
1218                        base_is_default: true,
1219                        approved: true,
1220                        closed_unmerged: false,
1221                    }),
1222                ),
1223            )
1224            .await
1225            .unwrap();
1226        let mut quota = RoutingTickQuota::new(10);
1227        tick_route_tasks(
1228            &backend,
1229            &task_manager,
1230            &mut router,
1231            &store,
1232            "owner/repo",
1233            &mut quota,
1234        )
1235        .await
1236        .unwrap();
1237        let routed = store.get(b).await.unwrap();
1238        assert_eq!(routed.status, TaskStatus::Routed);
1239        assert!(routed.agent.is_some());
1240    }
1241
1242    #[tokio::test]
1243    async fn round_robin_route_returns_waiting_on_dependency() {
1244        use crate::engine::router::{Router, RouterConfig};
1245        use crate::engine::tasks::store_task_to_external;
1246
1247        crate::engine::cooldown::reset_global_state().await;
1248        let store = Arc::new(TaskStore::open_memory().await.unwrap());
1249        let b = insert(&store, "owner/repo", "2", "B", "").await;
1250        store
1251            .replace_dependencies(
1252                b,
1253                &[NewDependency {
1254                    repo: "owner/repo".into(),
1255                    number: 1,
1256                    source: DepSource::Github,
1257                }],
1258            )
1259            .await
1260            .unwrap();
1261        let ext = store_task_to_external(&store.get(b).await.unwrap());
1262        let mut tier = HashMap::new();
1263        tier.insert("claude".to_string(), vec!["haiku".to_string()]);
1264        let mut model_map = HashMap::new();
1265        model_map.insert("medium".into(), tier);
1266        let config = RouterConfig {
1267            mode: "round_robin".into(),
1268            agents: vec!["claude".into()],
1269            model_map,
1270            ..Default::default()
1271        };
1272        let mut router = Router::new_for_test(config, vec!["claude".into()]);
1273        let err = router.route(&ext, &store, "owner/repo").await.unwrap_err();
1274        assert!(err
1275            .downcast_ref::<crate::store::WaitingOnDependency>()
1276            .is_some());
1277        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1278        assert_eq!(store.get(b).await.unwrap().route_attempts, 0);
1279    }
1280
1281    #[tokio::test]
1282    async fn refresh_holds_until_the_pr_is_approved_and_merged_into_the_default_branch() {
1283        use crate::backends::PullRequestCompletion;
1284
1285        let store = TaskStore::open_memory().await.unwrap();
1286        let a = insert(&store, "owner/repo", "1", "A", "").await;
1287        let b = insert(&store, "owner/repo", "2", "B", "").await;
1288        store
1289            .set_fields(a, &[("pr_number", serde_json::json!(10))])
1290            .await
1291            .unwrap();
1292        store
1293            .replace_dependencies(
1294                b,
1295                &[NewDependency {
1296                    repo: "owner/repo".into(),
1297                    number: 1,
1298                    source: DepSource::Github,
1299                }],
1300            )
1301            .await
1302            .unwrap();
1303        let backend = ScriptedBackend::new(HashMap::new());
1304        backend
1305            .close_state
1306            .lock()
1307            .unwrap()
1308            .insert(1, ("closed".into(), Some("completed".into())));
1309
1310        let open_pr = PullRequestCompletion {
1311            merged: false,
1312            base_ref: "main".into(),
1313            default_branch: "main".into(),
1314            approved: false,
1315            closed: false,
1316        };
1317        backend.pulls.lock().unwrap().insert(10, Some(open_pr));
1318        refresh_blocker_snapshots(&backend, &store).await.unwrap();
1319        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1320
1321        backend.pulls.lock().unwrap().insert(
1322            10,
1323            Some(PullRequestCompletion {
1324                merged: false,
1325                base_ref: "main".into(),
1326                default_branch: "main".into(),
1327                approved: true,
1328                closed: false,
1329            }),
1330        );
1331        refresh_blocker_snapshots(&backend, &store).await.unwrap();
1332        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1333
1334        backend.pulls.lock().unwrap().insert(
1335            10,
1336            Some(PullRequestCompletion {
1337                merged: true,
1338                base_ref: "feature".into(),
1339                default_branch: "main".into(),
1340                approved: true,
1341                closed: true,
1342            }),
1343        );
1344        refresh_blocker_snapshots(&backend, &store).await.unwrap();
1345        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1346
1347        backend.pulls.lock().unwrap().insert(
1348            10,
1349            Some(PullRequestCompletion {
1350                merged: false,
1351                base_ref: "main".into(),
1352                default_branch: "main".into(),
1353                approved: true,
1354                closed: true,
1355            }),
1356        );
1357        refresh_blocker_snapshots(&backend, &store).await.unwrap();
1358        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
1359
1360        backend.pulls.lock().unwrap().insert(
1361            10,
1362            Some(PullRequestCompletion {
1363                merged: true,
1364                base_ref: "main".into(),
1365                default_branch: "main".into(),
1366                approved: true,
1367                closed: true,
1368            }),
1369        );
1370        refresh_blocker_snapshots(&backend, &store).await.unwrap();
1371        assert_eq!(store.block_claim_if_waiting(b).await.unwrap(), None);
1372        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::New);
1373    }
1374
1375    #[test]
1376    fn depends_on_token_accepts_a_bare_number_and_a_cross_repo_ref() {
1377        assert_eq!(
1378            parse_depends_on_token("12", "acme/app"),
1379            Some(DepRef {
1380                repo: "acme/app".into(),
1381                number: 12
1382            })
1383        );
1384        assert_eq!(
1385            parse_depends_on_token("owner/repo#12", "acme/app"),
1386            Some(DepRef {
1387                repo: "owner/repo".into(),
1388                number: 12
1389            })
1390        );
1391    }
1392
1393    #[tokio::test]
1394    async fn two_route_calls_cannot_claim_a_waiting_task() {
1395        use crate::engine::router::{Router, RouterConfig};
1396        use crate::engine::tasks::store_task_to_external;
1397
1398        crate::engine::cooldown::reset_global_state().await;
1399        let store = Arc::new(TaskStore::open_memory().await.unwrap());
1400        let b = insert(&store, "owner/repo", "2", "B", "").await;
1401        store
1402            .replace_dependencies(
1403                b,
1404                &[NewDependency {
1405                    repo: "owner/repo".into(),
1406                    number: 1,
1407                    source: DepSource::Github,
1408                }],
1409            )
1410            .await
1411            .unwrap();
1412        let ext = store_task_to_external(&store.get(b).await.unwrap());
1413        let mut tier = HashMap::new();
1414        tier.insert("claude".to_string(), vec!["haiku".to_string()]);
1415        let mut model_map = HashMap::new();
1416        model_map.insert("medium".into(), tier);
1417        let config = RouterConfig {
1418            mode: "round_robin".into(),
1419            agents: vec!["claude".into()],
1420            model_map,
1421            ..Default::default()
1422        };
1423        let left_store = store.clone();
1424        let right_store = store.clone();
1425        let left_task = ext.clone();
1426        let right_task = ext;
1427        let right_config = config.clone();
1428        let left = tokio::spawn(async move {
1429            let mut router = Router::new_for_test(config, vec!["claude".into()]);
1430            router.route(&left_task, &left_store, "owner/repo").await
1431        });
1432        let right = tokio::spawn(async move {
1433            let mut router = Router::new_for_test(right_config, vec!["claude".into()]);
1434            router.route(&right_task, &right_store, "owner/repo").await
1435        });
1436        assert!(left.await.unwrap().is_err());
1437        assert!(right.await.unwrap().is_err());
1438        let task = store.get(b).await.unwrap();
1439        assert_eq!(task.status, TaskStatus::Waiting);
1440        assert!(task.agent.is_none());
1441    }
1442
1443    #[tokio::test]
1444    async fn refuse_if_waiting_blocks_manual_claim() {
1445        let store = TaskStore::open_memory().await.unwrap();
1446        let b = insert(&store, "owner/repo", "2", "B", "").await;
1447        store
1448            .replace_dependencies(
1449                b,
1450                &[NewDependency {
1451                    repo: "owner/repo".into(),
1452                    number: 1,
1453                    source: DepSource::Cli,
1454                }],
1455            )
1456            .await
1457            .unwrap();
1458        let err = refuse_if_waiting(&store, b).await.unwrap_err();
1459        assert!(err.to_string().contains("#1"));
1460        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
1461    }
1462}
orch/engine/dependencies.rs

orch/engine/
dependencies.rs
