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

orch/store/
dependencies.rs
