orch/store/
dependencies.rs

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

orch/store/
dependencies.rs
