liboxen 0.46.12

//! # Commits
//!
//! Create, read, and list commits
//!

use crate::core::versions::MinOxenVersion;
use crate::error::OxenError;
use crate::model::User;
use crate::model::{Commit, LocalRepository, MerkleHash};
use crate::opts::PaginateOpts;
use crate::util;
use crate::view::{PaginatedCommits, StatusMessage};
use crate::{core, resource};

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};

pub mod commit_writer;

/// # Commit the staged files in the repo
///
/// ```ignore
/// use liboxen::repositories;
/// use liboxen::util;
///
/// // Initialize the repository
/// let base_dir = Path::new("repo_dir_commit");
/// let repo = repositories::init(base_dir)?;
///
/// // Write file to disk
/// let hello_file = base_dir.join("hello.txt");
/// util::fs::write_to_path(&hello_file, "Hello World");
///
/// // Stage the file
/// repositories::add(&repo, &hello_file).await?;
///
/// // Commit staged
/// repositories::commit(&repo, "My commit message")?;
/// ```
pub fn commit(repo: &LocalRepository, message: &str) -> Result<Commit, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::commit(repo, message),
    }
}

pub fn commit_with_user(
    repo: &LocalRepository,
    message: &str,
    user: &User,
) -> Result<Commit, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::commit_with_user(repo, message, user),
    }
}

/// # Commit with --allow-empty flag
///
/// Allows creating a commit even when there are no staged changes.
/// This reuses the existing create_empty_commit infrastructure.
pub fn commit_allow_empty(repo: &LocalRepository, message: &str) -> Result<Commit, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::commit_allow_empty(repo, message),
    }
}

/// Iterate over all commits and get the one with the latest timestamp
pub fn latest_commit(repo: &LocalRepository) -> Result<Commit, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::latest_commit(repo),
    }
}

/// The current HEAD commit of the branch you currently have checked out
pub fn head_commit(repo: &LocalRepository) -> Result<Commit, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::head_commit(repo),
    }
}

/// Maybe get the head commit if it exists
/// Returns None if the head commit does not exist (empty repo)
pub fn head_commit_maybe(repo: &LocalRepository) -> Result<Option<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::head_commit_maybe(repo),
    }
}

/// Get the root commit of a repository
pub fn root_commit_maybe(repo: &LocalRepository) -> Result<Option<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::root_commit_maybe(repo),
    }
}

/// Get a commit by it's MerkleHash
pub fn get_by_hash(repo: &LocalRepository, hash: &MerkleHash) -> Result<Option<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::get_by_hash(repo, hash),
    }
}

/// Get a commit by it's string hash
pub fn get_by_id(
    repo: &LocalRepository,
    commit_id: impl AsRef<str>,
) -> Result<Option<Commit>, OxenError> {
    let commit_id = commit_id.as_ref();
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::get_by_id(repo, commit_id),
    }
}

/// Commit id exists
pub fn commit_id_exists(
    repo: &LocalRepository,
    commit_id: impl AsRef<str>,
) -> Result<bool, OxenError> {
    get_by_id(repo, commit_id.as_ref()).map(|commit| commit.is_some())
}

/// Create an empty commit off of the head commit of a branch
pub fn create_empty_commit(
    repo: &LocalRepository,
    branch_name: impl AsRef<str>,
    commit: &Commit,
) -> Result<Commit, OxenError> {
    let branch_name = branch_name.as_ref();
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("create_empty_commit not supported in v0.10.0"),
        _ => core::v_latest::commits::create_empty_commit(repo, branch_name, commit),
    }
}

/// Create an initial empty commit for an empty repository.
/// This creates the first commit with an empty tree and sets up the branch.
/// Returns an error if the repository already has commits.
pub fn create_initial_commit(
    repo: &LocalRepository,
    branch_name: impl AsRef<str>,
    user: &User,
    message: impl AsRef<str>,
) -> Result<Commit, OxenError> {
    let branch_name = branch_name.as_ref();
    let message = message.as_ref();
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("create_initial_commit not supported in v0.10.0"),
        _ => core::v_latest::commits::create_initial_commit(repo, branch_name, user, message),
    }
}

/// List commits on the current branch from HEAD
pub fn list(repo: &LocalRepository) -> Result<Vec<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::list(repo),
    }
}

/// List commits for the repository in no particular order
pub fn list_all(repo: &LocalRepository) -> Result<HashSet<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::list_all(repo),
    }
}

/// List unsynced commits for the repository (ie they are missing their .version/ files)
pub fn list_unsynced(repo: &LocalRepository) -> Result<HashSet<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("list_unsynced not supported in v0.10.0"),
        _ => core::v_latest::commits::list_unsynced(repo),
    }
}

/// List unsynced commits from a specific revision
pub fn list_unsynced_from(
    repo: &LocalRepository,
    revision: impl AsRef<str>,
) -> Result<HashSet<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("list_unsynced_from not supported in v0.10.0"),
        _ => core::v_latest::commits::list_unsynced_from(repo, revision),
    }
}
// Source
pub fn get_commit_or_head<S: AsRef<str> + Clone>(
    repo: &LocalRepository,
    commit_id_or_branch_name: Option<S>,
) -> Result<Commit, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => resource::get_commit_or_head(repo, commit_id_or_branch_name),
        _ => core::v_latest::commits::get_commit_or_head(repo, commit_id_or_branch_name),
    }
}

pub fn list_all_paginated(
    repo: &LocalRepository,
    pagination: PaginateOpts,
) -> Result<PaginatedCommits, OxenError> {
    log::info!("list_all_paginated: {:?} {:?}", repo.path, pagination);
    let commits = list_all(repo)?;
    let commits: Vec<Commit> = commits.into_iter().collect();
    let (commits, pagination) = util::paginate(commits, pagination.page_num, pagination.page_size);
    Ok(PaginatedCommits {
        status: StatusMessage::resource_found(),
        commits,
        pagination,
    })
}

/// List the history for a specific branch or commit (revision)
pub fn list_from(repo: &LocalRepository, revision: &str) -> Result<Vec<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::list_from(repo, revision),
    }
}

pub fn list_from_with_depth(
    repo: &LocalRepository,
    revision: &str,
) -> Result<HashMap<Commit, usize>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => Err(OxenError::basic_str(
            "list_from_with_depth not supported in v0.10.0",
        )),
        _ => core::v_latest::commits::list_from_with_depth(repo, revision),
    }
}

/// List the history between two commits
pub fn list_between(
    repo: &LocalRepository,
    base: &Commit,
    head: &Commit,
) -> Result<Vec<Commit>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::list_between(repo, base, head),
    }
}

/// Get a list of commits by the commit message
pub fn get_by_message(
    repo: &LocalRepository,
    msg: impl AsRef<str>,
) -> Result<Vec<Commit>, OxenError> {
    let commits = list_all(repo)?;
    let filtered: Vec<Commit> = commits
        .into_iter()
        .filter(|commit| commit.message == msg.as_ref())
        .collect();
    Ok(filtered)
}

/// Get the most recent commit by the commit message, starting at the HEAD commit
pub fn first_by_message(
    repo: &LocalRepository,
    msg: impl AsRef<str>,
) -> Result<Option<Commit>, OxenError> {
    let commits = list(repo)?;
    Ok(commits
        .into_iter()
        .find(|commit| commit.message == msg.as_ref()))
}

/// Retrieve entries with filepaths matching a provided glob pattern
pub fn search_entries(
    repo: &LocalRepository,
    commit: &Commit,
    pattern: &str,
) -> Result<HashSet<PathBuf>, OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::search_entries(repo, commit, pattern),
    }
}

/// List paginated commits starting from the given revision
pub fn list_from_paginated(
    repo: &LocalRepository,
    revision: &str,
    pagination: PaginateOpts,
) -> Result<PaginatedCommits, OxenError> {
    let _perf = crate::perf_guard!("commits::list_from_paginated");

    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => {
            // Calculate skip and limit based on pagination parameters
            let skip = if pagination.page_num == 0 {
                0
            } else {
                (pagination.page_num - 1) * pagination.page_size
            };
            let limit = pagination.page_size;

            let _perf_list = crate::perf_guard!("commits::list_from_paginated_optimized");
            let (commits, total_entries, cached) =
                core::v_latest::commits::list_from_paginated_impl(repo, revision, skip, limit)?;
            log::info!(
                "list_from_paginated {} got {} commits out of {} total (cached: {})",
                revision,
                commits.len(),
                total_entries,
                cached
            );
            drop(_perf_list);

            // Calculate pagination metadata
            let total_pages = if pagination.page_size > 0 {
                (total_entries as f64 / pagination.page_size as f64).ceil() as usize
            } else {
                0
            };

            let pagination = crate::view::Pagination {
                page_size: pagination.page_size,
                page_number: pagination.page_num,
                total_pages,
                total_entries,
            };

            Ok(PaginatedCommits {
                status: StatusMessage::resource_found(),
                commits,
                pagination,
            })
        }
    }
}

/// List paginated commits by resource
pub fn list_by_path_from_paginated(
    repo: &LocalRepository,
    commit: &Commit,
    path: &Path,
    pagination: PaginateOpts,
) -> Result<PaginatedCommits, OxenError> {
    let _perf = crate::perf_guard!("commits::list_by_path_from_paginated");

    log::info!("list_by_path_from_paginated: {commit:?} {path:?}");
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => panic!("v0.10.0 no longer supported"),
        _ => core::v_latest::commits::list_by_path_from_paginated(repo, commit, path, pagination),
    }
}

pub fn count_from(
    repo: &LocalRepository,
    revision: impl AsRef<str>,
) -> Result<(usize, bool), OxenError> {
    match repo.min_version() {
        MinOxenVersion::V0_10_0 => Err(OxenError::basic_str("count_from not supported in v0.10.0")),
        _ => core::v_latest::commits::count_from(repo, revision),
    }
}

pub fn commit_history_is_complete(
    repo: &LocalRepository,
    commit: &Commit,
) -> Result<bool, OxenError> {
    // Get full commit history from this head backwards
    let history = list_from(repo, &commit.id)?;

    // Ensure traces back to base commit
    let maybe_initial_commit = history.last().unwrap();
    if !maybe_initial_commit.parent_ids.is_empty() {
        // If it has parents, it isn't an initial commit
        log::debug!(
            "commit_history_is_complete ❌ last commit has parents: {maybe_initial_commit}"
        );
        return Ok(false);
    }

    // Ensure all commits and their parents are synced
    // Initialize commit reader
    for c in &history {
        log::debug!("commit_history_is_complete checking if commit is synced: {c}");

        if !core::commit_sync_status::commit_is_synced(repo, &c.id.parse()?) {
            log::debug!("commit_history_is_complete ❌ commit is not synced: {c}");
            return Ok(false);
        } else {
            log::debug!("commit_history_is_complete ✅ commit is synced: {c}");
        }
    }
    Ok(true)
}

#[cfg(test)]
mod tests {
    use crate::test;
    use std::path::Path;

    use crate::error::OxenError;
    use crate::model::EntryDataType;
    use crate::model::StagedEntryStatus;
    use crate::opts::CloneOpts;
    use crate::opts::RmOpts;
    use crate::repositories;

    use crate::util;

    use super::*;

    #[tokio::test]
    async fn test_command_commit_file() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Write to file
            let hello_file = repo.path.join("hello.txt");
            util::fs::write_to_path(&hello_file, "Hello World")?;

            // Track the file
            repositories::add(&repo, &hello_file).await?;
            // Commit the file
            let commit = repositories::commit(&repo, "My message")?;
            assert_eq!(commit.message, "My message");

            // Get status and make sure it is removed from the untracked and added
            let repo_status = repositories::status(&repo)?;
            assert_eq!(repo_status.staged_dirs.len(), 0);
            assert_eq!(repo_status.staged_files.len(), 0);
            assert_eq!(repo_status.untracked_files.len(), 0);
            assert_eq!(repo_status.untracked_dirs.len(), 0);

            let commits = repositories::commits::list(&repo)?;
            assert_eq!(commits.len(), 1);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_removed_file() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Write to file
            let hello_file = repo.path.join("hello.txt");
            util::fs::write_to_path(&hello_file, "Hello World")?;

            // Track the file
            repositories::add(&repo, &hello_file).await?;

            // Remove the file
            util::fs::remove_file(&hello_file)?;

            // Can still commit the file, since it is in the versions directory
            repositories::commit(&repo, "My message")?;

            // Get status and make sure the file was not committed
            let head = repositories::commits::head_commit(&repo)?;
            let commit_list = repositories::entries::list_for_commit(&repo, &head)?;
            assert_eq!(commit_list.len(), 1);

            // Add the removed file and commit
            repositories::add(&repo, &hello_file).await?;
            repositories::commit(&repo, "Second Message")?;

            // We should now have no entries
            let head = repositories::commits::head_commit(&repo)?;
            let commit_list = repositories::entries::list_for_commit(&repo, &head)?;
            assert_eq!(commit_list.len(), 0);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_command_commit_train_data_dir() -> Result<(), OxenError> {
        test::run_training_data_repo_test_no_commits_async(|repo| async move {
            // Track the file
            let train_dir = repo.path.join("train");
            repositories::add(&repo, train_dir).await?;
            // Commit the file
            let commit = repositories::commit(&repo, "Adding training data")?;

            let repo_status = repositories::status(&repo)?;
            repo_status.print();
            assert_eq!(repo_status.staged_dirs.len(), 0);
            assert_eq!(repo_status.staged_files.len(), 0);
            assert_eq!(repo_status.untracked_files.len(), 4);
            assert_eq!(repo_status.untracked_dirs.len(), 4);

            repositories::tree::print_tree(&repo, &commit)?;

            let dir_node =
                repositories::tree::get_node_by_path(&repo, &commit, PathBuf::from("train"))?;
            assert!(dir_node.is_some());

            let commits = repositories::commits::list(&repo)?;
            assert_eq!(commits.len(), 1);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_command_commit_dir_recursive() -> Result<(), OxenError> {
        test::run_training_data_repo_test_no_commits_async(|repo| async move {
            // Track the annotations dir, which has sub dirs
            let annotations_dir = repo.path.join("annotations");
            repositories::add(&repo, annotations_dir).await?;
            repositories::commit(&repo, "Adding annotations data dir, which has two levels")?;

            let repo_status = repositories::status(&repo)?;
            repo_status.print();

            assert_eq!(repo_status.staged_dirs.len(), 0);
            assert_eq!(repo_status.staged_files.len(), 0);
            assert_eq!(repo_status.untracked_files.len(), 4);
            assert_eq!(repo_status.untracked_dirs.len(), 4);

            let commits = repositories::commits::list(&repo)?;
            assert_eq!(commits.len(), 1);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_command_commit_second_level_dir_then_revert() -> Result<(), OxenError> {
        test::run_select_data_repo_test_no_commits_async("annotations", |repo| async move {
            // Track & commit (dir already created in helper)
            let new_dir_path = repo.path.join("annotations").join("train");
            repositories::add(&repo, &new_dir_path).await?;
            repositories::commit(&repo, "Adding train dir")?;

            // Get the original branch name
            let orig_branch = repositories::branches::current_branch(&repo)?.unwrap();

            // Create a branch to make the changes
            let branch_name = "feature/adding-annotations";
            repositories::branches::create_checkout(&repo, branch_name)?;

            // Track & commit (dir already created in helper)
            let test_dir_path = repo.path.join("annotations").join("test");
            let og_num_files = util::fs::rcount_files_in_dir(&test_dir_path);

            repositories::add(&repo, &test_dir_path).await?;
            repositories::commit(&repo, "Adding test dir")?;

            // checkout OG and make sure it removes the train dir
            repositories::checkout(&repo, orig_branch.name).await?;
            assert!(!test_dir_path.exists());

            // checkout branch again and make sure it reverts
            repositories::checkout(&repo, branch_name).await?;
            assert!(test_dir_path.exists());
            assert_eq!(util::fs::rcount_files_in_dir(&test_dir_path), og_num_files);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_command_commit_removed_dir() -> Result<(), OxenError> {
        test::run_training_data_repo_test_no_commits_async(|repo| async move {
            // (dir already created in helper)
            let dir_to_remove = repo.path.join("train");
            let og_file_count = util::fs::rcount_files_in_dir(&dir_to_remove);

            repositories::add(&repo, &dir_to_remove).await?;
            repositories::commit(&repo, "Adding train directory")?;

            // Delete the directory
            util::fs::remove_dir_all(&dir_to_remove)?;

            // Add the deleted dir, so that we can commit the deletion
            repositories::add(&repo, &dir_to_remove).await?;

            // Make sure we have the correct amount of files tagged as removed
            let status = repositories::status(&repo)?;
            assert_eq!(status.staged_files.len(), og_file_count);
            assert_eq!(
                status.staged_files.iter().next().unwrap().1.status,
                StagedEntryStatus::Removed
            );

            status.print();

            // Make sure they don't show up in the status
            assert_eq!(status.removed_files.len(), 0);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_after_merge_conflict() -> Result<(), OxenError> {
        test::run_select_data_repo_test_no_commits_async("labels", |repo| async move {
            let labels_path = repo.path.join("labels.txt");
            repositories::add(&repo, &labels_path).await?;
            repositories::commit(&repo, "adding initial labels file")?;

            let og_branch = repositories::branches::current_branch(&repo)?.unwrap();

            // Add a "none" category on a branch
            let branch_name = "change-labels";
            repositories::branches::create_checkout(&repo, branch_name)?;

            test::modify_txt_file(&labels_path, "cat\ndog\nnone")?;
            repositories::add(&repo, &labels_path).await?;
            repositories::commit(&repo, "adding none category")?;

            // Add a "person" category on a the main branch
            repositories::checkout(&repo, og_branch.name).await?;

            test::modify_txt_file(&labels_path, "cat\ndog\nperson")?;
            repositories::add(&repo, &labels_path).await?;
            repositories::commit(&repo, "adding person category")?;

            // Try to merge in the changes
            repositories::merge::merge(&repo, branch_name).await?;

            // We should have a conflict
            let status = repositories::status(&repo)?;
            assert_eq!(status.merge_conflicts.len(), 1);

            // Assume that we fixed the conflict and added the file
            let path = status.merge_conflicts[0].base_entry.path.clone();
            let fullpath = repo.path.join(path);
            repositories::add(&repo, fullpath).await?;

            // Should commit, and then see full commit history
            repositories::commit(&repo, "merging into main")?;

            // Should have commits:
            //  1) add labels
            //  2) change-labels branch modification
            //  3) main branch modification
            //  4) merge commit
            let history = repositories::commits::list(&repo)?;
            assert_eq!(history.len(), 4);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_with_no_staged_changes() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Add a text file
            let text_path = repo.path.join("text.txt");
            util::fs::write_to_path(&text_path, "Hello World")?;

            // Get the hash of the file at this timestamp
            repositories::add(&repo, &text_path).await?;
            repositories::commit(&repo, "Committing hello world")?;

            // Modify the text file
            util::fs::write_to_path(&text_path, "Goodbye, world!")?;

            let status = repositories::status(&repo)?;
            status.print();

            // There should be nothing to commit since the file is untracked
            let commit_result = repositories::commit(&repo, "Committing goodbye world");
            assert!(commit_result.is_err());

            // Make sure the entry is still there
            let head = repositories::commits::head_commit(&repo)?;
            let tree = repositories::tree::get_root_with_children(&repo, &head)?.unwrap();
            let text_entry = tree.get_by_path(Path::new("text.txt"))?;
            assert!(text_entry.is_some());

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_hash_on_modified_file() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Add a text file
            let text_path = repo.path.join("text.txt");
            util::fs::write_to_path(&text_path, "Hello World")?;

            // Get the hash of the file at this timestamp
            let hash_when_add =
                util::hasher::hash_file_contents(&text_path)?.parse::<MerkleHash>()?;
            repositories::add(&repo, &text_path).await?;

            let status = repositories::status(&repo)?;
            status.print();

            // Note v10 did not have this line, and we didn't copy to the versions dir on add
            repositories::commit(&repo, "Committing hello world")?;

            // Modify the text file
            util::fs::write_to_path(&text_path, "Goodbye, world!")?;

            // Get the new hash
            let hash_after_modification = util::hasher::hash_file_contents(&text_path)?.parse()?;

            // Add and commit the file
            repositories::add(&repo, &text_path).await?;
            repositories::commit(&repo, "Committing goodbye world")?;

            // Get the most recent commit - the new head commit
            let head = repositories::commits::head_commit(&repo)?;

            // get the merkle tree for the commit
            let tree = repositories::tree::get_root_with_children(&repo, &head)?.unwrap();

            // Get the commit entry for the text file
            let text_entry = tree.get_by_path(Path::new("text.txt"))?.unwrap();

            // Hashes should be different
            assert_ne!(hash_when_add, hash_after_modification);

            // Hash should match new hash
            assert_eq!(text_entry.hash, hash_after_modification);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_file_and_dir() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Create committer with no commits
            let repo_path = &repo.path;
            let train_dir = repo_path.join("training_data");
            util::fs::create_dir_all(&train_dir)?;
            let _ = test::add_txt_file_to_dir(&train_dir, "Train Ex 1")?;
            let _ = test::add_txt_file_to_dir(&train_dir, "Train Ex 2")?;
            let _ = test::add_txt_file_to_dir(&train_dir, "Train Ex 3")?;
            let annotation_file = test::add_txt_file_to_dir(repo_path, "some annotations...")?;

            let test_dir = repo_path.join("test_data");
            util::fs::create_dir_all(&test_dir)?;
            let _ = test::add_txt_file_to_dir(&test_dir, "Test Ex 1")?;
            let _ = test::add_txt_file_to_dir(&test_dir, "Test Ex 2")?;

            // Add a file and a directory
            repositories::add(&repo, &annotation_file).await?;
            repositories::add(&repo, &train_dir).await?;

            let message = "Adding training data to 🐂";
            repositories::commit(&repo, message)?;

            // should be one commit now
            let commit_history = repositories::commits::list(&repo)?;
            assert_eq!(commit_history.len(), 1);

            // Check that the files are no longer staged
            let status = repositories::status(&repo)?;
            let files = status.staged_files;
            let dirs = status.staged_dirs;
            assert_eq!(files.len(), 0);
            assert_eq!(dirs.len(), 0);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_history_is_complete() -> Result<(), OxenError> {
        test::run_training_data_fully_sync_remote(|_local_repo, remote_repo| async move {
            let cloned_remote = remote_repo.clone();

            // Clone with the --all flag
            test::run_empty_dir_test_async(|new_repo_dir| async move {
                let new_repo_dir = new_repo_dir.join("repoo");
                let deep_clone =
                    repositories::deep_clone_url(&remote_repo.remote.url, &new_repo_dir).await?;
                // Get head commit of deep_clone repo
                let head_commit = repositories::commits::head_commit(&deep_clone)?;
                assert!(commit_history_is_complete(&deep_clone, &head_commit)?);
                Ok(())
            })
            .await?;

            Ok(cloned_remote)
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_history_is_not_complete_standard_repo() -> Result<(), OxenError> {
        test::run_training_data_fully_sync_remote(|_local_repo, remote_repo| async move {
            let cloned_remote = remote_repo.clone();

            // Clone with the --all flag
            test::run_empty_dir_test_async(|new_repo_dir| async move {
                let clone = repositories::clone_url(
                    &remote_repo.remote.url,
                    &new_repo_dir.join("new_repo"),
                )
                .await?;
                // Get head commit of deep_clone repo
                let head_commit = repositories::commits::head_commit(&clone)?;
                assert!(!commit_history_is_complete(&clone, &head_commit)?);
                Ok(())
            })
            .await?;

            Ok(cloned_remote)
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_history_order() -> Result<(), OxenError> {
        test::run_training_data_repo_test_no_commits_async(|repo| async move {
            let train_dir = repo.path.join("train");
            repositories::add(&repo, train_dir).await?;
            let initial_commit_message = "adding train dir";
            repositories::commit(&repo, initial_commit_message)?;

            // Write a text file
            let text_path = repo.path.join("newnewnew.txt");
            util::fs::write_to_path(&text_path, "Hello World")?;
            repositories::add(&repo, &text_path).await?;
            repositories::commit(&repo, "adding text file")?;

            let test_dir = repo.path.join("test");
            repositories::add(&repo, test_dir).await?;
            let most_recent_message = "adding test dir";
            repositories::commit(&repo, most_recent_message)?;

            let history = repositories::commits::list(&repo)?;
            assert_eq!(history.len(), 3);

            assert_eq!(history.first().unwrap().message, most_recent_message);
            assert_eq!(history.last().unwrap().message, initial_commit_message);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_get_commit_history_list_between() -> Result<(), OxenError> {
        test::run_training_data_repo_test_fully_committed_async(|repo| async move {
            let new_file = repo.path.join("new_1.txt");
            test::write_txt_file_to_path(&new_file, "new 1")?;
            repositories::add(&repo, new_file).await?;
            let base_commit = repositories::commit(&repo, "commit 1")?;

            let new_file = repo.path.join("new_2.txt");
            test::write_txt_file_to_path(&new_file, "new 2")?;
            repositories::add(&repo, new_file).await?;
            repositories::commit(&repo, "commit 2")?;

            let new_file = repo.path.join("new_3.txt");
            test::write_txt_file_to_path(&new_file, "new 3")?;
            repositories::add(&repo, new_file).await?;
            let head_commit = repositories::commit(&repo, "commit 3")?;

            let new_file = repo.path.join("new_4.txt");
            test::write_txt_file_to_path(&new_file, "new 4")?;
            repositories::add(&repo, new_file).await?;
            repositories::commit(&repo, "commit 4")?;

            let history = repositories::commits::list_between(&repo, &base_commit, &head_commit)?;
            assert_eq!(history.len(), 3);

            assert_eq!(history.first().unwrap().message, head_commit.message);
            assert_eq!(history.last().unwrap().message, base_commit.message);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_subdir_then_root_file() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Make a dir
            let dir_path = Path::new("test_dir");
            let dir_repo_path = repo.path.join(dir_path);
            util::fs::create_dir_all(dir_repo_path)?;

            // File in the dir
            let file_path = dir_path.join(Path::new("test_file.txt"));
            let file_repo_path = repo.path.join(&file_path);
            util::fs::write_to_path(&file_repo_path, "test")?;

            // Add the dir
            repositories::add(&repo, &repo.path).await?;
            let commit_1 = repositories::commit(&repo, "adding test dir")?;

            // New file in root
            let file_path_2 = Path::new("test_file_2.txt");
            let file_repo_path_2 = repo.path.join(file_path_2);
            util::fs::write_to_path(&file_repo_path_2, "test")?;

            // Add the file
            repositories::add(&repo, &file_repo_path_2).await?;
            let commit_2 = repositories::commit(&repo, "adding test file")?;

            let tree_1 = repositories::tree::get_root_with_children(&repo, &commit_1)?.unwrap();
            let tree_2 = repositories::tree::get_root_with_children(&repo, &commit_2)?.unwrap();

            // Make sure the file is not in the first commit
            // This was biting us in an initial implementation
            // BECAUSE the file contents was the same, the hash was not updated
            let node_from_tree_1 = tree_1.get_by_path(file_path_2)?;
            assert!(node_from_tree_1.is_none());

            // Make sure the file is in the second commit
            let node_from_tree_2 = tree_2.get_by_path(file_path_2)?;
            assert!(node_from_tree_2.is_some());

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_allow_empty() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Create and commit an initial file
            let hello_file = repo.path.join("hello.txt");
            util::fs::write_to_path(&hello_file, "Hello World")?;
            repositories::add(&repo, &hello_file).await?;
            let first_commit = repositories::commit(&repo, "Initial commit")?;

            // Try to create an empty commit without --allow-empty (should fail)
            let result = repositories::commit(&repo, "Empty commit");
            assert!(result.is_err());

            // Create an empty commit with --allow-empty (should succeed)
            let empty_commit = commit_allow_empty(&repo, "Empty commit")?;
            assert_eq!(empty_commit.message, "Empty commit");
            assert_eq!(empty_commit.parent_ids, vec![first_commit.id.clone()]);

            // Verify the tree is the same as the parent
            let first_tree =
                repositories::tree::get_root_with_children(&repo, &first_commit)?.unwrap();
            let empty_tree =
                repositories::tree::get_root_with_children(&repo, &empty_commit)?.unwrap();

            // Both trees should have the same file
            let first_file = first_tree.get_by_path(Path::new("hello.txt"))?;
            let empty_file = empty_tree.get_by_path(Path::new("hello.txt"))?;
            assert!(first_file.is_some());
            assert!(empty_file.is_some());
            assert_eq!(first_file.unwrap().hash, empty_file.unwrap().hash);

            // Verify commit history
            let history = repositories::commits::list(&repo)?;
            assert_eq!(history.len(), 2);
            assert_eq!(history[0].message, "Empty commit");
            assert_eq!(history[1].message, "Initial commit");

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_allow_empty_with_changes() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Create and commit an initial file
            let hello_file = repo.path.join("hello.txt");
            util::fs::write_to_path(&hello_file, "Hello World")?;
            repositories::add(&repo, &hello_file).await?;
            repositories::commit(&repo, "Initial commit")?;

            // Stage a new file
            let goodbye_file = repo.path.join("goodbye.txt");
            util::fs::write_to_path(&goodbye_file, "Goodbye World")?;
            repositories::add(&repo, &goodbye_file).await?;

            // commit_allow_empty should commit the staged changes normally
            let commit = commit_allow_empty(&repo, "Add goodbye")?;
            assert_eq!(commit.message, "Add goodbye");

            // Verify both files are in the tree
            let tree = repositories::tree::get_root_with_children(&repo, &commit)?.unwrap();
            assert!(tree.get_by_path(Path::new("hello.txt"))?.is_some());
            assert!(tree.get_by_path(Path::new("goodbye.txt"))?.is_some());

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_10k_files_vnode_size_10k() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Make a dir
            let dir_path = Path::new("test_dir");
            let dir_repo_path = repo.path.join(dir_path);
            util::fs::create_dir_all(&dir_repo_path)?;

            for i in 0..10000 {
                let file_path = dir_path.join(format!("file_{i}.txt"));
                let file_repo_path = repo.path.join(&file_path);
                util::fs::write_to_path(&file_repo_path, "test")?;
            }

            // Add a file called "images.csv" at the root of the repo
            let images_csv_path = Path::new("images.csv");
            let images_csv_repo_path = repo.path.join(images_csv_path);
            util::fs::write_to_path(&images_csv_repo_path, "images,path\n1,test.jpg\n2,test.png")?;

            repositories::add(&repo, &dir_repo_path).await?;
            repositories::add(&repo, &images_csv_repo_path).await?;
            let commit = repositories::commit(&repo, "adding 10k files")?;

            repositories::tree::print_tree(&repo, &commit)?;

            let file_node = repositories::tree::get_file_by_path(&repo, &commit, images_csv_path)?;
            assert!(file_node.is_some());

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_add_and_rm_empty_dir() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Make an empty dir
            let empty_dir = repo.path.join("empty_dir");
            util::fs::create_dir_all(&empty_dir)?;

            let status = repositories::status(&repo)?;
            status.print();

            // Should find the untracked dir
            assert!(
                status
                    .untracked_dirs
                    .iter()
                    .any(|(path, _)| *path == Path::new("empty_dir"))
            );

            // Add the empty dir
            repositories::add(&repo, &empty_dir).await?;

            let status = repositories::status(&repo)?;
            status.print();

            let commit = repositories::commit(&repo, "adding empty dir")?;

            let tree = repositories::tree::get_root_with_children(&repo, &commit)?.unwrap();

            assert!(tree.get_by_path(PathBuf::from("empty_dir"))?.is_some());

            // Remove the empty dir
            let rm_opts = RmOpts {
                path: PathBuf::from("empty_dir"),
                recursive: true,
                ..Default::default()
            };

            repositories::rm(&repo, &rm_opts)?;
            let commit_2 = repositories::commit(&repo, "removing empty dir")?;

            let tree_2 = repositories::tree::get_root_with_children(&repo, &commit_2)?.unwrap();
            assert!(tree_2.get_by_path(PathBuf::from("empty_dir"))?.is_none());

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_commit_invalid_parquet_file() -> Result<(), OxenError> {
        test::run_empty_data_repo_test_no_commits_async(|repo| async move {
            let invalid_parquet_file = test::test_invalid_parquet_file();
            let full_path = repo.path.join("invalid.parquet");
            util::fs::copy(&invalid_parquet_file, &full_path)?;

            repositories::add(&repo, &full_path).await?;
            let commit = repositories::commit(&repo, "Adding invalid parquet file")?;

            let tree = repositories::tree::get_root_with_children(&repo, &commit)?.unwrap();
            let file_node = tree.get_by_path(PathBuf::from("invalid.parquet"))?;
            assert!(file_node.is_some());

            let file_entry = file_node.unwrap();
            let file_node = file_entry.file()?;
            assert_eq!(*file_node.data_type(), EntryDataType::Binary);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_clone_annotations_test_subtree_commit_file() -> Result<(), OxenError> {
        test::run_training_data_fully_sync_remote(|_local_repo, remote_repo| async move {
            let cloned_remote = remote_repo.clone();
            test::run_empty_dir_test_async(|dir| async move {
                let mut opts = CloneOpts::new(&remote_repo.remote.url, dir.join("new_repo"));
                opts.fetch_opts.subtree_paths =
                    Some(vec![PathBuf::from("annotations").join("test")]);
                let local_repo = repositories::clone::clone(&opts).await?;

                let annotations_test_dir = local_repo.path.join("annotations").join("test");

                // Add a new file
                let readme_file = annotations_test_dir.join("README.md");
                util::fs::write_to_path(
                    &readme_file,
                    r"
Q: What is a good alternative to git LFS?
A: Oxen.ai
",
                )?;
                repositories::add(&local_repo, &readme_file).await?;
                let _commit =
                    repositories::commit(&local_repo, "adding README.md to the test dir")?;

                Ok(())
            })
            .await?;
            Ok(cloned_remote)
        })
        .await
    }

    // Test for updating file size after cloning subtree
    // I cloned subtree, added an empty file, committed, pushed, then edited the file and committed again
    // The file size should be updated in the index
    #[tokio::test]
    async fn test_clone_subtree_commit_file_update_size() -> Result<(), OxenError> {
        test::run_training_data_fully_sync_remote(|_local_repo, remote_repo| async move {
            let cloned_remote = remote_repo.clone();
            test::run_empty_dir_test_async(|dir| async move {
                let mut opts = CloneOpts::new(&remote_repo.remote.url, dir.join("new_repo"));
                opts.fetch_opts.subtree_paths = Some(vec![PathBuf::from(".")]);
                let local_repo = repositories::clone::clone(&opts).await?;

                // Add a new file
                let empty_file = local_repo.path.join("empty.txt");
                util::fs::write_to_path(&empty_file, "")?;
                repositories::add(&local_repo, &empty_file).await?;
                let commit = repositories::commit(&local_repo, "adding empty file")?;

                let tree =
                    repositories::tree::get_root_with_children(&local_repo, &commit)?.unwrap();

                let file_node = tree.get_by_path(PathBuf::from("empty.txt"))?;
                assert!(file_node.is_some());
                let file_node = file_node.unwrap().file()?;
                assert_eq!(file_node.num_bytes(), 0);

                // Edit the file
                let raw_str = r"
Q: What should I use to store massive machine learning datasets?
A: Oxen.ai
";
                util::fs::write_to_path(&empty_file, raw_str)?;

                repositories::add(&local_repo, &empty_file).await?;
                let commit = repositories::commit(&local_repo, "adding README.md to the test dir")?;

                let tree =
                    repositories::tree::get_root_with_children(&local_repo, &commit)?.unwrap();

                let file_node = tree.get_by_path(PathBuf::from("empty.txt"))?;
                assert!(file_node.is_some());
                let file_node = file_node.unwrap().file()?;
                assert_eq!(file_node.num_bytes(), raw_str.len() as u64);

                Ok(())
            })
            .await?;
            Ok(cloned_remote)
        })
        .await
    }

    #[tokio::test]
    async fn test_list_by_path_from_paginated() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            let target_file_path = PathBuf::from("target_file.txt");
            let other_file_path_1 = PathBuf::from("other_file_1.txt");
            let dummy_dir_path = PathBuf::from("dummy_dir");

            // Commit a: Add target_file
            let full_target_path = repo.path.join(&target_file_path);
            util::fs::write_to_path(&full_target_path, "Initial content")?;
            repositories::add(&repo, &full_target_path).await?;
            let commit_a = repositories::commit(&repo, "Add target_file.txt")?;

            // Commit b: without impacting target_file
            let full_other_path_1 = repo.path.join(&other_file_path_1);
            util::fs::write_to_path(&full_other_path_1, "Some other content")?;
            repositories::add(&repo, &full_other_path_1).await?;
            let _commit_b = repositories::commit(&repo, "Add other_file_1.txt")?;

            // Commit c: Modify target_file
            util::fs::write_to_path(&full_target_path, "Modified content 1")?;
            repositories::add(&repo, &full_target_path).await?;
            let commit_c = repositories::commit(&repo, "Modify target_file.txt first time")?;

            // Commit d: without impacting target_file
            let full_dummy_dir_path = repo.path.join(&dummy_dir_path);
            util::fs::create_dir_all(&full_dummy_dir_path)?;
            repositories::add(&repo, &full_dummy_dir_path).await?;
            let _commit_d = repositories::commit(&repo, "Add dummy dir")?;

            // Commit e: modify target_file.txt
            util::fs::write_to_path(&full_target_path, "Modified content 2")?;
            repositories::add(&repo, &full_target_path).await?;
            let commit_e = repositories::commit(&repo, "Modify target_file.txt second time")?;

            // Get the HEAD commit (should be commit_e)
            let head_commit = repositories::commits::head_commit(&repo)?;
            assert_eq!(head_commit.id, commit_e.id);

            let expected_commits = [commit_e.clone(), commit_c.clone(), commit_a.clone()];

            let pagination_opts = PaginateOpts::default();
            let paginated_result = repositories::commits::list_by_path_from_paginated(
                &repo,
                &head_commit,
                &target_file_path,
                pagination_opts,
            )?;

            assert_eq!(paginated_result.commits.len(), expected_commits.len());

            for (i, commit) in paginated_result.commits.iter().enumerate() {
                assert_eq!(
                    commit.id, expected_commits[i].id,
                    "Commits should match expected list at index {i}"
                );
            }

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_create_initial_commit_on_empty_repo() -> Result<(), OxenError> {
        test::run_empty_data_repo_test_no_commits_async(|repo| async move {
            // Verify repo is empty (no commits)
            assert!(head_commit_maybe(&repo)?.is_none());
            assert!(repositories::branches::list(&repo)?.is_empty());

            // Create initial commit
            let user = crate::model::User {
                name: "Test User".to_string(),
                email: "test@example.com".to_string(),
            };
            let commit = create_initial_commit(&repo, "main", &user, "Initial commit")?;

            // Verify commit was created correctly
            assert_eq!(commit.message, "Initial commit");
            assert_eq!(commit.author, "Test User");
            assert_eq!(commit.email, "test@example.com");
            assert!(commit.parent_ids.is_empty()); // No parents for initial commit

            // Verify HEAD now points to the commit
            let head = head_commit_maybe(&repo)?;
            assert!(head.is_some());
            assert_eq!(head.unwrap().id, commit.id);

            // Verify branch was created
            let branches = repositories::branches::list(&repo)?;
            assert_eq!(branches.len(), 1);
            assert_eq!(branches[0].name, "main");
            assert_eq!(branches[0].commit_id, commit.id);

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_create_initial_commit_fails_on_non_empty_repo() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // First create a file and commit it to make the repo non-empty
            let hello_file = repo.path.join("hello.txt");
            util::fs::write_to_path(&hello_file, "Hello World")?;
            repositories::add(&repo, &hello_file).await?;
            repositories::commit(&repo, "First commit")?;

            // Now create_initial_commit should fail
            let user = crate::model::User {
                name: "Test User".to_string(),
                email: "test@example.com".to_string(),
            };
            let result = create_initial_commit(&repo, "another-branch", &user, "Should fail");

            assert!(result.is_err());
            let err = result.unwrap_err();
            assert!(err.to_string().contains("already has commits"));

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_create_initial_commit_with_custom_branch_name() -> Result<(), OxenError> {
        test::run_empty_data_repo_test_no_commits_async(|repo| async move {
            // Create initial commit on a custom branch name
            let user = crate::model::User {
                name: "Test User".to_string(),
                email: "test@example.com".to_string(),
            };
            let commit = create_initial_commit(&repo, "develop", &user, "Initial commit")?;

            // Verify branch was created with custom name
            let branches = repositories::branches::list(&repo)?;
            assert_eq!(branches.len(), 1);
            assert_eq!(branches[0].name, "develop");
            assert_eq!(branches[0].commit_id, commit.id);

            // Verify HEAD points to the custom branch
            let current_branch = repositories::branches::current_branch(&repo)?;
            assert!(current_branch.is_some());
            assert_eq!(current_branch.unwrap().name, "develop");

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_create_initial_commit_then_second_commit() -> Result<(), OxenError> {
        test::run_empty_data_repo_test_no_commits_async(|repo| async move {
            // Create initial commit on empty repo
            let user = crate::model::User {
                name: "Test User".to_string(),
                email: "test@example.com".to_string(),
            };
            let initial_commit = create_initial_commit(&repo, "main", &user, "Initial commit")?;
            assert_eq!(initial_commit.message, "Initial commit");

            // Now add a file and try to commit again
            let hello_file = repo.path.join("hello.txt");
            util::fs::write_to_path(&hello_file, "Hello World")?;
            repositories::add(&repo, &hello_file).await?;

            // This second commit should succeed
            let second_commit = repositories::commit(&repo, "Add hello.txt")?;
            assert_eq!(second_commit.message, "Add hello.txt");

            // Verify the file is in the commit
            let status = repositories::status(&repo)?;
            assert!(status.is_clean());

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_list_by_path_commit_count_with_edge_cases() -> Result<(), OxenError> {
        test::run_empty_local_repo_test_async(|repo| async move {
            // Commit 1: Create data/a.txt and src/b.txt
            let data_a = repo.path.join("data").join("a.txt");
            let src_b = repo.path.join("src").join("b.txt");
            util::fs::write_to_path(&data_a, "a-v1")?;
            util::fs::write_to_path(&src_b, "b-v1")?;
            repositories::add(&repo, &repo.path.join("data")).await?;
            repositories::add(&repo, &repo.path.join("src")).await?;
            let _c1 = repositories::commit(&repo, "c1: add data/a.txt and src/b.txt")?;

            // Commit 2: Modify data/a.txt
            util::fs::write_to_path(&data_a, "a-v2")?;
            repositories::add(&repo, &data_a).await?;
            let _c2 = repositories::commit(&repo, "c2: modify data/a.txt")?;

            // Commit 3: Empty commit (no changes)
            let _c3 = commit_allow_empty(&repo, "c3: empty commit")?;

            // Commit 4: Add src/c.txt
            let src_c = repo.path.join("src").join("c.txt");
            util::fs::write_to_path(&src_c, "c-v1")?;
            repositories::add(&repo, &src_c).await?;
            let _c4 = repositories::commit(&repo, "c4: add src/c.txt")?;

            // Commit 5: Empty commit (no tree changes, should not appear in any path history)
            let _c5 = commit_allow_empty(&repo, "c5: another empty commit")?;

            // Commit 6: Add root-level other.txt
            let other = repo.path.join("other.txt");
            util::fs::write_to_path(&other, "other-v1")?;
            repositories::add(&repo, &other).await?;
            let _c6 = repositories::commit(&repo, "c6: add other.txt")?;

            // Commit 7: Modify src/b.txt
            util::fs::write_to_path(&src_b, "b-v2")?;
            repositories::add(&repo, &src_b).await?;
            let _c7 = repositories::commit(&repo, "c7: modify src/b.txt")?;

            // Commit 8: Modify data/a.txt
            util::fs::write_to_path(&data_a, "a-v3")?;
            repositories::add(&repo, &data_a).await?;
            let _c8 = repositories::commit(&repo, "c8: modify data/a.txt again")?;

            // Commit 9: Revert data/a.txt back to its c1 content ("a-v1").
            // The content matches c1, but the tree hash differs from parent c8,
            // so this should count as a real commit for data/a.txt.
            util::fs::write_to_path(&data_a, "a-v1")?;
            repositories::add(&repo, &data_a).await?;
            let _c9 = repositories::commit(&repo, "c9: revert data/a.txt to v1")?;

            let head = repositories::commits::head_commit(&repo)?;
            let opts = PaginateOpts::default();

            // data/a.txt: should have 4 commits (c9, c8, c2, c1)
            // c9 reverts to c1 content, but it differs from parent c8 so it counts
            let result = repositories::commits::list_by_path_from_paginated(
                &repo,
                &head,
                &PathBuf::from("data/a.txt"),
                opts.clone(),
            )?;
            assert_eq!(
                result.commits.len(),
                4,
                "data/a.txt expected 4 commits, got {}",
                result.commits.len()
            );

            // src/b.txt: should have 2 commits (c7, c1)
            let result = repositories::commits::list_by_path_from_paginated(
                &repo,
                &head,
                &PathBuf::from("src/b.txt"),
                opts.clone(),
            )?;
            assert_eq!(
                result.commits.len(),
                2,
                "src/b.txt expected 2 commits, got {}",
                result.commits.len()
            );

            // src/c.txt: should have 1 commit (c4)
            let result = repositories::commits::list_by_path_from_paginated(
                &repo,
                &head,
                &PathBuf::from("src/c.txt"),
                opts.clone(),
            )?;
            assert_eq!(
                result.commits.len(),
                1,
                "src/c.txt expected 1 commit, got {}",
                result.commits.len()
            );

            // other.txt: should have 1 commit (c6)
            let result = repositories::commits::list_by_path_from_paginated(
                &repo,
                &head,
                &PathBuf::from("other.txt"),
                opts.clone(),
            )?;
            assert_eq!(
                result.commits.len(),
                1,
                "other.txt expected 1 commit, got {}",
                result.commits.len()
            );

            // data/ directory: should have 4 commits (c9, c8, c2, c1)
            let result = repositories::commits::list_by_path_from_paginated(
                &repo,
                &head,
                &PathBuf::from("data"),
                opts.clone(),
            )?;
            assert_eq!(
                result.commits.len(),
                4,
                "data/ expected 4 commits, got {}",
                result.commits.len()
            );

            // src/ directory: should have 3 commits (c7, c4, c1)
            let result = repositories::commits::list_by_path_from_paginated(
                &repo,
                &head,
                &PathBuf::from("src"),
                opts.clone(),
            )?;
            assert_eq!(
                result.commits.len(),
                3,
                "src/ expected 3 commits, got {}",
                result.commits.len()
            );

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_count_from_after_three_way_merge() -> Result<(), OxenError> {
        // Verifies that count_from returns the correct number of unique commits
        // after a three-way merge (no double-counting shared ancestors).
        //
        // Graph:
        //   init - A - C - D - M (merge)
        //             \       /
        //              B --- E
        //
        // init = 1 (from run_one_commit_local_repo_test_async)
        // A, B, C, D, E = 5 commits on branches
        // M = merge commit
        // Total unique commits = 7

        test::run_one_commit_local_repo_test_async(|repo| async move {
            let main_branch = repositories::branches::current_branch(&repo)?.unwrap();

            // Commit A on main (shared ancestor)
            let a_path = repo.path.join("a.txt");
            util::fs::write_to_path(&a_path, "a")?;
            repositories::add(&repo, &a_path).await?;
            repositories::commit(&repo, "Commit A")?;

            // Branch off and create commits B, E
            let merge_branch_name = "feature";
            repositories::branches::create_checkout(&repo, merge_branch_name)?;
            let b_path = repo.path.join("b.txt");
            util::fs::write_to_path(&b_path, "b")?;
            repositories::add(&repo, &b_path).await?;
            repositories::commit(&repo, "Commit B")?;

            let e_path = repo.path.join("e.txt");
            util::fs::write_to_path(&e_path, "e")?;
            repositories::add(&repo, &e_path).await?;
            repositories::commit(&repo, "Commit E")?;

            // Back to main, create commits C, D
            repositories::checkout(&repo, &main_branch.name).await?;
            let c_path = repo.path.join("c.txt");
            util::fs::write_to_path(&c_path, "c")?;
            repositories::add(&repo, &c_path).await?;
            repositories::commit(&repo, "Commit C")?;

            let d_path = repo.path.join("d.txt");
            util::fs::write_to_path(&d_path, "d")?;
            repositories::add(&repo, &d_path).await?;
            repositories::commit(&repo, "Commit D")?;

            // Warm the cache for both branch tips before merging
            let (main_count, _) = repositories::commits::count_from(&repo, &main_branch.name)?;
            let (feature_count, _) = repositories::commits::count_from(&repo, merge_branch_name)?;

            // main: init -> A -> C -> D = 4 commits
            assert_eq!(main_count, 4, "main branch should have 4 commits");
            // feature: init -> A -> B -> E = 4 commits
            assert_eq!(feature_count, 4, "feature branch should have 4 commits");

            // Merge feature into main
            let merge_commit = repositories::merge::merge(&repo, merge_branch_name)
                .await?
                .expect("merge should produce a commit");
            assert_eq!(merge_commit.parent_ids.len(), 2);

            // Verify count_from on merge commit
            let (merge_count, _) = repositories::commits::count_from(&repo, &merge_commit.id)?;

            // Total unique commits: init, A, B, C, D, E, M = 7
            assert_eq!(
                merge_count, 7,
                "merge commit count should be 7 (unique commits), got {merge_count}"
            );

            // Also verify that list_from returns the same number
            let all_commits = repositories::commits::list_from(&repo, &merge_commit.id)?;
            assert_eq!(
                all_commits.len(),
                7,
                "list_from should return 7 commits, got {}",
                all_commits.len()
            );

            // count_from should agree with list_from
            assert_eq!(
                merge_count,
                all_commits.len(),
                "count_from ({merge_count}) should match list_from ({})",
                all_commits.len()
            );

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_count_from_after_merge_with_deep_shared_history() -> Result<(), OxenError> {
        // Verifies correct counting when branches share a long history.
        // The bug would cause near-doubling in this scenario.
        //
        // Graph:
        //   init - H1 - H2 - H3 - H4 - H5 - main_commit - M (merge)
        //                                    \             /
        //                                     feat_commit
        //
        // Total unique commits = 9

        test::run_one_commit_local_repo_test_async(|repo| async move {
            let main_branch = repositories::branches::current_branch(&repo)?.unwrap();

            // Create 5 commits of shared history
            for i in 1..=5 {
                let path = repo.path.join(format!("shared_{i}.txt"));
                util::fs::write_to_path(&path, format!("shared {i}"))?;
                repositories::add(&repo, &path).await?;
                repositories::commit(&repo, &format!("Shared commit {i}"))?;
            }

            // Branch off for feature
            let merge_branch_name = "feature";
            repositories::branches::create_checkout(&repo, merge_branch_name)?;
            let feat_path = repo.path.join("feature.txt");
            util::fs::write_to_path(&feat_path, "feature")?;
            repositories::add(&repo, &feat_path).await?;
            repositories::commit(&repo, "Feature commit")?;

            // Back to main, add one more commit
            repositories::checkout(&repo, &main_branch.name).await?;
            let main_path = repo.path.join("main_only.txt");
            util::fs::write_to_path(&main_path, "main only")?;
            repositories::add(&repo, &main_path).await?;
            repositories::commit(&repo, "Main-only commit")?;

            // Warm cache for both branches
            let (main_count, _) = repositories::commits::count_from(&repo, &main_branch.name)?;
            let (feature_count, _) = repositories::commits::count_from(&repo, merge_branch_name)?;
            assert_eq!(main_count, 7, "main: init + 5 shared + 1 main-only = 7");
            assert_eq!(feature_count, 7, "feature: init + 5 shared + 1 feature = 7");

            // Merge
            let merge_commit = repositories::merge::merge(&repo, merge_branch_name)
                .await?
                .expect("merge should produce a commit");

            let (merge_count, _) = repositories::commits::count_from(&repo, &merge_commit.id)?;

            // Unique: init + 5 shared + main_only + feat + merge = 9
            let all_commits = repositories::commits::list_from(&repo, &merge_commit.id)?;
            assert_eq!(
                merge_count,
                all_commits.len(),
                "count_from ({merge_count}) should match list_from ({})",
                all_commits.len()
            );
            assert_eq!(merge_count, 9, "should be 9, got {merge_count}");

            Ok(())
        })
        .await
    }

    #[tokio::test]
    async fn test_count_from_after_merge_no_prior_cache() -> Result<(), OxenError> {
        // Verifies correct count when parent branch counts are NOT cached
        // before the merge (no warm-up step).

        test::run_one_commit_local_repo_test_async(|repo| async move {
            let main_branch = repositories::branches::current_branch(&repo)?.unwrap();

            let a_path = repo.path.join("a.txt");
            util::fs::write_to_path(&a_path, "a")?;
            repositories::add(&repo, &a_path).await?;
            repositories::commit(&repo, "Commit A")?;

            // Branch and commit
            repositories::branches::create_checkout(&repo, "feature")?;
            let b_path = repo.path.join("b.txt");
            util::fs::write_to_path(&b_path, "b")?;
            repositories::add(&repo, &b_path).await?;
            repositories::commit(&repo, "Commit B")?;

            // Back to main and commit
            repositories::checkout(&repo, &main_branch.name).await?;
            let c_path = repo.path.join("c.txt");
            util::fs::write_to_path(&c_path, "c")?;
            repositories::add(&repo, &c_path).await?;
            repositories::commit(&repo, "Commit C")?;

            // Merge WITHOUT warming cache first
            let merge_commit = repositories::merge::merge(&repo, "feature")
                .await?
                .expect("merge should produce a commit");

            let (merge_count, _) = repositories::commits::count_from(&repo, &merge_commit.id)?;
            let all_commits = repositories::commits::list_from(&repo, &merge_commit.id)?;

            // init, A, B, C, M = 5
            assert_eq!(
                merge_count,
                all_commits.len(),
                "count_from ({merge_count}) should match list_from ({})",
                all_commits.len()
            );
            assert_eq!(merge_count, 5, "should be 5, got {merge_count}");

            Ok(())
        })
        .await
    }
}