orch/store/
dependencies.rs

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

orch/store/
dependencies.rs
