magi/

blockers.rs

//! Dependency inventory for `blocked` tasks.
//!
//! [`crate::daemon::resolve_blockers`] frees a task when its dependency is
//! `done` (or its question answered) and quarantines it when the dependency no
//! longer exists. It has nothing to say about a dependency that is alive but
//! *stuck*: a `held` task never becomes `done` on its own, so everything
//! waiting on it waits forever, and nothing shows why. [`Inventory`] answers
//! that from one snapshot of the queue and the questions:
//!
//! - [`Inventory::waits_on`] - what a blocked task waits on and each
//!   dependency's state, following a chain of blocked dependencies
//!   (`4135 (blocked → 9db7 held)`), for `magi task list` and the web UI.
//! - [`Inventory::stuck_roots`] - the tasks nothing in the loop will ever run
//!   that a blocked task is frozen behind. [`crate::triage`] asks about each
//!   such root once, naming everything it freezes.
//!
//! A blocked task is *stuck* when it has at least one unresolved dependency
//! and every one of them is either `held` or itself a stuck blocked task.
//! Anything that can still make progress - a queued, running or failed
//! (retried) task, an open question, a dependency that no longer exists (which
//! `resolve_blockers` quarantines) - makes it not stuck. A dependency cycle
//! terminates: its members are stuck, and the smallest id in the cycle is its
//! root.

use std::collections::{BTreeMap, BTreeSet};

use crate::ask::{Question, QuestionStatus};
use crate::queue::{Task, TaskStatus};

/// How many hops [`Inventory::waits_on`] follows before it stops describing.
const CHAIN_DEPTH: usize = 4;

/// One snapshot of the tasks and questions a `blocked_by` id can name.
#[derive(Debug, Clone, Default)]
pub struct Inventory {
    tasks: BTreeMap<String, Task>,
    questions: BTreeMap<String, QuestionStatus>,
    ids: Vec<String>,
    complete: bool,
}

enum Dep<'a> {
    Task(&'a Task),
    Question(QuestionStatus),
    Missing,
    Unknown,
}

type Roots = Option<BTreeSet<String>>;

/// The last dash-separated part of an id: what `Task::short` shows.
fn short(id: &str) -> &str {
    id.split('-').next_back().unwrap_or(id)
}

impl Inventory {
    /// Build the snapshot. Callers take `queue.list()` and `questions.list()`
    /// once, so a listing over many tasks never rescans the disk per task.
    pub fn new(tasks: Vec<Task>, questions: &[Question]) -> Self {
        let ids = tasks
            .iter()
            .map(|t| t.id.clone())
            .chain(questions.iter().map(|q| q.id.clone()))
            .collect();
        Self {
            ids,
            complete: true,
            tasks: tasks.into_iter().map(|t| (t.id.clone(), t)).collect(),
            questions: questions.iter().map(|q| (q.id.clone(), q.status)).collect(),
        }
    }

    /// Include unreadable records and tombstones in id resolution; list()
    /// alone cannot prove uniqueness or absence.
    pub fn load(queue: &crate::queue::Queue, questions: &crate::ask::Questions) -> Self {
        Self::load_with_tasks(queue, questions, queue.list())
    }

    /// Reuse a listing already read by the caller. Still include tombstones
    /// and unreadable ids when resolving dependency ambiguity.
    pub fn load_with_tasks(
        queue: &crate::queue::Queue,
        questions: &crate::ask::Questions,
        tasks: Vec<Task>,
    ) -> Self {
        let mut inventory = Self::new(tasks, &questions.list());
        let ids = (|| -> anyhow::Result<Vec<String>> {
            let mut ids = crate::queue::record_ids(queue.root(), "json")?;
            ids.extend(crate::queue::record_ids(questions.root(), "json")?);
            ids.extend(crate::queue::record_ids(queue.root(), "removed")?);
            ids.sort();
            ids.dedup();
            Ok(ids)
        })();
        match ids {
            Ok(ids) => inventory.ids = ids,
            Err(_) => inventory.complete = false,
        }
        inventory
    }

    /// Read only the dependency closure for a detail view. Names across both
    /// stores (including tombstones and corrupt records) still establish
    /// ambiguity, and unreadable records remain unknown rather than resolved.
    pub fn load_for(
        queue: &crate::queue::Queue,
        questions: &crate::ask::Questions,
        task: &Task,
    ) -> Self {
        Self::load_for_tasks(queue, questions, std::slice::from_ref(task))
    }

    /// Share id enumeration and dependency reads across one list page.
    pub fn load_for_tasks(
        queue: &crate::queue::Queue,
        questions: &crate::ask::Questions,
        tasks: &[Task],
    ) -> Self {
        let mut inventory = Self::new(tasks.to_vec(), &[]);
        if tasks.iter().all(|task| task.blocked_by.is_empty()) {
            return inventory;
        }
        let ids = (|| -> anyhow::Result<Vec<String>> {
            let mut ids = crate::queue::record_ids(queue.root(), "json")?;
            ids.extend(crate::queue::record_ids(questions.root(), "json")?);
            ids.extend(crate::queue::record_ids(queue.root(), "removed")?);
            ids.sort();
            ids.dedup();
            Ok(ids)
        })();
        match ids {
            Ok(ids) => inventory.ids = ids,
            Err(_) => inventory.complete = false,
        }
        let mut pending: Vec<_> = tasks
            .iter()
            .flat_map(|task| task.blocked_by.iter().cloned())
            .collect();
        let mut seen = BTreeSet::new();
        while let Some(reference) = pending.pop() {
            let full =
                if queue.path_of(&reference).is_file() || questions.path_of(&reference).is_file() {
                    reference
                } else if inventory.complete {
                    match crate::queue::resolve_reference(
                        &reference,
                        inventory.ids.iter().map(String::as_str),
                    ) {
                        crate::queue::Reference::Found(full) => full,
                        _ => continue,
                    }
                } else {
                    continue;
                };
            if !seen.insert(full.clone()) {
                continue;
            }
            if queue.path_of(&full).is_file()
                && let Ok(t) = queue.get(&full)
            {
                pending.extend(t.blocked_by.iter().cloned());
                inventory.tasks.insert(full.clone(), t);
            }
            if questions.path_of(&full).is_file()
                && let Ok(q) = questions.get(&full)
            {
                inventory.questions.insert(full, q.status);
            }
        }
        inventory
    }

    fn dep(&self, id: &str) -> Dep<'_> {
        // A partial scan does not invalidate readable exact records. It does
        // prevent proving that a prefix/suffix is unique across both stores.
        if let Some(t) = self.tasks.get(id) {
            return Dep::Task(t);
        }
        if let Some(s) = self.questions.get(id) {
            return Dep::Question(*s);
        }
        if !self.complete {
            return Dep::Unknown;
        }
        match crate::queue::resolve_reference(id, self.ids.iter().map(String::as_str)) {
            crate::queue::Reference::Found(full) => {
                if let Some(t) = self.tasks.get(&full) {
                    Dep::Task(t)
                } else if let Some(s) = self.questions.get(&full) {
                    Dep::Question(*s)
                } else {
                    Dep::Unknown
                }
            }
            crate::queue::Reference::Missing => Dep::Missing,
            crate::queue::Reference::Ambiguous => Dep::Unknown,
        }
    }

    /// A dependency that has nothing left to wait for.
    fn resolved(&self, id: &str) -> bool {
        match self.dep(id) {
            Dep::Task(t) => t.status == TaskStatus::Done,
            Dep::Question(s) => s == QuestionStatus::Answered,
            Dep::Missing | Dep::Unknown => false,
        }
    }

    fn walk(
        &self,
        id: &str,
        path: &mut Vec<String>,
        memo: &mut BTreeMap<String, Roots>,
    ) -> (Roots, bool) {
        if let Some(v) = memo.get(id) {
            return (v.clone(), false);
        }
        let Some(task) = self.tasks.get(id) else {
            return (None, false);
        };
        path.push(id.to_owned());
        let mut roots = BTreeSet::new();
        let mut moving = false;
        let mut cyclic = false;
        for b in &task.blocked_by {
            match self.dep(b) {
                Dep::Task(t) => match t.status {
                    TaskStatus::Done => {}
                    TaskStatus::Held => {
                        roots.insert(t.id.clone());
                    }
                    TaskStatus::Blocked => {
                        if let Some(pos) = path.iter().position(|p| p == &t.id) {
                            // A cycle: its smallest id speaks for all of it, so
                            // every entry point names the same root.
                            if let Some(min) = path[pos..].iter().min() {
                                roots.insert(min.clone());
                            }
                            cyclic = true;
                        } else {
                            let (sub, c) = self.walk(&t.id, path, memo);
                            cyclic |= c;
                            match sub {
                                Some(r) => roots.extend(r),
                                None => moving = true,
                            }
                        }
                    }
                    TaskStatus::Queued
                    | TaskStatus::Running
                    | TaskStatus::Failed
                    | TaskStatus::Parked => {
                        moving = true;
                    }
                },
                Dep::Question(QuestionStatus::Answered) => {}
                Dep::Question(_) | Dep::Missing | Dep::Unknown => moving = true,
            }
        }
        path.pop();
        let verdict = (!moving && !roots.is_empty()).then_some(roots);
        if !cyclic {
            memo.insert(id.to_owned(), verdict.clone());
        }
        (verdict, cyclic)
    }

    /// Every stuck `blocked` task, with the root ids it is frozen behind.
    pub fn stuck(&self) -> BTreeMap<String, BTreeSet<String>> {
        let mut memo = BTreeMap::new();
        let mut out = BTreeMap::new();
        for (id, t) in &self.tasks {
            if t.status != TaskStatus::Blocked {
                continue;
            }
            let (verdict, _) = self.walk(id, &mut Vec::new(), &mut memo);
            if let Some(roots) = verdict {
                out.insert(id.clone(), roots);
            }
        }
        out
    }

    /// The roots `task` is frozen behind; empty when it is not stuck.
    pub fn stuck_roots(&self, task: &Task) -> BTreeSet<String> {
        if task.status != TaskStatus::Blocked {
            return BTreeSet::new();
        }
        self.walk(&task.id, &mut Vec::new(), &mut BTreeMap::new())
            .0
            .unwrap_or_default()
    }

    /// Each dependency of `task` with its state, e.g. `9db7 (held)` or
    /// `4135 (blocked → 9db7 held)`. Empty unless `task` is `blocked`.
    pub fn waits_on(&self, task: &Task) -> Vec<String> {
        if task.status != TaskStatus::Blocked {
            return Vec::new();
        }
        task.blocked_by
            .iter()
            .map(|b| {
                let chain = self.chain(b, &mut vec![task.id.clone()]);
                format!("{} ({chain})", short(b))
            })
            .collect()
    }

    fn chain(&self, id: &str, seen: &mut Vec<String>) -> String {
        match self.dep(id) {
            Dep::Missing => "missing".to_owned(),
            Dep::Unknown => "unavailable or ambiguous".to_owned(),
            Dep::Question(s) => format!("question {}", s.as_str()),
            Dep::Task(t) => {
                let mut out = t.status.as_str().to_owned();
                if t.status != TaskStatus::Blocked {
                    return out;
                }
                if seen.iter().any(|s| s == &t.id) {
                    return format!("{out}, cycle");
                }
                if seen.len() > CHAIN_DEPTH {
                    return format!("{out} → …");
                }
                seen.push(t.id.clone());
                if let Some(next) = t.blocked_by.iter().find(|b| !self.resolved(b)) {
                    out.push_str(&format!(" → {} {}", short(next), self.chain(next, seen)));
                }
                out
            }
        }
    }

    /// The task with this id, if it is one.
    pub fn task(&self, id: &str) -> Option<&Task> {
        match self.dep(id) {
            Dep::Task(t) => Some(t),
            _ => None,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::queue::Source;
    use std::path::PathBuf;

    fn t(id: &str, status: TaskStatus, blocked_by: &[&str]) -> Task {
        let mut t = Task::new(
            id.to_owned(),
            "x".to_owned(),
            PathBuf::from("/repo"),
            Source::Human,
        );
        t.id = format!("2026-{id}");
        t.status = status;
        t.blocked_by = blocked_by.iter().map(|b| format!("2026-{b}")).collect();
        t
    }

    fn inv(tasks: Vec<Task>) -> Inventory {
        Inventory::new(tasks, &[])
    }

    #[test]
    fn responsiveness_dependency_closure_matches_the_full_inventory() {
        let dir = tempfile::tempdir().unwrap();
        let queue = crate::queue::Queue::at(dir.path().join("queue"));
        let questions = crate::ask::Questions::at(dir.path().join("questions"));
        let mut held = t("held", TaskStatus::Held, &[]);
        let mut middle = t("middle", TaskStatus::Blocked, &["held"]);
        let mut waiting = t("waiting", TaskStatus::Blocked, &["middle"]);
        for task in [&mut held, &mut middle, &mut waiting] {
            queue.put(task).unwrap();
        }
        for n in 0..150 {
            let mut unrelated = t(&format!("noise{n}"), TaskStatus::Done, &[]);
            unrelated.instruction = "history ".repeat(4096);
            queue.put(&mut unrelated).unwrap();
        }
        for reference in ["2026-middle", "middle", "missing", "collision"] {
            waiting.blocked_by = vec![reference.into()];
            queue.put(&mut waiting).unwrap();
            std::fs::write(queue.root().join("a-collision.json"), "corrupt").unwrap();
            std::fs::write(queue.root().join("b-collision.removed"), "removed").unwrap();
            let full = Inventory::load(&queue, &questions);
            let small = Inventory::load_for(&queue, &questions, &waiting);
            assert_eq!(small.waits_on(&waiting), full.waits_on(&waiting));
            assert_eq!(small.stuck_roots(&waiting), full.stuck_roots(&waiting));
            assert!(small.tasks.len() <= 3);
        }
        // A cycle uses the same recursion and root rules as the full inventory.
        held.status = TaskStatus::Blocked;
        held.blocked_by = vec![middle.id.clone()];
        queue.put(&mut held).unwrap();
        waiting.blocked_by = vec![middle.id.clone()];
        queue.put(&mut waiting).unwrap();
        let full = Inventory::load(&queue, &questions);
        let small = Inventory::load_for(&queue, &questions, &waiting);
        assert_eq!(small.stuck_roots(&waiting), full.stuck_roots(&waiting));
    }

    #[test]
    fn dependency_resolution_partial_scan_keeps_readable_exact_records() {
        let dir = tempfile::tempdir().unwrap();
        let queue = crate::queue::Queue::at(dir.path().join("queue"));
        let questions = crate::ask::Questions::at(dir.path().join("questions"));
        let mut running = t("feb2", TaskStatus::Running, &[]);
        let mut middle = t("ca5b", TaskStatus::Blocked, &["feb2"]);
        let mut waiting = t("ee37", TaskStatus::Blocked, &["ca5b"]);
        let mut held = t("held", TaskStatus::Held, &[]);
        let mut short_wait = t("short", TaskStatus::Blocked, &[]);
        short_wait.blocked_by = vec!["held".into()];
        for task in [
            &mut running,
            &mut middle,
            &mut waiting,
            &mut held,
            &mut short_wait,
        ] {
            queue.put(task).unwrap();
        }
        // Reading the queue succeeds, while scanning the question store fails.
        std::fs::write(questions.root(), "not a directory").unwrap();
        let inventory = Inventory::load(&queue, &questions);
        assert!(!inventory.complete);
        assert!(inventory.ids.contains(&running.id));
        assert_eq!(
            inventory.task(&running.id).unwrap().status,
            TaskStatus::Running
        );
        assert_eq!(
            inventory.waits_on(&waiting),
            ["ca5b (blocked → feb2 running)"]
        );
        assert_eq!(
            inventory.waits_on(&short_wait),
            ["held (unavailable or ambiguous)"]
        );
        assert!(
            inventory.stuck().is_empty(),
            "partial scans cannot prove short-id uniqueness"
        );
        assert!(matches!(inventory.dep("not-seen"), Dep::Unknown));
    }

    #[test]
    fn a_chain_reads_at_a_glance_and_its_root_is_the_held_task() {
        let i = inv(vec![
            t("9db7", TaskStatus::Held, &[]),
            t("4135", TaskStatus::Blocked, &["9db7"]),
            t("6081", TaskStatus::Blocked, &["4135"]),
        ]);
        let six = i.task("2026-6081").unwrap();
        assert_eq!(i.waits_on(six), ["4135 (blocked → 9db7 held)"]);
        let four = i.task("2026-4135").unwrap();
        assert_eq!(i.waits_on(four), ["9db7 (held)"]);
        let stuck = i.stuck();
        assert_eq!(stuck.len(), 2);
        assert!(stuck.values().all(|r| r.iter().eq(["2026-9db7"].iter())));
    }

    #[test]
    fn a_dependency_that_can_still_run_is_not_stuck() {
        let i = inv(vec![
            t("aaaa", TaskStatus::Queued, &[]),
            t("bbbb", TaskStatus::Held, &[]),
            t("cccc", TaskStatus::Blocked, &["aaaa", "bbbb"]),
            t("dddd", TaskStatus::Blocked, &["missing1"]),
            t("eeee", TaskStatus::Blocked, &["cccc"]),
        ]);
        assert!(i.stuck().is_empty(), "{:?}", i.stuck());
    }

    #[test]
    fn a_done_dependency_does_not_hide_a_held_one() {
        let i = inv(vec![
            t("aaaa", TaskStatus::Done, &[]),
            t("bbbb", TaskStatus::Held, &[]),
            t("cccc", TaskStatus::Blocked, &["aaaa", "bbbb"]),
        ]);
        assert_eq!(i.stuck().len(), 1);
    }

    #[test]
    fn a_cycle_terminates_and_names_its_smallest_id_from_every_entry() {
        let i = inv(vec![
            t("bbbb", TaskStatus::Blocked, &["cccc"]),
            t("cccc", TaskStatus::Blocked, &["aaaa"]),
            t("aaaa", TaskStatus::Blocked, &["bbbb"]),
            t("dddd", TaskStatus::Blocked, &["cccc"]),
        ]);
        let stuck = i.stuck();
        assert_eq!(stuck.len(), 4);
        for roots in stuck.values() {
            assert!(roots.iter().eq(["2026-aaaa"].iter()), "{roots:?}");
        }
        let d = i.task("2026-dddd").unwrap();
        assert!(i.waits_on(d)[0].contains("cycle"), "{:?}", i.waits_on(d));
    }

    #[test]
    fn a_self_dependency_is_a_stuck_cycle_of_one() {
        let i = inv(vec![t("aaaa", TaskStatus::Blocked, &["aaaa"])]);
        assert_eq!(i.stuck().len(), 1);
    }

    #[test]
    fn an_open_question_keeps_a_task_alive_and_an_answered_one_resolves() {
        let mut open = Question::new(
            "r".into(),
            "n".into(),
            "s".into(),
            "?".into(),
            String::new(),
            vec![],
        );
        open.id = "2026-qqqq".into();
        let mut answered = open.clone();
        answered.id = "2026-rrrr".into();
        answered.status = QuestionStatus::Answered;
        let tasks = vec![
            t("bbbb", TaskStatus::Held, &[]),
            t("cccc", TaskStatus::Blocked, &["bbbb", "qqqq"]),
            t("dddd", TaskStatus::Blocked, &["bbbb", "rrrr"]),
        ];
        let i = Inventory::new(tasks, &[open, answered]);
        let stuck = i.stuck();
        assert!(!stuck.contains_key("2026-cccc"));
        assert!(stuck.contains_key("2026-dddd"));
        let c = i.task("2026-cccc").unwrap();
        assert_eq!(i.waits_on(c)[1], "qqqq (question open)");
    }
}