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

orch/store/
dependencies.rs
