dev_prune/adapters/

uv.rs

// Copyright 2026 VKrishna04
// SPDX-License-Identifier: Apache-2.0

// uv package manager adapter for Python projects.

use super::venv::{BASELINE_DISTRIBUTIONS, installed_distributions, normalize_package_name};
use super::{
    BloatDir, EnforcePolicy, PackageManager, dir_size, enforce_two_tier, run_command_with_timeout,
};
use crate::declared::{Gap, RebuildCheck, label_of, on_path, path_of, relative_parts};
use anyhow::{Result, anyhow};
use std::collections::HashSet;
use std::fs;
use std::path::Path;

/// Adapter for uv-based Python projects.
pub struct Uv;

/// Whether `.venv` was created by uv itself — uv stamps a `uv = <version>` key into
/// `pyvenv.cfg`. A venv without the stamp was built by some other tool, and uv can make
/// no claims about what is inside it.
fn venv_is_uv_managed(path: &Path) -> bool {
    fs::read_to_string(path.join(".venv").join("pyvenv.cfg"))
        .map(|content| {
            content.lines().any(|line| {
                line.trim_start()
                    .strip_prefix("uv")
                    .is_some_and(|rest| rest.trim_start().starts_with('='))
            })
        })
        .unwrap_or(false)
}

/// Every package name `uv.lock` pins, normalised.
///
/// `uv.lock` records the full transitive closure — including the project itself — so a
/// plain name scan is enough; no dependency graph needed. `None` when no names could be
/// read at all, in which case the caller skips the drift comparison rather than refusing
/// on a lockfile format this scan does not understand.
pub(super) fn lockfile_package_names(lockfile: &Path) -> Option<HashSet<String>> {
    let content = fs::read_to_string(lockfile).ok()?;
    let mut names = HashSet::new();
    let mut in_package = false;
    for raw in content.lines() {
        let line = raw.trim();
        if line.starts_with('[') {
            in_package = line == "[[package]]";
            continue;
        }
        if !in_package {
            continue;
        }
        if let Some(rest) = line.strip_prefix("name")
            && let Some(value) = rest.trim_start().strip_prefix('=')
        {
            let value = value.trim().trim_matches('"');
            if !value.is_empty() {
                names.insert(normalize_package_name(value));
            }
            // Only the first `name` in each `[[package]]` entry is the package's own.
            in_package = false;
        }
    }
    (!names.is_empty()).then_some(names)
}

/// Installed distributions in `.venv` that `uv.lock` does not record.
///
/// Those were installed ad hoc (`uv pip install …`) and `uv sync` after deletion would
/// not bring them back — exactly what this tool promises never to lose.
pub(super) fn unlocked_packages(path: &Path, locked: &HashSet<String>) -> Vec<String> {
    let Some(installed) = installed_distributions(&path.join(".venv")) else {
        return Vec::new();
    };
    let mut extras: Vec<String> = installed
        .keys()
        .filter(|name| !locked.contains(*name) && !BASELINE_DISTRIBUTIONS.contains(&name.as_str()))
        .cloned()
        .collect();
    extras.sort();
    extras
}

impl PackageManager for Uv {
    fn name(&self) -> &'static str {
        "uv"
    }

    fn detect(&self, path: &Path) -> bool {
        let uv_lock = path.join("uv.lock");
        if uv_lock.exists() {
            return true;
        }

        let pyproject = path.join("pyproject.toml");
        if pyproject.exists()
            && let Ok(content) = fs::read_to_string(&pyproject)
            && content.contains("[tool.uv]")
        {
            return true;
        }

        false
    }

    fn bloat_dirs(&self, path: &Path) -> Vec<BloatDir> {
        let mut dirs = Vec::new();
        let venv_path = path.join(".venv");
        if venv_path.exists() {
            dirs.push(BloatDir {
                name: ".venv".to_string(),
                path: venv_path.clone(),
                size_bytes: dir_size(&venv_path),
                shared_bytes: 0,
            });
        }
        dirs
    }

    /// Enforces the lockfile without writing it.
    ///
    /// `uv lock --locked` asserts that `uv.lock` is already up to date with
    /// `pyproject.toml` and exits non-zero instead of rewriting it when it is not.
    /// Plain `uv lock` is the writing form, for the case where no lockfile exists yet.
    fn enforce_lockfile(&self, path: &Path, policy: EnforcePolicy) -> Result<()> {
        let lockfile = path.join("uv.lock");

        // Generating a lockfile from `pyproject.toml` only proves the *declared*
        // dependencies resolve — it says nothing about what is actually installed in
        // `.venv`. When the venv was not even created by uv, `uv sync` against that
        // fresh lock could rebuild a different environment than the one deleted, so
        // refuse instead of manufacturing proof.
        if !lockfile.exists() && path.join(".venv").exists() && !venv_is_uv_managed(path) {
            return Err(anyhow!(
                "`pyproject.toml` declares `[tool.uv]` but there is no `uv.lock`, and \
                 `.venv` was not created by uv — a generated lockfile could not prove the \
                 environment's contents are recoverable. Rebuild the environment under uv \
                 first: `uv lock` then `uv sync`."
            ));
        }

        enforce_two_tier(
            &lockfile,
            "uv",
            &["lock", "--locked"],
            &["lock"],
            path,
            policy,
        )?;

        // The environment can hold packages the lockfile never recorded — a
        // `uv pip install foo` nobody wrote back. `uv.lock` pins the full transitive
        // closure, so anything installed but absent from it is recoverable from
        // nowhere, which is exactly what this tool promises never to delete.
        if let Some(locked) = lockfile_package_names(&lockfile) {
            let extras = unlocked_packages(path, &locked);
            if !extras.is_empty() {
                let shown = extras
                    .iter()
                    .take(10)
                    .cloned()
                    .collect::<Vec<_>>()
                    .join(", ");
                let suffix = if extras.len() > 10 {
                    format!(", … and {} more", extras.len() - 10)
                } else {
                    String::new()
                };
                return Err(anyhow!(
                    "`.venv` holds {} package(s) that uv.lock does not record \
                     ({shown}{suffix}). They were installed ad hoc and `uv sync` would \
                     not bring them back. Record them first: `uv add <package>`.",
                    extras.len()
                ));
            }
        }
        Ok(())
    }

    fn restore(&self, path: &Path, timeout: std::time::Duration) -> Result<()> {
        self.restore_named(path, ".venv", None, timeout)
    }

    /// `uv sync --python 3.12` rebuilds on that interpreter and, unlike every other
    /// route here, *downloads* it when the machine does not have it — which is why the
    /// tag is passed straight through without an availability check first.
    fn restore_named(
        &self,
        path: &Path,
        _dir_name: &str,
        runtime: Option<&str>,
        timeout: std::time::Duration,
    ) -> Result<()> {
        match runtime.filter(|tag| super::is_valid_runtime_tag(tag)) {
            Some(tag) => run_command_with_timeout("uv", &["sync", "--python", tag], path, timeout),
            None => run_command_with_timeout("uv", &["sync"], path, timeout),
        }
    }

    /// The interpreter `.venv` was built with, so a restore can rebuild on it.
    fn runtime_tag(&self, path: &Path, dir_name: &str) -> Option<String> {
        super::venv_runtime_tag(&path.join(dir_name))
    }

    fn lockfiles(&self) -> &'static [&'static str] {
        &["uv.lock"]
    }

    /// The comparison `enforce_lockfile` refuses on, as data: distributions in `.venv`
    /// that `uv.lock` does not pin.
    fn drift(&self, path: &Path) -> Vec<super::DriftReport> {
        let Some(locked) = lockfile_package_names(&path.join("uv.lock")) else {
            return Vec::new();
        };
        let extras = unlocked_packages(path, &locked);
        if extras.is_empty() {
            return Vec::new();
        }
        vec![super::DriftReport {
            directory: ".venv".to_string(),
            unrecorded: extras,
            record_command: "uv add <package>",
        }]
    }
}

/// The rebuild check for `uv` declarations.
pub(crate) struct UvScripts;

impl RebuildCheck for UvScripts {
    fn tools(&self) -> &'static [&'static str] {
        &["uv"]
    }

    fn gap(&self, repo_path: &Path, _tool: &str, args: &[&str]) -> Option<Gap> {
        uv_script_gap(repo_path, args)
    }
}

/// What `pyproject.toml` had to say about the name `uv run` was given.
struct PyprojectLookup {
    /// `[project.scripts]` — the one table uv resolves entry points from.
    defined: bool,
    /// The name appears as a requirement, so a dependency's console script provides it.
    from_dependency: bool,
    /// `[tool.uv.scripts]` names it. That table is not one uv reads.
    in_tool_uv_scripts: bool,
}

/// A `uv run` target that `pyproject.toml` does not provide.
fn uv_script_gap(repo_path: &Path, args: &[&str]) -> Option<Gap> {
    let mut dir = None;
    let mut saw_run = false;
    let mut script = None;
    let mut i = 0;
    while i < args.len() {
        let arg = args[i];
        if arg == "--" {
            return None;
        }
        if let Some(value) = arg
            .strip_prefix("--directory=")
            .or_else(|| arg.strip_prefix("--project="))
        {
            dir = Some(value.to_string());
        } else if arg == "--directory" || arg == "--project" {
            dir = Some((*args.get(i + 1)?).to_string());
            i += 1;
        } else if arg == "run" {
            saw_run = true;
        } else if arg.starts_with('-') {
            return None;
        } else if saw_run {
            script = Some(arg.to_string());
            break;
        } else {
            // `uv sync`, `uv pip install …` — not a script invocation at all.
            return None;
        }
        i += 1;
    }
    let script = script?;
    // `uv run ./tools/gen.py` runs a file. Whether that file is there is not a question
    // `pyproject.toml` answers.
    if script.contains(['/', '\\']) || script.ends_with(".py") {
        return None;
    }
    // `uv run` also falls back to anything already on `PATH`.
    if on_path(&script) {
        return None;
    }
    let parts = relative_parts(dir.as_deref())?;
    let manifest = label_of(&parts, "pyproject.toml");
    let content = fs::read_to_string(path_of(repo_path, &parts, "pyproject.toml")).ok()?;
    let found = pyproject_lookup(&content, &script)?;
    if found.defined || found.from_dependency {
        return None;
    }
    // A dependency's console script is not named in `pyproject.toml` at all when the
    // requirement is only pinned in the lockfile.
    if lockfile_records(&path_of(repo_path, &parts, "uv.lock"), &script) {
        return None;
    }
    let fix = if found.in_tool_uv_scripts {
        format!(
            "`[tool.uv.scripts]` is not a table uv reads — it is silently ignored. Move \
             `{script}` to `[project.scripts]` in `{manifest}`, or fix the command."
        )
    } else {
        format!("Add `{script}` to `[project.scripts]` in `{manifest}`, or fix the command.")
    };
    Some(Gap {
        what: format!("`{manifest}` defines no `{script}` entry point"),
        fix,
    })
}

/// Where `pyproject.toml` does and does not mention one name.
///
/// Line-based for the same reason the poetry adapter's read of this file is: there is no
/// TOML dependency in this crate, and table headers only ever start a line. Any shape
/// the scan cannot see through returns `None`, which allows the prune.
fn pyproject_lookup(content: &str, wanted: &str) -> Option<PyprojectLookup> {
    let mut found = PyprojectLookup {
        defined: false,
        from_dependency: false,
        in_tool_uv_scripts: false,
    };
    let mut table = "";
    for raw in content.lines() {
        let line = raw.trim();
        if line.is_empty() || line.starts_with('#') {
            continue;
        }
        if line.starts_with('[') {
            table = line;
            continue;
        }
        // A dotted key puts entry points somewhere this scan does not look.
        if line.starts_with("project.scripts") {
            return None;
        }
        // Any quoted requirement anywhere in the file — `"pytest>=8"` in a dependency
        // array, wherever that array happens to be written. `uv run pytest` runs a
        // console script that a dependency installed, and no table here lists it.
        if quoted_strings(line).any(|value| same_name(requirement_head(value), wanted)) {
            found.from_dependency = true;
        }
        let Some((key, _)) = line.split_once('=') else {
            continue;
        };
        let key = key.trim().trim_matches(['"', '\'']);
        match table {
            // An inline `scripts = { regen = "…" }` is a shape this cannot read.
            "[project]" if key == "scripts" => return None,
            "[project.scripts]" | "[project.gui-scripts]" if same_name(key, wanted) => {
                found.defined = true;
            }
            "[tool.uv.scripts]" if same_name(key, wanted) => found.in_tool_uv_scripts = true,
            _ => {}
        }
    }
    Some(found)
}

/// The double-quoted runs of a line.
fn quoted_strings(line: &str) -> impl Iterator<Item = &str> {
    line.split('"').skip(1).step_by(2)
}

/// The distribution name at the front of a requirement string like `pytest>=8,<9`.
fn requirement_head(raw: &str) -> &str {
    raw.trim()
        .split(|c: char| !(c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.'))
        .next()
        .unwrap_or("")
}

/// Python treats `-` and `_` in a distribution or entry-point name as the same character.
fn same_name(a: &str, b: &str) -> bool {
    a.replace('_', "-")
        .eq_ignore_ascii_case(&b.replace('_', "-"))
}

/// Does a `uv.lock` record a package under this name?
fn lockfile_records(lockfile: &Path, wanted: &str) -> bool {
    let Ok(content) = fs::read_to_string(lockfile) else {
        return false;
    };
    content.lines().any(|raw| {
        raw.trim()
            .strip_prefix("name")
            .and_then(|rest| rest.trim_start().strip_prefix('='))
            .is_some_and(|value| same_name(value.trim().trim_matches('"'), wanted))
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs::File;
    use std::io::Write;
    use tempfile::tempdir;

    #[test]
    fn test_name() {
        let adapter = Uv;
        assert_eq!(adapter.name(), "uv");
    }

    #[test]
    fn test_detect_positive_lock() {
        let dir = tempdir().unwrap();
        File::create(dir.path().join("uv.lock")).unwrap();

        let adapter = Uv;
        assert!(adapter.detect(dir.path()));
    }

    #[test]
    fn test_detect_positive_toml() {
        let dir = tempdir().unwrap();
        let mut file = File::create(dir.path().join("pyproject.toml")).unwrap();
        writeln!(file, "[tool.uv]").unwrap();

        let adapter = Uv;
        assert!(adapter.detect(dir.path()));
    }

    #[test]
    fn test_detect_negative() {
        let dir = tempdir().unwrap();

        let adapter = Uv;
        assert!(!adapter.detect(dir.path()));
    }

    #[test]
    fn test_bloat_dirs_present() {
        let dir = tempdir().unwrap();
        fs::create_dir(dir.path().join(".venv")).unwrap();

        let adapter = Uv;
        let dirs = adapter.bloat_dirs(dir.path());
        assert_eq!(dirs.len(), 1);
        assert_eq!(dirs[0].name, ".venv");
    }

    #[test]
    fn test_bloat_dirs_absent() {
        let dir = tempdir().unwrap();

        let adapter = Uv;
        let dirs = adapter.bloat_dirs(dir.path());
        assert!(dirs.is_empty());
    }

    #[test]
    fn a_uv_stamped_pyvenv_cfg_marks_the_venv_as_uv_managed() {
        let dir = tempdir().unwrap();
        let venv = dir.path().join(".venv");
        fs::create_dir(&venv).unwrap();
        let mut cfg = File::create(venv.join("pyvenv.cfg")).unwrap();
        writeln!(cfg, "home = /usr/bin").unwrap();
        writeln!(cfg, "uv = 0.5.9").unwrap();
        assert!(venv_is_uv_managed(dir.path()));
    }

    #[test]
    fn a_pip_built_venv_is_not_mistaken_for_a_uv_one() {
        let dir = tempdir().unwrap();
        let venv = dir.path().join(".venv");
        fs::create_dir(&venv).unwrap();
        // `uvloop` starts with "uv" but is not the uv stamp.
        let mut cfg = File::create(venv.join("pyvenv.cfg")).unwrap();
        writeln!(cfg, "home = /usr/bin").unwrap();
        writeln!(cfg, "uvloop = 1.0").unwrap();
        assert!(!venv_is_uv_managed(dir.path()));
    }

    #[test]
    fn lockfile_names_come_from_package_entries_only() {
        let dir = tempdir().unwrap();
        let lock = dir.path().join("uv.lock");
        fs::write(
            &lock,
            "version = 1\n\n[[package]]\nname = \"Requests\"\nversion = \"2.32.3\"\n\n\
             [package.metadata]\nname = \"not-a-package\"\n\n[[package]]\nname = \"my-proj\"\n",
        )
        .unwrap();
        let names = lockfile_package_names(&lock).unwrap();
        assert!(names.contains("requests"));
        assert!(names.contains("my-proj"));
        assert!(!names.contains("not-a-package"));
        assert_eq!(names.len(), 2);
    }

    #[test]
    fn an_ad_hoc_install_missing_from_the_lockfile_is_flagged() {
        let dir = tempdir().unwrap();
        let sp = dir.path().join(".venv").join("Lib").join("site-packages");
        fs::create_dir_all(sp.join("requests-2.32.3.dist-info")).unwrap();
        fs::create_dir_all(sp.join("sneaky_pkg-1.0.dist-info")).unwrap();
        fs::create_dir_all(sp.join("pip-24.0.dist-info")).unwrap();

        let locked: HashSet<String> = ["requests".to_string()].into();
        assert_eq!(unlocked_packages(dir.path(), &locked), vec!["sneaky-pkg"]);
    }

    #[test]
    fn a_foreign_venv_next_to_tool_uv_without_a_lock_is_refused() {
        let dir = tempdir().unwrap();
        let mut file = File::create(dir.path().join("pyproject.toml")).unwrap();
        writeln!(file, "[tool.uv]").unwrap();
        let venv = dir.path().join(".venv");
        fs::create_dir(&venv).unwrap();
        File::create(venv.join("pyvenv.cfg")).unwrap();

        let err = Uv
            .enforce_lockfile(dir.path(), EnforcePolicy::default())
            .unwrap_err();
        assert!(err.to_string().contains("not created by uv"));
    }

    #[test]
    fn drift_reports_the_ad_hoc_install_as_data() {
        let dir = tempdir().unwrap();
        fs::write(
            dir.path().join("uv.lock"),
            "[[package]]\nname = \"requests\"\nversion = \"2.32.3\"\n",
        )
        .unwrap();
        let sp = dir.path().join(".venv").join("Lib").join("site-packages");
        fs::create_dir_all(sp.join("requests-2.32.3.dist-info")).unwrap();
        fs::create_dir_all(sp.join("sneaky_pkg-1.0.dist-info")).unwrap();

        let reports = Uv.drift(dir.path());
        assert_eq!(reports.len(), 1);
        assert_eq!(reports[0].directory, ".venv");
        assert_eq!(reports[0].unrecorded, vec!["sneaky-pkg"]);
        assert_eq!(reports[0].record_command, "uv add <package>");
    }

    #[test]
    fn drift_is_silent_when_the_lockfile_records_everything() {
        let dir = tempdir().unwrap();
        fs::write(
            dir.path().join("uv.lock"),
            "[[package]]\nname = \"requests\"\nversion = \"2.32.3\"\n",
        )
        .unwrap();
        let sp = dir.path().join(".venv").join("Lib").join("site-packages");
        fs::create_dir_all(sp.join("requests-2.32.3.dist-info")).unwrap();

        assert!(Uv.drift(dir.path()).is_empty());
    }
}