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

orch/store/
dependencies.rs
