orch/store/
dependencies.rs

1//! Issue dependency rows and the claim gate that keeps a waiting task from starting.
2//!
3//! A dependency is done when its blocker snapshot says the blocking pull request
4//! was approved and merged into the default branch, or the issue has no pull
5//! request and was closed as completed. Approval includes an orch
6//! `## Automated Review` comment from the reviewer account. A local `done` row
7//! is not enough.
8//! `not_planned` and `duplicate` are not done. `new`, `routed`, `waiting`,
9//! `blocked`, and `needs_review` move to `waiting` while a dependency is open.
10//! `in_progress` and `in_review` stay where they are.
11//! A `blocked-external` note in the issue body is an unmet blocker until a later
12//! import removes it
13
14use super::{TaskStatus, TaskStore};
15use crate::cluster::Command;
16use anyhow::Context;
17use sqlx::Row;
18use sqlx::SqliteConnection;
19use std::collections::HashMap;
20
21/// Returned by routing and dispatch when a task still has an open dependency.
22///
23/// Callers log this and leave the task alone. It does not count as a routing failure
24#[derive(Debug, thiserror::Error)]
25#[error("waiting on {}", .blockers.join(", "))]
26pub struct WaitingOnDependency {
27    pub blockers: Vec<String>,
28}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
31pub enum DepSource {
32    Github,
33    Body,
34    Cli,
35}
36
37impl DepSource {
38    pub fn as_str(self) -> &'static str {
39        match self {
40            Self::Github => "github",
41            Self::Body => "body",
42            Self::Cli => "cli",
43        }
44    }
45
46    pub fn parse(value: &str) -> Option<Self> {
47        match value {
48            "github" => Some(Self::Github),
49            "body" => Some(Self::Body),
50            "cli" => Some(Self::Cli),
51            _ => None,
52        }
53    }
54}
55
56#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
57pub struct NewDependency {
58    pub repo: String,
59    pub number: i64,
60    pub source: DepSource,
61}
62
63#[derive(Debug, Clone)]
64pub struct TaskDependency {
65    pub task_id: i64,
66    pub depends_on_task_id: Option<i64>,
67    pub depends_on_repo: String,
68    pub depends_on_number: i64,
69    pub source: String,
70    pub blocker_status: Option<String>,
71}
72
73#[derive(Debug, Clone)]
74pub struct SchedulableTask {
75    pub id: i64,
76    pub repo: String,
77    pub status: String,
78    pub external_id: Option<String>,
79    pub body: String,
80    pub title: String,
81}
82
83#[derive(Debug, Clone)]
84pub struct DepEdge {
85    pub task_id: i64,
86    pub task_repo: String,
87    pub task_external_id: Option<String>,
88    pub task_title: String,
89    pub depends_on_repo: String,
90    pub depends_on_number: i64,
91}
92
93#[derive(Debug, Clone)]
94struct BlockerView {
95    label: String,
96    satisfied: bool,
97}
98
99/// Pull request fields the engine observed on GitHub.
100///
101/// The store does not talk to GitHub. The engine fills this and the snapshot
102/// decides whether the blocker is done
103#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
104pub struct ObservedPullRequest {
105    pub merged: bool,
106    pub base_is_default: bool,
107    pub approved: bool,
108    pub closed_unmerged: bool,
109}
110
111/// What orch last read about one blocking issue.
112///
113/// Missing JSON fails closed: the dependency stays open
114#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
115pub struct BlockerSnapshot {
116    pub issue_state: String,
117    pub state_reason: Option<String>,
118    pub has_pr: bool,
119    pub pr_merged: bool,
120    pub pr_merged_into_default: bool,
121    pub pr_approved: bool,
122    pub pr_closed: bool,
123}
124
125impl BlockerSnapshot {
126    pub fn observe(
127        issue_state: &str,
128        state_reason: Option<&str>,
129        pr: Option<&ObservedPullRequest>,
130    ) -> Self {
131        let normalized = normalize_close_reason(issue_state, state_reason);
132        let (has_pr, merged, into_default, approved, pr_closed) = match pr {
133            Some(pr) => (
134                true,
135                pr.merged,
136                pr.base_is_default,
137                pr.approved,
138                pr.closed_unmerged,
139            ),
140            None => (false, false, false, false, false),
141        };
142        Self {
143            issue_state: if issue_state.eq_ignore_ascii_case("closed") {
144                "closed".to_string()
145            } else {
146                "open".to_string()
147            },
148            state_reason: Some(normalized.to_string()),
149            has_pr,
150            pr_merged: merged,
151            pr_merged_into_default: into_default,
152            pr_approved: approved,
153            pr_closed,
154        }
155    }
156
157    /// Record a new issue state without clearing pull request fields.
158    ///
159    /// A later issue sync must not wipe a snapshot that already knows the PR
160    pub fn note_issue_state(&mut self, issue_state: &str, state_reason: Option<&str>) {
161        let normalized = normalize_close_reason(issue_state, state_reason);
162        self.issue_state = if issue_state.eq_ignore_ascii_case("closed") {
163            "closed".to_string()
164        } else {
165            "open".to_string()
166        };
167        self.state_reason = Some(normalized.to_string());
168    }
169
170    pub fn is_satisfied(&self) -> bool {
171        let reason = self.state_reason.as_deref().unwrap_or("");
172        if reason == "not_planned" || reason == "duplicate" {
173            return false;
174        }
175        if self.has_pr {
176            return self.pr_merged && self.pr_merged_into_default && self.pr_approved;
177        }
178        self.issue_state == "closed" && (reason.is_empty() || reason == "completed")
179    }
180}
181
182/// Map a GitHub issue `state` and `state_reason` onto the values stored in KV.
183///
184/// A classic close with no reason counts as completed. An open issue counts as open
185pub fn normalize_close_reason(state: &str, state_reason: Option<&str>) -> &'static str {
186    if !state.eq_ignore_ascii_case("closed") {
187        return "open";
188    }
189    match state_reason
190        .unwrap_or("")
191        .trim()
192        .to_ascii_lowercase()
193        .as_str()
194    {
195        "not_planned" => "not_planned",
196        "duplicate" => "duplicate",
197        "reopened" => "open",
198        _ => "completed",
199    }
200}
201
202pub fn blocker_snapshot_key(repo: &str, number: i64) -> String {
203    format!("dep_blocker:{repo}#{number}")
204}
205
206pub fn worktree_refresh_key(task_id: i64) -> String {
207    format!("dep_worktree_refresh:{task_id}")
208}
209
210pub fn dep_label(task_repo: &str, dep_repo: &str, number: i64) -> String {
211    if task_repo == dep_repo {
212        format!("#{number}")
213    } else {
214        format!("{dep_repo}#{number}")
215    }
216}
217
218/// A missing snapshot is not done. Local task status is not an input
219pub fn blocker_is_satisfied(snapshot: Option<&BlockerSnapshot>) -> bool {
220    snapshot.is_some_and(BlockerSnapshot::is_satisfied)
221}
222
223/// `not_planned` and a still-open `blocked` task each get one notice.
224///
225/// `duplicate` does not. No snapshot means no notice
226pub fn blocker_failure_kind(
227    task_status: Option<&str>,
228    snapshot: Option<&BlockerSnapshot>,
229) -> Option<&'static str> {
230    let snap = snapshot?;
231    if snap.state_reason.as_deref() == Some("not_planned") {
232        return Some("not_planned");
233    }
234    if task_status == Some("blocked") && !snap.is_satisfied() {
235        return Some("blocked");
236    }
237    None
238}
239
240impl TaskStore {
241    /// Rename a repo slug on every stored row in one transaction.
242    ///
243    /// Refuses when the new slug already has tasks. Returns the number of moved tasks
244    pub async fn move_repo(&self, old: &str, new: &str) -> anyhow::Result<u64> {
245        let cmd = Command::MoveRepo {
246            old: old.to_string(),
247            new: new.to_string(),
248        };
249        Ok(u64::try_from(self.write_int(cmd).await?)?)
250    }
251
252    /// Replace one task's dependency rows in a single transaction.
253    pub async fn replace_dependencies(
254        &self,
255        task_id: i64,
256        rows: &[NewDependency],
257    ) -> anyhow::Result<()> {
258        let cmd = Command::DepReplace {
259            task_id,
260            rows: rows.to_vec(),
261        };
262        self.write(cmd).await.map(|_| ())
263    }
264
265    /// Point dependency rows at a task orch now knows.
266    pub async fn link_known_task(
267        &self,
268        task_id: i64,
269        repo: &str,
270        number: i64,
271    ) -> anyhow::Result<()> {
272        let cmd = Command::DepLinkKnown {
273            task_id,
274            repo: repo.to_string(),
275            number,
276        };
277        self.write(cmd).await.map(|_| ())
278    }
279
280    pub async fn list_dependencies(&self, task_id: i64) -> anyhow::Result<Vec<TaskDependency>> {
281        let rows = sqlx::query(
282            "SELECT d.task_id, d.depends_on_task_id, d.depends_on_repo, d.depends_on_number, d.source,
283                    t.status AS blocker_status
284             FROM task_dependencies d
285             LEFT JOIN tasks t
286               ON t.repo = d.depends_on_repo
287              AND t.external_id = CAST(d.depends_on_number AS TEXT)
288             WHERE d.task_id = ?
289             ORDER BY d.depends_on_repo, d.depends_on_number",
290        )
291        .bind(task_id)
292        .fetch_all(&self.pool)
293        .await?;
294        let mut out = Vec::with_capacity(rows.len());
295        for row in rows {
296            out.push(TaskDependency {
297                task_id: row.try_get("task_id")?,
298                depends_on_task_id: row.try_get("depends_on_task_id")?,
299                depends_on_repo: row.try_get("depends_on_repo")?,
300                depends_on_number: row.try_get("depends_on_number")?,
301                source: row.try_get("source")?,
302                blocker_status: row.try_get("blocker_status")?,
303            });
304        }
305        Ok(out)
306    }
307
308    pub async fn list_dependents(
309        &self,
310        repo: &str,
311        number: i64,
312    ) -> anyhow::Result<Vec<TaskDependency>> {
313        let rows = sqlx::query(
314            "SELECT d.task_id, d.depends_on_task_id, d.depends_on_repo, d.depends_on_number, d.source,
315                    t.status AS blocker_status
316             FROM task_dependencies d
317             JOIN tasks t ON t.id = d.task_id
318             WHERE d.depends_on_repo = ? AND d.depends_on_number = ?
319             ORDER BY d.task_id",
320        )
321        .bind(repo)
322        .bind(number)
323        .fetch_all(&self.pool)
324        .await?;
325        let mut out = Vec::with_capacity(rows.len());
326        for row in rows {
327            out.push(TaskDependency {
328                task_id: row.try_get("task_id")?,
329                depends_on_task_id: row.try_get("depends_on_task_id")?,
330                depends_on_repo: row.try_get("depends_on_repo")?,
331                depends_on_number: row.try_get("depends_on_number")?,
332                source: row.try_get("source")?,
333                blocker_status: row.try_get("blocker_status")?,
334            });
335        }
336        Ok(out)
337    }
338
339    pub async fn list_schedulable(&self) -> anyhow::Result<Vec<SchedulableTask>> {
340        let rows = sqlx::query(
341            "SELECT id, repo, status, external_id, body, title
342             FROM tasks
343             WHERE status IN ('new', 'routed', 'waiting')
344             ORDER BY id",
345        )
346        .fetch_all(&self.pool)
347        .await?;
348        let mut out = Vec::with_capacity(rows.len());
349        for row in rows {
350            out.push(SchedulableTask {
351                id: row.try_get("id")?,
352                repo: row.try_get("repo")?,
353                status: row.try_get("status")?,
354                external_id: row.try_get("external_id")?,
355                body: row.try_get("body")?,
356                title: row.try_get("title")?,
357            });
358        }
359        Ok(out)
360    }
361
362    pub async fn list_dependency_edges(&self) -> anyhow::Result<Vec<DepEdge>> {
363        let rows = sqlx::query(
364            "SELECT d.task_id, t.repo AS task_repo, t.external_id AS task_external_id, t.title AS task_title,
365                    d.depends_on_repo, d.depends_on_number
366             FROM task_dependencies d
367             JOIN tasks t ON t.id = d.task_id",
368        )
369        .fetch_all(&self.pool)
370        .await?;
371        let mut out = Vec::with_capacity(rows.len());
372        for row in rows {
373            out.push(DepEdge {
374                task_id: row.try_get("task_id")?,
375                task_repo: row.try_get("task_repo")?,
376                task_external_id: row.try_get("task_external_id")?,
377                task_title: row.try_get("task_title")?,
378                depends_on_repo: row.try_get("depends_on_repo")?,
379                depends_on_number: row.try_get("depends_on_number")?,
380            });
381        }
382        Ok(out)
383    }
384
385    /// Labels of unmet blockers, keyed by the waiting task id.
386    pub async fn unmet_labels_by_task(&self) -> anyhow::Result<HashMap<i64, Vec<String>>> {
387        let mut conn = self.pool.acquire().await?;
388        let ids: Vec<i64> = sqlx::query("SELECT DISTINCT task_id FROM task_dependencies")
389            .fetch_all(&mut *conn)
390            .await?
391            .into_iter()
392            .map(|row| row.try_get("task_id"))
393            .collect::<Result<Vec<_>, _>>()?;
394        let mut out = HashMap::new();
395        for task_id in ids {
396            let repo: String = sqlx::query("SELECT repo FROM tasks WHERE id = ?")
397                .bind(task_id)
398                .fetch_optional(&mut *conn)
399                .await?
400                .and_then(|row| row.try_get("repo").ok())
401                .unwrap_or_default();
402            let views = load_blockers(&mut conn, task_id, &repo).await?;
403            let labels: Vec<String> = views
404                .into_iter()
405                .filter(|view| !view.satisfied)
406                .map(|view| view.label)
407                .collect();
408            if !labels.is_empty() {
409                out.insert(task_id, labels);
410            }
411        }
412        Ok(out)
413    }
414
415    pub async fn unmet_blocker_labels(&self, task_id: i64) -> anyhow::Result<Vec<String>> {
416        let mut conn = self.pool.acquire().await?;
417        let repo: String = sqlx::query("SELECT repo FROM tasks WHERE id = ?")
418            .bind(task_id)
419            .fetch_optional(&mut *conn)
420            .await?
421            .and_then(|row| row.try_get("repo").ok())
422            .unwrap_or_default();
423        let views = load_blockers(&mut conn, task_id, &repo).await?;
424        Ok(views
425            .into_iter()
426            .filter(|view| !view.satisfied)
427            .map(|view| view.label)
428            .collect())
429    }
430
431    pub async fn save_blocker_snapshot(
432        &self,
433        repo: &str,
434        number: i64,
435        snap: &BlockerSnapshot,
436    ) -> anyhow::Result<()> {
437        let raw = serde_json::to_string(snap).context("encode blocker snapshot")?;
438        self.kv_set(&blocker_snapshot_key(repo, number), &raw).await
439    }
440
441    pub async fn blocker_snapshot(
442        &self,
443        repo: &str,
444        number: i64,
445    ) -> anyhow::Result<Option<BlockerSnapshot>> {
446        let Some(raw) = self.kv_get(&blocker_snapshot_key(repo, number)).await? else {
447            return Ok(None);
448        };
449        Ok(serde_json::from_str(&raw).ok())
450    }
451
452    /// Merge an issue state into the stored snapshot. Pull request fields stay.
453    pub async fn note_blocker_issue_state(
454        &self,
455        repo: &str,
456        number: i64,
457        state: &str,
458        reason: Option<&str>,
459    ) -> anyhow::Result<()> {
460        let mut snap = match self.blocker_snapshot(repo, number).await? {
461            Some(existing) => existing,
462            None => BlockerSnapshot::observe(state, reason, None),
463        };
464        snap.note_issue_state(state, reason);
465        self.save_blocker_snapshot(repo, number, &snap).await
466    }
467
468    /// Hold the task when a dependency is open.
469    ///
470    /// `None` means the task may be claimed. `Some` means it is waiting, and the
471    /// vec is the blocking issue labels. The check and the status write share one
472    /// immediate transaction, so two ticks cannot both route the same waiting task.
473    /// `in_progress` and `in_review` are not rewritten
474    pub async fn block_claim_if_waiting(
475        &self,
476        task_id: i64,
477    ) -> anyhow::Result<Option<Vec<String>>> {
478        match self.write(Command::DepBlockClaim { task_id }).await? {
479            crate::cluster::Reply::Unmet(unmet) => Ok(unmet),
480            other => anyhow::bail!("block_claim_if_waiting got {other:?}"),
481        }
482    }
483
484    /// Where `unblock` should leave a task. Waiting when any dependency is still open
485    pub async fn status_after_unblock(&self, task_id: i64) -> anyhow::Result<TaskStatus> {
486        match self.block_claim_if_waiting(task_id).await? {
487            Some(_) => Ok(TaskStatus::Waiting),
488            None => Ok(TaskStatus::New),
489        }
490    }
491}
492
493/// Plan-import note that a blocker outside the milestone has no GitHub issue yet
494pub fn blocked_external_from_body(body: &str) -> Vec<String> {
495    let mut ids = Vec::new();
496    let mut push = |token: &str| {
497        let token = token.trim();
498        if token.is_empty() {
499            return;
500        }
501        if !ids.iter().any(|have| have == token) {
502            ids.push(token.to_string());
503        }
504    };
505    for line in body.lines() {
506        let trimmed = line.trim();
507        if let Some(rest) = trimmed.strip_prefix("<!--") {
508            let rest = rest.trim();
509            let Some(rest) = rest.strip_prefix("orch-blocked-external:") else {
510                continue;
511            };
512            let rest = rest.trim().trim_end_matches("-->").trim();
513            for token in rest.split(',') {
514                push(token);
515            }
516            continue;
517        }
518        let lower = trimmed.to_ascii_lowercase();
519        if !lower.starts_with("blocked-external") {
520            continue;
521        }
522        let rest = trimmed["blocked-external".len()..]
523            .trim_start()
524            .trim_start_matches(':')
525            .trim();
526        let rest = rest.strip_suffix("(no issue yet)").unwrap_or(rest).trim();
527        for token in rest.split(',') {
528            push(token);
529        }
530    }
531    ids
532}
533
534async fn load_blockers(
535    conn: &mut SqliteConnection,
536    task_id: i64,
537    task_repo: &str,
538) -> anyhow::Result<Vec<BlockerView>> {
539    let rows = sqlx::query(
540        "SELECT depends_on_repo, depends_on_number
541         FROM task_dependencies
542         WHERE task_id = ?
543         ORDER BY depends_on_repo, depends_on_number",
544    )
545    .bind(task_id)
546    .fetch_all(&mut *conn)
547    .await?;
548    let mut views = Vec::with_capacity(rows.len());
549    for row in rows {
550        let dep_repo: String = row.try_get("depends_on_repo")?;
551        let number: i64 = row.try_get("depends_on_number")?;
552        let raw: Option<String> = sqlx::query("SELECT value FROM kv WHERE key = ?")
553            .bind(blocker_snapshot_key(&dep_repo, number))
554            .fetch_optional(&mut *conn)
555            .await?
556            .and_then(|found| found.try_get("value").ok());
557        let snap = raw.and_then(|text| serde_json::from_str::<BlockerSnapshot>(&text).ok());
558        views.push(BlockerView {
559            label: dep_label(task_repo, &dep_repo, number),
560            satisfied: snap.as_ref().is_some_and(BlockerSnapshot::is_satisfied),
561        });
562    }
563    Ok(views)
564}
565
566#[cfg(test)]
567mod tests {
568    use super::*;
569    use crate::store::{NewTask, TaskStore};
570
571    async fn task(store: &TaskStore, repo: &str, number: &str, title: &str) -> i64 {
572        task_with_body(store, repo, number, title, "").await
573    }
574
575    async fn task_with_body(
576        store: &TaskStore,
577        repo: &str,
578        number: &str,
579        title: &str,
580        body: &str,
581    ) -> i64 {
582        store
583            .create(&NewTask {
584                external_id: Some(number.to_string()),
585                repo: repo.to_string(),
586                origin: "github".to_string(),
587                title: title.to_string(),
588                body: body.to_string(),
589                source: "manual".to_string(),
590                source_id: String::new(),
591                author: String::new(),
592                url: String::new(),
593                labels: vec![],
594                parent_id: None,
595            })
596            .await
597            .unwrap()
598    }
599
600    fn approved_merge(into_default: bool) -> BlockerSnapshot {
601        BlockerSnapshot::observe(
602            "closed",
603            Some("completed"),
604            Some(&ObservedPullRequest {
605                merged: true,
606                base_is_default: into_default,
607                approved: true,
608                closed_unmerged: false,
609            }),
610        )
611    }
612
613    #[tokio::test]
614    async fn unmet_dependency_demotes_new_and_promotes_when_the_pr_is_done() {
615        let store = TaskStore::open_memory().await.unwrap();
616        let a = task(&store, "owner/repo", "1", "A").await;
617        let b = task(&store, "owner/repo", "2", "B").await;
618        store
619            .set_fields(
620                b,
621                &[
622                    ("branch", serde_json::json!("old-branch")),
623                    ("worktree", serde_json::json!("/tmp/old")),
624                ],
625            )
626            .await
627            .unwrap();
628        store
629            .replace_dependencies(
630                b,
631                &[NewDependency {
632                    repo: "owner/repo".to_string(),
633                    number: 1,
634                    source: DepSource::Github,
635                }],
636            )
637            .await
638            .unwrap();
639
640        let held = store.block_claim_if_waiting(b).await.unwrap();
641        assert_eq!(held, Some(vec!["#1".to_string()]));
642        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
643
644        store.update_status(a, TaskStatus::Done).await.unwrap();
645        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
646        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
647
648        store
649            .save_blocker_snapshot("owner/repo", 1, &approved_merge(true))
650            .await
651            .unwrap();
652        let released = store.block_claim_if_waiting(b).await.unwrap();
653        assert_eq!(released, None);
654        let task = store.get(b).await.unwrap();
655        assert_eq!(task.status, TaskStatus::New);
656        assert!(task.branch.is_empty());
657        assert!(task.worktree.is_empty());
658        assert_eq!(
659            store
660                .kv_get(&worktree_refresh_key(b))
661                .await
662                .unwrap()
663                .as_deref(),
664            Some("1")
665        );
666    }
667
668    #[tokio::test]
669    async fn not_planned_beats_a_merged_pr_and_a_completed_issue_with_no_pr_is_done() {
670        let store = TaskStore::open_memory().await.unwrap();
671        let a = task(&store, "owner/repo", "1", "A").await;
672        let b = task(&store, "owner/repo", "2", "B").await;
673        store.update_status(a, TaskStatus::Done).await.unwrap();
674        let mut not_planned = approved_merge(true);
675        not_planned.state_reason = Some("not_planned".to_string());
676        store
677            .save_blocker_snapshot("owner/repo", 1, &not_planned)
678            .await
679            .unwrap();
680        store
681            .replace_dependencies(
682                b,
683                &[NewDependency {
684                    repo: "owner/repo".to_string(),
685                    number: 1,
686                    source: DepSource::Github,
687                }],
688            )
689            .await
690            .unwrap();
691        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
692
693        let c = task(&store, "owner/repo", "3", "C").await;
694        store
695            .save_blocker_snapshot(
696                "other/repo",
697                9,
698                &BlockerSnapshot::observe("closed", Some("completed"), None),
699            )
700            .await
701            .unwrap();
702        store
703            .replace_dependencies(
704                c,
705                &[NewDependency {
706                    repo: "other/repo".to_string(),
707                    number: 9,
708                    source: DepSource::Body,
709                }],
710            )
711            .await
712            .unwrap();
713        assert_eq!(store.block_claim_if_waiting(c).await.unwrap(), None);
714    }
715
716    #[test]
717    fn pull_request_completion_cases() {
718        let open = BlockerSnapshot::observe("open", None, None);
719        assert!(!open.is_satisfied());
720
721        let pr_open = BlockerSnapshot::observe(
722            "open",
723            None,
724            Some(&ObservedPullRequest {
725                merged: false,
726                base_is_default: true,
727                approved: false,
728                closed_unmerged: false,
729            }),
730        );
731        assert!(!pr_open.is_satisfied());
732
733        let approved_open = BlockerSnapshot::observe(
734            "open",
735            None,
736            Some(&ObservedPullRequest {
737                merged: false,
738                base_is_default: true,
739                approved: true,
740                closed_unmerged: false,
741            }),
742        );
743        assert!(!approved_open.is_satisfied());
744
745        assert!(approved_merge(true).is_satisfied());
746        assert!(!approved_merge(false).is_satisfied());
747
748        let closed_unmerged = BlockerSnapshot::observe(
749            "closed",
750            Some("completed"),
751            Some(&ObservedPullRequest {
752                merged: false,
753                base_is_default: true,
754                approved: true,
755                closed_unmerged: true,
756            }),
757        );
758        assert!(!closed_unmerged.is_satisfied());
759
760        let merged_unapproved = BlockerSnapshot::observe(
761            "closed",
762            Some("completed"),
763            Some(&ObservedPullRequest {
764                merged: true,
765                base_is_default: true,
766                approved: false,
767                closed_unmerged: false,
768            }),
769        );
770        assert!(!merged_unapproved.is_satisfied());
771
772        assert!(BlockerSnapshot::observe("closed", None, None).is_satisfied());
773        assert!(BlockerSnapshot::observe("closed", Some("completed"), None).is_satisfied());
774        assert!(!BlockerSnapshot::observe("closed", Some("not_planned"), None).is_satisfied());
775        assert!(!BlockerSnapshot::observe("closed", Some("duplicate"), None).is_satisfied());
776    }
777
778    #[tokio::test]
779    async fn blocked_and_duplicate_stay_unmet_and_in_progress_is_not_rewritten() {
780        let store = TaskStore::open_memory().await.unwrap();
781        let a = task(&store, "owner/repo", "1", "A").await;
782        let b = task(&store, "owner/repo", "2", "B").await;
783        store.update_status(a, TaskStatus::Blocked).await.unwrap();
784        store
785            .replace_dependencies(
786                b,
787                &[NewDependency {
788                    repo: "owner/repo".to_string(),
789                    number: 1,
790                    source: DepSource::Cli,
791                }],
792            )
793            .await
794            .unwrap();
795        store
796            .update_status(b, TaskStatus::InProgress)
797            .await
798            .unwrap();
799        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
800        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::InProgress);
801
802        store
803            .save_blocker_snapshot(
804                "owner/repo",
805                1,
806                &BlockerSnapshot::observe("closed", Some("duplicate"), None),
807            )
808            .await
809            .unwrap();
810        store.update_status(a, TaskStatus::Done).await.unwrap();
811        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
812        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::InProgress);
813    }
814
815    #[tokio::test]
816    async fn concurrent_claims_both_report_waiting() {
817        let store = std::sync::Arc::new(TaskStore::open_memory().await.unwrap());
818        let b = task(&store, "owner/repo", "2", "B").await;
819        store
820            .replace_dependencies(
821                b,
822                &[NewDependency {
823                    repo: "owner/repo".to_string(),
824                    number: 1,
825                    source: DepSource::Github,
826                }],
827            )
828            .await
829            .unwrap();
830        let left = store.clone();
831        let right = store.clone();
832        let (a, c) = tokio::join!(
833            left.block_claim_if_waiting(b),
834            right.block_claim_if_waiting(b)
835        );
836        assert!(a.unwrap().is_some());
837        assert!(c.unwrap().is_some());
838        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
839    }
840
841    #[tokio::test]
842    async fn unblock_selection_skips_waiting_and_blocked_with_deps_stays_waiting() {
843        let store = TaskStore::open_memory().await.unwrap();
844        let a = task(&store, "owner/repo", "1", "A").await;
845        let b = task(&store, "owner/repo", "2", "B").await;
846        store.update_status(a, TaskStatus::Blocked).await.unwrap();
847        store
848            .replace_dependencies(
849                b,
850                &[NewDependency {
851                    repo: "owner/repo".to_string(),
852                    number: 1,
853                    source: DepSource::Github,
854                }],
855            )
856            .await
857            .unwrap();
858        store.block_claim_if_waiting(b).await.unwrap();
859        let blocked = store
860            .list_by_status("owner/repo", TaskStatus::Blocked)
861            .await
862            .unwrap();
863        assert!(blocked.iter().all(|task| task.id != b));
864        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
865        assert_eq!(
866            store.status_after_unblock(a).await.unwrap(),
867            TaskStatus::New
868        );
869        store.update_status(a, TaskStatus::Blocked).await.unwrap();
870        store
871            .replace_dependencies(
872                a,
873                &[NewDependency {
874                    repo: "owner/repo".to_string(),
875                    number: 9,
876                    source: DepSource::Cli,
877                }],
878            )
879            .await
880            .unwrap();
881        assert_eq!(
882            store.status_after_unblock(a).await.unwrap(),
883            TaskStatus::Waiting
884        );
885    }
886
887    #[tokio::test]
888    async fn entering_waiting_clears_a_stored_branch() {
889        let store = TaskStore::open_memory().await.unwrap();
890        let b = task(&store, "owner/repo", "2", "B").await;
891        store
892            .set_fields(
893                b,
894                &[
895                    ("branch", serde_json::json!("old-branch")),
896                    ("worktree", serde_json::json!("/tmp/old")),
897                ],
898            )
899            .await
900            .unwrap();
901        store
902            .replace_dependencies(
903                b,
904                &[NewDependency {
905                    repo: "owner/repo".to_string(),
906                    number: 1,
907                    source: DepSource::Github,
908                }],
909            )
910            .await
911            .unwrap();
912        store.block_claim_if_waiting(b).await.unwrap();
913        let task = store.get(b).await.unwrap();
914        assert_eq!(task.status, TaskStatus::Waiting);
915        assert!(task.branch.is_empty());
916        assert!(task.worktree.is_empty());
917    }
918
919    #[test]
920    fn blocked_external_from_body_reads_the_note_and_the_marker() {
921        let body = concat!(
922            "blocked-external: M1-05, M1-01 (no issue yet)\n\n",
923            "<!-- orch-blocked-external: M1-05, M1-01 -->\n",
924            "<!-- orch-plan: M6-01 -->\n",
925        );
926        assert_eq!(
927            blocked_external_from_body(body),
928            vec!["M1-05".to_string(), "M1-01".to_string()]
929        );
930        assert!(blocked_external_from_body("<!-- orch-plan: M6-01 -->").is_empty());
931    }
932
933    #[tokio::test]
934    async fn blocked_external_note_holds_the_task_and_clears_when_removed() {
935        let store = TaskStore::open_memory().await.unwrap();
936        let body = concat!(
937            "blocked-external: M3-01 (no issue yet)\n\n",
938            "<!-- orch-blocked-external: M3-01 -->\n",
939        );
940        let b = task_with_body(&store, "owner/repo", "4", "M4-01 API", body).await;
941        let held = store.block_claim_if_waiting(b).await.unwrap();
942        assert_eq!(held, Some(vec!["M3-01".to_string()]));
943        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
944
945        store
946            .upsert_external(&crate::store::UpsertExternal {
947                repo: "owner/repo",
948                ext_id: "4",
949                title: "M4-01 API",
950                body: "<!-- orch-plan: M4-01 -->\n",
951                author: "",
952                url: "",
953                labels: &[],
954                origin: "github",
955            })
956            .await
957            .unwrap();
958        let released = store.block_claim_if_waiting(b).await.unwrap();
959        assert_eq!(released, None);
960        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::New);
961    }
962
963    #[tokio::test]
964    async fn migration_028_applies_over_existing_tasks() {
965        let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
966        let options = sqlx::sqlite::SqliteConnectOptions::new()
967            .filename(":memory:")
968            .pragma("foreign_keys", "ON");
969        let pool = sqlx::sqlite::SqlitePoolOptions::new()
970            .max_connections(1)
971            .connect_with(options)
972            .await
973            .unwrap();
974        let mut files: Vec<_> = std::fs::read_dir(&dir)
975            .unwrap()
976            .map(|entry| entry.unwrap().path())
977            .filter(|path| path.extension().and_then(|ext| ext.to_str()) == Some("sql"))
978            .collect();
979        files.sort();
980        for path in &files {
981            let name = path.file_name().unwrap().to_string_lossy();
982            if name.starts_with("028") {
983                continue;
984            }
985            let sql = std::fs::read_to_string(path).unwrap();
986            sqlx::raw_sql(&sql)
987                .execute(&pool)
988                .await
989                .unwrap_or_else(|err| panic!("migration {name} failed: {err}"));
990        }
991        sqlx::query("INSERT INTO tasks (repo, title) VALUES ('owner/repo', 'existing')")
992            .execute(&pool)
993            .await
994            .unwrap();
995        let sql = std::fs::read_to_string(dir.join("028_task_dependencies.sql")).unwrap();
996        sqlx::raw_sql(&sql).execute(&pool).await.unwrap();
997        let (id, title): (i64, String) = sqlx::query_as("SELECT id, title FROM tasks")
998            .fetch_one(&pool)
999            .await
1000            .unwrap();
1001        assert_eq!(title, "existing");
1002        sqlx::query(
1003            "INSERT INTO task_dependencies (task_id, depends_on_repo, depends_on_number, source)
1004             VALUES (?, 'owner/repo', 1, 'cli')",
1005        )
1006        .bind(id)
1007        .execute(&pool)
1008        .await
1009        .unwrap();
1010        let (count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM task_dependencies")
1011            .fetch_one(&pool)
1012            .await
1013            .unwrap();
1014        assert_eq!(count, 1);
1015    }
1016}
1017
1018/// The SQL behind each replicated dependency write
1019pub(crate) mod apply {
1020    use super::*;
1021
1022    pub(crate) async fn move_repo(
1023        conn: &mut SqliteConnection,
1024        old: &str,
1025        new: &str,
1026    ) -> anyhow::Result<u64> {
1027        let existing: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tasks WHERE repo = ?")
1028            .bind(new)
1029            .fetch_one(&mut *conn)
1030            .await?;
1031        if existing > 0 {
1032            anyhow::bail!("{new} already has {existing} task(s), refusing to move {old} onto it");
1033        }
1034        let moved = sqlx::query("UPDATE tasks SET repo = ? WHERE repo = ?")
1035            .bind(new)
1036            .bind(old)
1037            .execute(&mut *conn)
1038            .await?
1039            .rows_affected();
1040        for sql in [
1041            "UPDATE task_dependencies SET depends_on_repo = ? WHERE depends_on_repo = ?",
1042            "UPDATE task_metrics SET repo = ? WHERE repo = ?",
1043            "UPDATE job_state SET repo = ? WHERE repo = ?",
1044            "UPDATE channel_subscriptions SET repo = ? WHERE repo = ?",
1045        ] {
1046            sqlx::query(sql)
1047                .bind(new)
1048                .bind(old)
1049                .execute(&mut *conn)
1050                .await?;
1051        }
1052        Ok(moved)
1053    }
1054
1055    pub(crate) async fn replace_dependencies(
1056        conn: &mut SqliteConnection,
1057        task_id: i64,
1058        rows: &[NewDependency],
1059    ) -> anyhow::Result<()> {
1060        sqlx::query("DELETE FROM task_dependencies WHERE task_id = ?")
1061            .bind(task_id)
1062            .execute(&mut *conn)
1063            .await?;
1064        for row in rows {
1065            if row.number <= 0 || row.repo.is_empty() {
1066                continue;
1067            }
1068            sqlx::query(
1069                "INSERT INTO task_dependencies
1070                 (task_id, depends_on_repo, depends_on_number, source)
1071                 VALUES (?, ?, ?, ?)",
1072            )
1073            .bind(task_id)
1074            .bind(&row.repo)
1075            .bind(row.number)
1076            .bind(row.source.as_str())
1077            .execute(&mut *conn)
1078            .await?;
1079        }
1080        sqlx::query(
1081            "UPDATE task_dependencies
1082             SET depends_on_task_id = (
1083                 SELECT id FROM tasks
1084                 WHERE tasks.repo = task_dependencies.depends_on_repo
1085                   AND tasks.external_id = CAST(task_dependencies.depends_on_number AS TEXT)
1086                 LIMIT 1
1087             )
1088             WHERE task_id = ?",
1089        )
1090        .bind(task_id)
1091        .execute(&mut *conn)
1092        .await?;
1093        Ok(())
1094    }
1095
1096    pub(crate) async fn link_known_task(
1097        conn: &mut SqliteConnection,
1098        task_id: i64,
1099        repo: &str,
1100        number: i64,
1101    ) -> anyhow::Result<()> {
1102        sqlx::query(
1103            "UPDATE task_dependencies
1104             SET depends_on_task_id = ?
1105             WHERE depends_on_repo = ? AND depends_on_number = ?",
1106        )
1107        .bind(task_id)
1108        .bind(repo)
1109        .bind(number)
1110        .execute(conn)
1111        .await?;
1112        Ok(())
1113    }
1114
1115    pub(crate) async fn block_claim(
1116        conn: &mut SqliteConnection,
1117        now: &str,
1118        task_id: i64,
1119    ) -> anyhow::Result<Option<Vec<String>>> {
1120        // sqlx 0.8.1 starts a deferred transaction. This write takes the reserved
1121        // lock before the read, so a second claim waits and then sees this one
1122        sqlx::query("UPDATE tasks SET updated_at = updated_at WHERE id = ?")
1123            .bind(task_id)
1124            .execute(&mut *conn)
1125            .await
1126            .context("lock task for dependency claim")?;
1127        let row = sqlx::query("SELECT status, agent, model, repo, body FROM tasks WHERE id = ?")
1128            .bind(task_id)
1129            .fetch_optional(&mut *conn)
1130            .await?;
1131        let Some(row) = row else {
1132            return Ok(None);
1133        };
1134        let status: String = row.try_get("status")?;
1135        let agent: Option<String> = row.try_get("agent")?;
1136        let model: Option<String> = row.try_get("model")?;
1137        let repo: String = row.try_get("repo")?;
1138        let body: String = row.try_get("body").unwrap_or_default();
1139        let views = load_blockers(conn, task_id, &repo).await?;
1140        let mut unmet: Vec<String> = views
1141            .iter()
1142            .filter(|view| !view.satisfied)
1143            .map(|view| view.label.clone())
1144            .collect();
1145        for id in blocked_external_from_body(&body) {
1146            if !unmet.iter().any(|have| have == &id) {
1147                unmet.push(id);
1148            }
1149        }
1150
1151        let holdable = matches!(
1152            status.as_str(),
1153            "new" | "routed" | "waiting" | "blocked" | "needs_review"
1154        );
1155        if unmet.is_empty() {
1156            if status == "waiting" {
1157                sqlx::query(
1158                    "UPDATE tasks
1159                     SET status = 'new', block_reason = NULL, branch = '', worktree = '',
1160                         updated_at = ?
1161                     WHERE id = ?",
1162                )
1163                .bind(now)
1164                .bind(task_id)
1165                .execute(&mut *conn)
1166                .await?;
1167                sqlx::query(
1168                    "INSERT INTO kv (key, value, updated_at)
1169                     VALUES (?, '1', ?)
1170                     ON CONFLICT(key) DO UPDATE
1171                       SET value = excluded.value, updated_at = excluded.updated_at",
1172                )
1173                .bind(worktree_refresh_key(task_id))
1174                .bind(now)
1175                .execute(&mut *conn)
1176                .await?;
1177                crate::store::tasks::apply::activity(
1178                    conn,
1179                    task_id,
1180                    "status_change",
1181                    Some("waiting"),
1182                    Some("new"),
1183                    agent.as_deref(),
1184                    model.as_deref(),
1185                    None,
1186                )
1187                .await?;
1188            }
1189            return Ok(None);
1190        }
1191
1192        if holdable && status != "waiting" {
1193            sqlx::query(
1194                "UPDATE tasks
1195                 SET status = 'waiting', branch = '', worktree = '', block_reason = NULL,
1196                     updated_at = ?
1197                 WHERE id = ?",
1198            )
1199            .bind(now)
1200            .bind(task_id)
1201            .execute(&mut *conn)
1202            .await?;
1203            crate::store::tasks::apply::activity(
1204                conn,
1205                task_id,
1206                "status_change",
1207                Some(&status),
1208                Some("waiting"),
1209                agent.as_deref(),
1210                model.as_deref(),
1211                None,
1212            )
1213            .await?;
1214        }
1215        Ok(Some(unmet))
1216    }
1217}
orch/store/dependencies.rs

orch/store/
dependencies.rs
