quilt-rs 0.44.0

use std::collections::BTreeSet;
use std::collections::HashSet;
use std::path::PathBuf;

use tracing::log;

use crate::Error;
use crate::Res;
use crate::error::LoginError;
use crate::error::PackageOpError;
use crate::flow;
use crate::flow::UserMeta;
use std::sync::Arc;

use crate::flow::cache_remote_manifest;
use crate::io::remote::HostConfig;
use crate::io::remote::Remote;
use crate::io::remote::RemoteS3;
use crate::io::remote::WORKFLOWS_CONFIG_KEY;
use crate::io::remote::WorkflowIntent;
use crate::io::remote::WorkflowsConfig;
use crate::io::remote::fetch_workflow_rules;
use crate::io::remote::fetch_workflows_config;
use crate::io::remote::resolve_workflow;
use crate::io::remote::resolve_workflow_from_config;
use crate::io::storage::LocalStorage;
use crate::io::storage::LockGuard;
use crate::io::storage::Storage;
use crate::lineage;
use crate::lineage::CommitState;
use crate::lineage::InstalledPackageStatus;
use crate::lineage::LineagePaths;
use crate::lineage::SyncScope;
use crate::lineage::UpstreamState;
use crate::manifest::Manifest;
use crate::manifest::Workflow;
use crate::package_lock;
use crate::paths;
use crate::paths::copy_cached_to_installed;
use crate::workflow::WorkflowRules;
use quilt_uri::Host;
use quilt_uri::ManifestUri;
use quilt_uri::Namespace;
use quilt_uri::S3Uri;
use quilt_uri::UriError;

/// Result of a push operation visible to callers outside `quilt-rs`.
pub struct PushOutcome {
    pub manifest_uri: ManifestUri,
    /// Whether the pushed revision was certified as "latest".
    /// `false` when the remote's latest tag moved since we last checked
    /// (i.e. someone else pushed in the meantime).
    pub certified_latest: bool,
}

/// Result of a publish operation visible to callers outside `quilt-rs`.
/// Alias of [`flow::PublishOutcome`] parameterized over the public
/// [`PushOutcome`], so external callers see a non-generic type name.
pub type PublishOutcome = flow::PublishOutcome<PushOutcome>;

/// Every early return from the dry run means the same thing — nothing for a
/// pull to do, and so no incoming paths or newer revision to name.
fn nothing_to_pull() -> flow::PullPreview {
    flow::PullPreview {
        outcome: flow::PullOutcome::UpToDate,
        added: Vec::new(),
        changed: Vec::new(),
        removed: Vec::new(),
        latest_hash: None,
    }
}

/// Result of [`InstalledPackage::set_remote`].
///
/// The remote was set (and, on the first-push recommit path, a workflow may
/// have been stamped). `resolution_warning` is `Some(reason)` only on the
/// best-effort `BucketDefault` path where the remote was persisted but the
/// bucket's default workflow could **not** be resolved — the operation still
/// succeeds and no workflow is stamped, but the caller should surface the
/// reason so the user is not silently left ungoverned until push time. Every
/// other success path leaves it `None`.
#[derive(Debug, Default)]
pub struct SetRemoteOutcome {
    pub resolution_warning: Option<String>,
}

/// Similar to `LocalDomain` because it has access to the same lineage file and remote/storage
/// traits.
/// But it only manages one particular installed package.
/// It can be instantiated from `LocalDomain` by installing new or listing existing packages.
#[derive(Debug)]
pub struct InstalledPackage<S: Storage = LocalStorage, R: Remote = RemoteS3> {
    pub lineage: lineage::PackageLineageIo,
    pub paths: paths::DomainPaths,
    pub remote: Arc<R>,
    pub storage: S,
    pub namespace: Namespace,
}

impl<S: Storage + Clone, R: Remote> Clone for InstalledPackage<S, R> {
    fn clone(&self) -> Self {
        Self {
            lineage: self.lineage.clone(),
            paths: self.paths.clone(),
            remote: Arc::clone(&self.remote),
            storage: self.storage.clone(),
            namespace: self.namespace.clone(),
        }
    }
}

/// An [`InstalledPackage`] with its lock held: no other writer of this
/// package, in this process or another, runs until it drops.
///
/// Take one with [`InstalledPackage::lock`] or [`InstalledPackage::try_lock`].
/// Its writers are the package's writers without the lock, so a caller that
/// needs more than one call under one lock (the autopull tick classifies,
/// then pulls or publishes) makes them all on the handle. The handle has no
/// way to lock again: the lock is not reentrant, and a task that took its own
/// package's lock twice would wait on itself forever.
#[derive(Debug)]
pub struct LockedPackage<S: Storage = LocalStorage, R: Remote = RemoteS3> {
    package: InstalledPackage<S, R>,
    _held: LockGuard,
}

impl std::fmt::Display for InstalledPackage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, r#"Installed package "{}""#, self.namespace)
    }
}

impl<S: Storage + Clone + Sync, R: Remote> InstalledPackage<S, R> {
    /// Takes this package's lock, waiting while another writer holds it.
    ///
    /// Every public writer here takes it before its first read of the entry
    /// and holds it through its write. See [`LockedPackage`]. Each boxes the
    /// body it runs on the handle: inline, that body would sit inside every
    /// caller's future, and the CLI's command future then outgrows the
    /// compiler's layout depth limit.
    pub async fn lock(&self) -> Res<LockedPackage<S, R>> {
        let held = package_lock::lock(&self.storage, &self.paths, &self.namespace).await?;
        Ok(LockedPackage {
            package: self.clone(),
            _held: held,
        })
    }

    /// [`Self::lock`] without the wait: `None` while another writer holds it.
    pub async fn try_lock(&self) -> Res<Option<LockedPackage<S, R>>> {
        let held = package_lock::try_lock(&self.storage, &self.paths, &self.namespace).await?;
        Ok(held.map(|held| LockedPackage {
            package: self.clone(),
            _held: held,
        }))
    }

    pub async fn scaffold_paths(&self) -> Res {
        let home = self.lineage.domain_home(&self.storage).await?;
        self.paths
            .scaffold_for_installing(&self.storage, &home, &self.namespace)
            .await
    }

    pub async fn scaffold_paths_for_caching(&self, bucket: &str) -> Res {
        self.paths.scaffold_for_caching(&self.storage, bucket).await
    }

    /// The revisions this copy has, newest first.
    ///
    /// See [`flow::list_revisions`] for what "newest" means — it is acquisition
    /// order, not the order the revisions were made in.
    pub async fn revisions(&self) -> Res<Vec<flow::Revision>> {
        flow::list_revisions(&self.paths, &self.storage, &self.namespace).await
    }

    /// How many revisions this copy has, without parsing any of them.
    ///
    /// Equal to `self.revisions().await?.len()` when every manifest parses; a
    /// damaged one is still counted, where [`Self::revisions`] fails on it.
    pub async fn revision_count(&self) -> Res<usize> {
        flow::count_revisions(&self.paths, &self.storage, &self.namespace).await
    }

    /// [`Self::revisions`], each marked by whether the registry of
    /// `lineage`'s remote lists it — a timestamped pointer to it exists.
    /// One listing for the whole package, never inferred from `CommitState`.
    /// No remote, or a remote with no catalog host (so no registry to ask):
    /// no remote call, and every entry is unpublished.
    pub async fn revision_history(
        &self,
        lineage: &lineage::PackageLineage,
    ) -> Res<Vec<flow::HistoryEntry>> {
        let revisions = self.revisions().await?;
        let published = self.registry_listing(lineage).await?;
        Ok(revisions
            .into_iter()
            .map(|revision| flow::HistoryEntry {
                published: published.contains(&revision.hash),
                revision,
            })
            .collect())
    }

    /// What removing this package's revisions would free: which objects each
    /// uses that nothing else does. A read, under no lock, for showing; the
    /// removal measures again under its locks. See [`flow::RevisionUsage`].
    pub async fn revision_usage(&self) -> Res<flow::RevisionUsage> {
        flow::RevisionUsage::read(&self.paths, &self.storage, &self.namespace).await
    }

    /// Remove `hashes`, old revisions this copy holds, and the objects only
    /// they used. See [`flow::remove_revisions`].
    ///
    /// Try-locks the package: while another writer holds it, nothing is
    /// removed and the error is [`Error::PackageBusy`]. Under the lock it
    /// re-reads the lineage and the registry listing, and a hash that is
    /// [protected](flow::protection) or not held refuses the whole request
    /// with [`Error::RevisionNotRemovable`]. A listing that fails refuses it
    /// too, since an unpublished revision cannot be told apart without it.
    pub async fn remove_revisions(&self, hashes: &BTreeSet<String>) -> Res<flow::RemovalReport> {
        let Some(_locked) = self.try_lock().await? else {
            return Err(Error::PackageBusy(self.namespace.clone()));
        };
        let lineage = self.lineage().await?;
        let history = self.revision_history(&lineage).await?;
        let kept = flow::protection(&lineage, &history);
        for hash in hashes {
            let why = match kept.get(hash) {
                Some(reasons) => format!(
                    "it is kept ({})",
                    reasons
                        .iter()
                        .map(ToString::to_string)
                        .collect::<Vec<_>>()
                        .join(" \u{b7} ")
                ),
                None if !history.iter().any(|entry| entry.revision.hash == *hash) => {
                    "this copy does not hold it".to_string()
                }
                None => continue,
            };
            return Err(Error::RevisionNotRemovable {
                namespace: self.namespace.clone(),
                hash: hash.clone(),
                why,
            });
        }
        let domain = self.lineage.read_domain(&self.storage).await?;
        flow::remove_revisions(&self.paths, &self.storage, &domain, &self.namespace, hashes).await
    }

    /// The revisions the registry of `lineage`'s remote lists. No remote,
    /// or no catalog host (so no registry to ask): no request, and none.
    async fn registry_listing(&self, lineage: &lineage::PackageLineage) -> Res<HashSet<String>> {
        Ok(match lineage.remote_uri.as_ref() {
            Some(ManifestUri {
                origin: Some(host),
                bucket,
                namespace,
                ..
            }) => self
                .remote
                .published_revisions(host, bucket, namespace)
                .await?
                .into_iter()
                .collect(),
            _ => HashSet::new(),
        })
    }

    /// The revision selected by one lineage snapshot.
    ///
    /// Unlike [`Self::revisions`], this reads only the manifest selected by
    /// [`lineage::PackageLineage::current_hash`]. That distinction keeps an
    /// unrelated damaged historical manifest from breaking a read of the
    /// current package state. The caller supplies the snapshot so adjacent
    /// facts such as the bucket cannot come from a different revision if the
    /// lineage changes while a page read is in flight.
    pub async fn current_revision(
        &self,
        lineage: &lineage::PackageLineage,
    ) -> Res<Option<flow::Revision>> {
        let Some(hash) = lineage.current_hash().map(str::to_owned) else {
            return Ok(None);
        };

        let manifest = self.manifest_from_lineage(lineage).await?;
        let installed_path = self.paths.installed_manifest(&self.namespace, &hash);
        let obtained = self.storage.modified_timestamp(&installed_path).await?;

        Ok(Some(flow::Revision {
            hash,
            obtained,
            message: manifest.header.message,
        }))
    }

    /// [`flow::compare_for_resolve`] over the manifest `lineage` selects and the
    /// published `latest`, resolved now rather than read from `lineage`, whose
    /// `latest_hash` is only as fresh as the last write. The published manifest
    /// is cached by hash. The listing follows [`Self::revision_history`]'s rule:
    /// no catalog host, no request, and the whole chain is unpublished.
    ///
    /// # Errors
    /// No remote; the tag read, the manifest fetch or the listing failing.
    pub async fn resolve_comparison(
        &self,
        lineage: &lineage::PackageLineage,
    ) -> Res<flow::ResolveComparison> {
        let remote_uri = lineage.remote()?.clone();
        let fresh = flow::refresh_latest_hash(lineage.clone(), &*self.remote).await?;
        let published_uri = ManifestUri {
            hash: fresh.latest_hash,
            ..remote_uri.clone()
        };
        self.scaffold_paths_for_caching(&remote_uri.bucket).await?;
        let published =
            cache_remote_manifest(&self.paths, &self.storage, &*self.remote, &published_uri)
                .await?;
        let current = self.manifest_from_lineage(lineage).await?;
        let listed = self.registry_listing(lineage).await?;
        Ok(flow::compare_for_resolve(
            &current,
            &published,
            lineage.commit.as_ref(),
            &listed,
        ))
    }

    pub async fn manifest(&self) -> Res<Manifest> {
        let (_, lineage) = self.lineage.read(&self.storage).await?;
        self.manifest_from_lineage(&lineage).await
    }

    /// Read the manifest selected by an already-read lineage snapshot.
    ///
    /// Keeping selection and recovery on that same value prevents a concurrent
    /// lineage edit from changing which manifest supplies the message halfway
    /// through a larger read.
    pub async fn manifest_from_lineage(&self, lineage: &lineage::PackageLineage) -> Res<Manifest> {
        let Some(hash) = lineage.current_hash() else {
            return Ok(Manifest::default());
        };
        let installed_path = self.paths.installed_manifest(&self.namespace, hash);
        match Manifest::from_path(&self.storage, &installed_path).await {
            Ok(manifest) => return Ok(manifest),

            Err(e) => {
                log::warn!(
                    "Failed to read installed manifest at {}: {}",
                    installed_path.display(),
                    e
                );
            }
        }

        // If installed failed, try to recover from cache (only if we have a remote)
        match lineage.remote_uri.as_ref() {
            Some(remote_uri) => {
                log::info!("Attempting to recover from cache at {remote_uri}");
                let cached_manifest =
                    cache_remote_manifest(&self.paths, &self.storage, &*self.remote, remote_uri)
                        .await?;
                copy_cached_to_installed(&self.paths, &self.storage, remote_uri).await?;
                Ok(cached_manifest)
            }
            None => Err(Error::Uri(UriError::ManifestPath(
                "No installed manifest and no remote to recover from".to_string(),
            ))),
        }
    }

    pub async fn lineage(&self) -> Res<lineage::PackageLineage> {
        let (_, lineage) = self.lineage.read(&self.storage).await?;
        Ok(lineage)
    }

    pub async fn package_home(&self) -> Res<PathBuf> {
        self.lineage.package_home(&self.storage).await
    }

    /// Recompute working-tree status against the cached manifest without
    /// contacting the remote. Caller accepts that `upstream_state` reflects
    /// the last-known `latest_hash` rather than a freshly-resolved one;
    /// pair with `status` (which calls `refresh_latest_hash`) when remote
    /// freshness matters.
    pub async fn recompute_local_status(
        &self,
        host_config_opt: Option<HostConfig>,
    ) -> Res<InstalledPackageStatus> {
        let (package_home, lineage) = self.lineage.read(&self.storage).await?;
        let manifest = self.manifest().await?;

        let host_config = match host_config_opt {
            Some(hc) => hc,
            None => match lineage.remote_uri.as_ref() {
                Some(remote_uri) if !remote_uri.bucket.is_empty() => {
                    self.remote.host_config(remote_uri.origin.as_ref()).await?
                }
                _ => HostConfig::default(),
            },
        };

        let (_, status) = flow::status(
            lineage,
            &self.storage,
            &manifest,
            &package_home,
            host_config,
        )
        .await?;
        Ok(status)
    }

    pub async fn status(&self, host_config_opt: Option<HostConfig>) -> Res<InstalledPackageStatus> {
        let (_, status) = self.status_with_lineage(host_config_opt).await?;
        Ok(status)
    }

    /// [`Self::status`] and the lineage it was computed from — one read, so a
    /// caller that needs more of the lineage (the local commit) cannot pair the
    /// status with a different snapshot.
    async fn status_with_lineage(
        &self,
        host_config_opt: Option<HostConfig>,
    ) -> Res<(lineage::PackageLineage, InstalledPackageStatus)> {
        let (package_home, lineage) = self.lineage.read(&self.storage).await?;

        // Only refresh latest hash if we have a remote
        let (lineage, latest_refreshed) = match lineage.remote_uri.as_ref() {
            Some(_) => match flow::refresh_latest_hash(lineage.clone(), &*self.remote).await {
                Ok(lineage) => (lineage, true),
                Err(Error::Login(LoginError::NoSession(_))) => {
                    return Err(Error::Login(LoginError::NoSession(
                        lineage.remote_uri.as_ref().and_then(|r| r.origin.clone()),
                    )));
                }
                // A denial is not a degradable failure. Every other error
                // here means "the remote is unreachable right now", and
                // continuing on stale lineage is exactly right — it is what
                // lets the app work offline. A denial means the request
                // arrived and was refused: the answer will not change until
                // the role does, and swallowing it leaves the caller unable
                // to tell the user why. Callers distinguish it with
                // [`Error::is_access_denied`].
                Err(err) if err.is_access_denied() => return Err(err),
                // Nor is a rejected credential: the session is dead, and
                // stale lineage would report the package as fine.
                Err(err) if err.is_session_absent() => return Err(err),
                // No `latest` tag, and none ever seen: the remote answered,
                // and its answer is that nothing was published yet. That is
                // a read, not an outage, so the verdict is fresh.
                Err(err) if err.is_not_found() && lineage.latest_hash.is_empty() => (lineage, true),
                Err(err) => {
                    log::warn!("Failed to refresh latest hash: {err}");
                    (lineage, false)
                }
            },
            None => (lineage, false),
        };
        let manifest = self.manifest().await?;

        let host_config = match host_config_opt {
            Some(hc) => hc,
            None => match lineage.remote_uri.as_ref() {
                Some(remote_uri) if !remote_uri.bucket.is_empty() => {
                    self.remote.host_config(remote_uri.origin.as_ref()).await?
                }
                _ => HostConfig::default(),
            },
        };

        let (lineage, status) = flow::status(
            lineage,
            &self.storage,
            &manifest,
            &package_home,
            host_config,
        )
        .await?;
        Ok((
            lineage,
            InstalledPackageStatus {
                latest_refreshed,
                ..status
            },
        ))
    }

    /// This package's [`PackageState`](lineage::PackageState), as every front
    /// end shows it.
    ///
    /// Asks the remote only when there is something to ask: a remote with no
    /// catalog host is `Unknown` and a package with no remote is resolved from
    /// its lineage, both without the network or a working-tree walk. Otherwise
    /// this is [`Self::status`] — the remote's `latest` tip plus the working
    /// tree — resolved by [`PackageState::resolve`](lineage::PackageState::resolve).
    /// The fetched tip is not written back to lineage.
    ///
    /// Errors are `status`'s, returned as they are: a denial
    /// ([`Error::is_access_denied`]), a missing or rejected session, or any other
    /// failure. What a front end shows for each is its own policy.
    pub async fn state(&self) -> Res<lineage::PackageStateReport> {
        let lineage = self.lineage().await?;
        if let Some(report) = lineage::PackageStateReport::without_remote(&lineage) {
            return Ok(report);
        }
        // The commit flag comes from the lineage the status was computed from,
        // not the read above: a push landing in between would otherwise pair a
        // fresh `UpToDate` with a stale commit and say "revision not published".
        let (lineage, status) = self.status_with_lineage(None).await?;
        let changed_files = status.changes.len();
        Ok(lineage::PackageStateReport {
            state: lineage::PackageState::resolve(
                status.upstream_state,
                lineage.commit.is_some(),
                lineage.remote_uri.is_some(),
                Some(changed_files),
            ),
            upstream_state: status.upstream_state,
            changed_files: Some(changed_files),
            latest_refreshed: status.latest_refreshed,
        })
    }

    /// Downloads `paths` and starts tracking them. A path whose bytes the
    /// remote no longer holds is skipped, not an error: see
    /// [`InstallPathsReport::skipped`](flow::InstallPathsReport::skipped).
    pub async fn install_paths(&self, paths: &[PathBuf]) -> Res<flow::InstallPathsReport> {
        if paths.is_empty() {
            return Ok(flow::InstallPathsReport::default());
        }
        Box::pin(self.lock().await?.install_paths(paths)).await
    }

    async fn install_paths_unlocked(&self, paths: &[PathBuf]) -> Res<flow::InstallPathsReport> {
        self.scaffold_paths().await?;

        let (package_home, lineage) = self.lineage.read(&self.storage).await?;
        let remote_uri = lineage.remote()?;

        self.scaffold_paths_for_caching(&remote_uri.bucket).await?;

        let mut manifest = self.manifest_from_lineage(&lineage).await?;
        let (lineage, skipped) = flow::install_paths(
            lineage,
            &mut manifest,
            &self.paths,
            package_home,
            self.namespace.clone(),
            &self.storage,
            &*self.remote,
            &paths.iter().collect::<Vec<&PathBuf>>(),
        )
        .await?;
        let lineage = self.lineage.write(&self.storage, lineage).await?;
        Ok(flow::InstallPathsReport {
            paths: lineage.paths,
            skipped,
        })
    }

    pub async fn uninstall_paths(&self, paths: &Vec<PathBuf>) -> Res<LineagePaths> {
        Box::pin(self.lock().await?.uninstall_paths(paths)).await
    }

    async fn uninstall_paths_unlocked(&self, paths: &Vec<PathBuf>) -> Res<LineagePaths> {
        let (package_home, lineage) = self.lineage.read(&self.storage).await?;
        let lineage = flow::uninstall_paths(lineage, package_home, &self.storage, paths).await?;
        let lineage = self.lineage.write(&self.storage, lineage).await?;
        Ok(lineage.paths)
    }

    pub async fn revert_paths(&self, paths: &Vec<String>) -> Res {
        log::debug!("revert_paths: {paths:?}");
        unimplemented!()
    }

    /// Commit the package's pending changes as a new revision.
    ///
    /// See [`UserMeta`] for the metadata contract: `Keep` inherits the
    /// previous revision's package-level metadata, `Clear` removes it,
    /// `Set` replaces it.
    pub async fn commit(
        &self,
        message: String,
        user_meta: UserMeta,
        workflow: Option<Workflow>,
        host_config_opt: Option<HostConfig>,
    ) -> Res<CommitState> {
        Box::pin(
            self.lock()
                .await?
                .commit(message, user_meta, workflow, host_config_opt),
        )
        .await
    }

    async fn commit_unlocked(
        &self,
        message: String,
        user_meta: UserMeta,
        workflow: Option<Workflow>,
        host_config_opt: Option<HostConfig>,
    ) -> Res<CommitState> {
        self.scaffold_paths().await?;

        let (package_home, lineage) = self.lineage.read(&self.storage).await?;
        let mut manifest = self.manifest_from_lineage(&lineage).await?;

        let host_config = match host_config_opt {
            Some(hc) => hc,
            None => match lineage.remote_uri.as_ref() {
                Some(remote_uri) if !remote_uri.bucket.is_empty() => {
                    self.remote.host_config(remote_uri.origin.as_ref()).await?
                }
                _ => HostConfig::default(),
            },
        };

        // Captured before `host_config` moves into `flow::status`: the commit
        // gate fetches the workflow's config + schemas from the same origin the
        // workflow was resolved against.
        let host = host_config.host.clone();

        let (lineage, status) = flow::status(
            lineage,
            &self.storage,
            &manifest,
            &package_home,
            host_config,
        )
        .await?;

        let (lineage, commit) = flow::commit(
            lineage,
            &mut manifest,
            &self.paths,
            &self.storage,
            &*self.remote,
            host.as_ref(),
            package_home,
            status,
            self.namespace.clone(),
            message,
            user_meta,
            workflow,
        )
        .await?;
        self.lineage.write(&self.storage, lineage).await?;
        Ok(commit)
    }

    /// Commit any working-directory changes (if any) and push the revision to
    /// the remote in one step. Errors if the package has no remote or nothing
    /// to publish.
    ///
    /// Walks the working tree under the lock. A caller that already has a
    /// status walked under the same lock passes it to
    /// [`LockedPackage::publish`] instead.
    pub async fn publish(
        &self,
        message: String,
        user_meta: UserMeta,
        workflow: Option<Workflow>,
        host_config_opt: Option<HostConfig>,
    ) -> Res<PublishOutcome> {
        Box::pin(
            self.lock()
                .await?
                .publish(message, user_meta, workflow, host_config_opt, None),
        )
        .await
    }

    async fn publish_unlocked(
        &self,
        message: String,
        user_meta: UserMeta,
        workflow: Option<Workflow>,
        host_config_opt: Option<HostConfig>,
        status_opt: Option<InstalledPackageStatus>,
    ) -> Res<PublishOutcome> {
        self.scaffold_paths().await?;

        let (package_home, lineage) = self.lineage.read(&self.storage).await?;
        let remote_uri = match lineage.remote_uri.as_ref() {
            Some(uri) if !uri.bucket.is_empty() => uri.clone(),
            Some(_) => {
                return Err(Error::PackageOp(PackageOpError::Publish(
                    "Remote bucket not set. Use set_remote first.".to_string(),
                )));
            }
            None => {
                return Err(Error::PackageOp(PackageOpError::Publish(
                    "No remote configured. Use set_remote first.".to_string(),
                )));
            }
        };

        self.scaffold_paths_for_caching(&remote_uri.bucket).await?;

        let mut manifest = self.manifest_from_lineage(&lineage).await?;
        let host_config =
            host_config_opt.unwrap_or(self.remote.host_config(remote_uri.origin.as_ref()).await?);

        let (lineage, status) = match status_opt {
            Some(status) => (lineage, status),
            None => {
                flow::status(
                    lineage,
                    &self.storage,
                    &manifest,
                    &package_home,
                    host_config.clone(),
                )
                .await?
            }
        };

        let outcome = flow::publish(
            lineage,
            &mut manifest,
            &self.paths,
            &self.storage,
            &*self.remote,
            package_home,
            status,
            self.namespace.clone(),
            host_config,
            flow::CommitOptions {
                message,
                user_meta,
                workflow,
            },
        )
        .await?;

        let (committed, push_result) = match outcome {
            flow::PublishOutcome::CommittedAndPushed(p) => (true, p),
            flow::PublishOutcome::PushedOnly(p) => (false, p),
        };
        let certified_latest = push_result.certified_latest;
        let lineage = self
            .lineage
            .write(&self.storage, push_result.lineage)
            .await?;
        let push = PushOutcome {
            manifest_uri: lineage.remote()?.clone(),
            certified_latest,
        };
        Ok(if committed {
            PublishOutcome::CommittedAndPushed(push)
        } else {
            PublishOutcome::PushedOnly(push)
        })
    }

    /// Push the local revision to the remote.
    pub async fn push(&self, host_config_opt: Option<HostConfig>) -> Res<PushOutcome> {
        Box::pin(self.lock().await?.push(host_config_opt)).await
    }

    async fn push_unlocked(&self, host_config_opt: Option<HostConfig>) -> Res<PushOutcome> {
        let (_, lineage) = self.lineage.read(&self.storage).await?;
        Ok(self.push_from(lineage, host_config_opt).await?.0)
    }

    /// Pushes from the entry the caller read under the lock, and returns the
    /// entry it wrote with the outcome.
    async fn push_from(
        &self,
        lineage: lineage::PackageLineage,
        host_config_opt: Option<HostConfig>,
    ) -> Res<(PushOutcome, lineage::PackageLineage)> {
        self.scaffold_paths().await?;

        let remote_uri = match lineage.remote_uri.as_ref() {
            Some(uri) if !uri.bucket.is_empty() => uri.clone(),
            Some(_) => {
                return Err(Error::PackageOp(PackageOpError::Push(
                    "Remote bucket not set. Use set_remote first.".to_string(),
                )));
            }
            None => {
                return Err(Error::PackageOp(PackageOpError::Push(
                    "No remote configured. Use set_remote first.".to_string(),
                )));
            }
        };

        if lineage.commit.is_none() {
            return Err(Error::PackageOp(PackageOpError::Push(
                "No commits to push".to_string(),
            )));
        }

        self.scaffold_paths_for_caching(&remote_uri.bucket).await?;

        let manifest = self.manifest_from_lineage(&lineage).await?;

        let host_config =
            host_config_opt.unwrap_or(self.remote.host_config(remote_uri.origin.as_ref()).await?);

        let result = flow::push(
            lineage,
            manifest,
            &self.paths,
            &self.storage,
            &*self.remote,
            Some(self.namespace.clone()),
            host_config,
        )
        .await?;
        let certified_latest = result.certified_latest;
        let lineage = self.lineage.write(&self.storage, result.lineage).await?;
        let outcome = PushOutcome {
            manifest_uri: lineage.remote()?.clone(),
            certified_latest,
        };
        Ok((outcome, lineage))
    }

    /// Record this package's standing [`SyncScope`].
    ///
    /// Storage only, and deliberately separate from [`Self::pull`]: writing the
    /// choice and acting on it are different decisions, made by different
    /// callers. Nothing in this crate reads the stored value back.
    ///
    /// Like every writer here it holds the package's lock, so a choice made
    /// while the autosync tick pulls waits for the pull and then lands on the
    /// entry the pull wrote.
    pub async fn set_sync_scope(&self, scope: SyncScope) -> Res<()> {
        Box::pin(self.lock().await?.set_sync_scope(scope)).await
    }

    async fn set_sync_scope_unlocked(&self, scope: SyncScope) -> Res<()> {
        self.lineage
            .edit(&self.storage, |l| l.sync_scope = scope)
            .await?;
        Ok(())
    }

    /// `scope` comes from the caller, not from the package's stored
    /// [`SyncScope`]. Both faces of this engine —
    /// the desktop app and the `quilt` CLI — reach pull through here, and only
    /// one of them has a setting; reading the field at this level would let
    /// state the app wrote change what the CLI does.
    pub async fn pull(
        &self,
        host_config_opt: Option<HostConfig>,
        scope: SyncScope,
    ) -> Res<flow::PullReport> {
        Box::pin(self.lock().await?.pull(host_config_opt, scope)).await
    }

    async fn pull_unlocked(
        &self,
        host_config_opt: Option<HostConfig>,
        scope: SyncScope,
    ) -> Res<flow::PullReport> {
        self.scaffold_paths().await?;

        let (package_home, lineage) = self.lineage.read(&self.storage).await?;
        let remote_uri = lineage.remote()?.clone();

        self.scaffold_paths_for_caching(&remote_uri.bucket).await?;

        let mut manifest = self.manifest_from_lineage(&lineage).await?;

        let host_config =
            host_config_opt.unwrap_or(self.remote.host_config(remote_uri.origin.as_ref()).await?);

        // All network (tag resolve + manifest fetch) happens here, before the
        // status walk — the snapshot is the freshest classification input.
        let (lineage, snapshot) = flow::snapshot_for_pull(
            lineage,
            &manifest,
            &self.paths,
            &self.storage,
            &*self.remote,
            &package_home,
            host_config,
        )
        .await?;
        let (lineage, report) = flow::pull(
            lineage,
            &mut manifest,
            &self.paths,
            &self.storage,
            &*self.remote,
            package_home,
            snapshot,
            self.namespace.clone(),
            scope,
        )
        .await?;
        self.lineage.write(&self.storage, lineage).await?;
        Ok(report)
    }

    /// Dry-run: what would `pull` do right now, without mutating anything?
    ///
    /// Sequence:
    /// - A **Local** package (no usable remote, per [`UpstreamState::Local`]:
    ///   `remote_uri` is `None`, a bucket-less remote, or a bucket that has
    ///   never been pushed) → [`PullOutcome::UpToDate`] with no network. There
    ///   is no `latest` tag to resolve for these shapes, so the tag read is
    ///   skipped rather than failing on the missing remote / absent tag.
    /// - Otherwise the `latest` tag is resolved once; if the resolved tip
    ///   already equals `base_hash` → `UpToDate` (that single tag read is the
    ///   only network paid for).
    /// - Otherwise the `latest` manifest is fetched + cached, the working tree
    ///   is walked, and the outcome is classified. Non-`Behind` upstream states
    ///   (`Ahead`/`Diverged`) report `UpToDate` — there is nothing to pull.
    ///
    /// [`PullOutcome::UpToDate`] here means "nothing for pull to do" and is
    /// returned for ALL non-`Behind` states (`Ahead`/`Local`/`Diverged`), not
    /// only when the package is genuinely current.
    ///
    /// Network-light — the caller (watcher / UI) uses it for two-phase render
    /// and routing. A caller that needs only the verdict uses
    /// [`Self::pull_verdict`], which skips collecting the incoming paths.
    ///
    /// # Errors
    /// For a package with a real remote, propagates tag-resolution, manifest
    /// read, and remote fetch errors. The Local early return never touches the
    /// network, so those shapes cannot produce those errors.
    pub async fn pull_outcome(
        &self,
        host_config_opt: Option<HostConfig>,
    ) -> Res<flow::PullPreview> {
        self.pull_dry_run(host_config_opt, true).await
    }

    /// [`Self::pull_outcome`]'s verdict alone: the same reads and the same
    /// answer, without collecting the paths the newer revision brings. For a
    /// caller that only routes on the verdict, such as a background check run
    /// on every package at an interval.
    ///
    /// # Errors
    /// As [`Self::pull_outcome`].
    pub async fn pull_verdict(
        &self,
        host_config_opt: Option<HostConfig>,
    ) -> Res<flow::PullOutcome> {
        Ok(self.pull_dry_run(host_config_opt, false).await?.outcome)
    }

    /// The dry run behind [`Self::pull_outcome`] and [`Self::pull_verdict`].
    /// With `with_paths` unset the preview's path lists stay empty.
    async fn pull_dry_run(
        &self,
        host_config_opt: Option<HostConfig>,
        with_paths: bool,
    ) -> Res<flow::PullPreview> {
        let (package_home, lineage) = self.lineage.read(&self.storage).await?;

        // A local-only package has no `latest` tag to resolve: `snapshot_for_pull`
        // would either error at `remote()?` (no `remote_uri`) or 404 on the
        // never-created `latest` tag. Mirror `UpstreamState::from`'s Local shape
        // and report `UpToDate` without any network, restoring the pre-reorder
        // contract. A remote whose local hash is empty but whose `latest` tag has
        // moved classifies as `Diverged` (not `Local`), so it still fetches below.
        if UpstreamState::from(lineage.clone()) == UpstreamState::Local {
            return Ok(nothing_to_pull());
        }

        // Divergence-by-hash is a purely lineage-local fact: `UpstreamState::from`
        // reports `Diverged` from on-disk state when the local side is BOTH ahead
        // (`base != current_hash`) and behind (`base != latest_hash`). The
        // "ahead" component involves only the local commit/remote hash — a moved
        // `latest` tag can neither cause nor cure it — so no network is needed to
        // decide it, and this can never mask a genuine `Behind` (which is
        // ahead-free). Short-circuit before the snapshot constructor, symmetric
        // with the `Local` return above.
        //
        // The OTHER `Diverged` shape — a pending local commit atop a base whose
        // `latest_hash` is still stale (equal to `base`) on disk — reads as
        // `Ahead` here, not `Diverged`; it only becomes `Diverged` once the tag
        // read refreshes `latest_hash`, so it correctly falls through to the
        // post-walk `!= Behind` check below rather than being caught here.
        if UpstreamState::from(lineage.clone()) == UpstreamState::Diverged {
            return Ok(nothing_to_pull());
        }

        let remote_uri = lineage.remote()?.clone();
        let base = self.manifest().await?;
        let host_config =
            host_config_opt.unwrap_or(self.remote.host_config(remote_uri.origin.as_ref()).await?);

        // Build the classification snapshot with the same ctor `pull` uses: one
        // tag resolution, then the manifest fetch, then the walk. The ctor
        // short-circuits `base == latest` before any fetch, so `Ahead` (where
        // `latest == base`) costs no network here.
        let snapshot = match flow::snapshot_for_pull(
            lineage,
            &base,
            &self.paths,
            &self.storage,
            &*self.remote,
            &package_home,
            host_config,
        )
        .await
        {
            Ok((_, snapshot)) => snapshot,
            Err(Error::PackageOp(PackageOpError::AlreadyUpToDate)) => {
                return Ok(nothing_to_pull());
            }
            Err(err) => return Err(err),
        };

        // `upstream_state` is computed from the ctor-refreshed lineage, so
        // `Diverged`/`Ahead` still report `UpToDate` (nothing to pull), matching
        // the previous contract.
        if snapshot.status.upstream_state != UpstreamState::Behind {
            return Ok(nothing_to_pull());
        }
        // Same reconciling pass the pull itself runs: a preview that called a
        // byte-identical edit a conflict would send the user to resolve one the
        // pull would not raise.
        let identical = flow::identical_to_latest(
            &self.storage,
            &package_home,
            &snapshot.status,
            &base,
            &snapshot.latest_manifest,
        )
        .await?;
        // One delta serves both the verdict and the incoming paths, so building
        // it twice would double the path and hash cloning on a large manifest.
        let delta = flow::pull_delta(&base, &snapshot.latest_manifest);
        let outcome = flow::classify_pull_with_delta(
            &snapshot.status,
            &base,
            &snapshot.latest_manifest,
            &identical,
            &delta,
        );
        let flow::IncomingPaths {
            added,
            changed,
            removed,
        } = if with_paths {
            flow::IncomingPaths::from_delta(&delta)
        } else {
            flow::IncomingPaths::default()
        };
        Ok(flow::PullPreview {
            outcome,
            added,
            changed,
            removed,
            // The hash the `latest` tag resolved to — the same one the
            // snapshot fetched the manifest by and wrote into the lineage's
            // `latest_hash`.
            latest_hash: Some(snapshot.latest.hash),
        })
    }

    /// Pushes any pending local commit, then promotes the resulting remote
    /// hash to `latest`. Last-writer-wins: any concurrent move of the
    /// `latest` tag between push and tag is overwritten. Invoked from the
    /// merge page when the user resolves a `Diverged` state in favor of
    /// their own revision.
    pub async fn certify_latest(&self) -> Res<ManifestUri> {
        Box::pin(self.lock().await?.certify_latest()).await
    }

    async fn certify_latest_unlocked(&self) -> Res<ManifestUri> {
        let (_, lineage) = self.lineage.read(&self.storage).await?;

        // Push first so the hash we tag exists on remote, under the lock this
        // call already holds, and carry on from the entry the push wrote.
        let lineage = if lineage.commit.is_some() {
            self.push_from(lineage, None).await?.1
        } else {
            lineage
        };

        let pushed_manifest_uri = lineage.remote()?.clone();
        let lineage = flow::certify_latest(lineage, &*self.remote, pushed_manifest_uri).await?;
        let lineage = self.lineage.write(&self.storage, lineage).await?;
        Ok(lineage.remote()?.clone())
    }

    pub async fn reset_to_latest(&self) -> Res<ManifestUri> {
        Box::pin(self.lock().await?.reset_to_latest()).await
    }

    async fn reset_to_latest_unlocked(&self) -> Res<ManifestUri> {
        self.scaffold_paths().await?;

        let (package_home, lineage) = self.lineage.read(&self.storage).await?;
        let remote_uri = lineage.remote()?.clone();

        self.scaffold_paths_for_caching(&remote_uri.bucket).await?;

        let mut manifest = self.manifest_from_lineage(&lineage).await?;
        let lineage = flow::reset_to_latest(
            lineage,
            &mut manifest,
            &self.paths,
            &self.storage,
            &*self.remote,
            package_home,
            self.namespace.clone(),
        )
        .await?;
        let lineage = self.lineage.write(&self.storage, lineage).await?;
        Ok(lineage.remote()?.clone())
    }

    /// Discard this package's newest local commit and restore the revision
    /// before it. See [`flow::undo_commit`].
    ///
    /// The remote check below is narrower than the chain rule: a remote-backed
    /// package with unpushed commits still has a chain, but undoing there would
    /// leave a pending commit equal to its own base.
    pub async fn undo_commit(&self) -> Res<CommitState> {
        Box::pin(self.lock().await?.undo_commit()).await
    }

    async fn undo_commit_unlocked(&self) -> Res<CommitState> {
        self.scaffold_paths().await?;

        let (package_home, lineage) = self.lineage.read(&self.storage).await?;
        if lineage.remote_uri.is_some() {
            return Err(Error::PackageOp(crate::PackageOpError::Undo(
                "this package has a remote, so its commit chain has been consumed by pushing; \
                 undo is only available before the first push"
                    .to_string(),
            )));
        }

        let (lineage, commit) = flow::undo_commit(
            lineage,
            &self.paths,
            &self.storage,
            package_home,
            self.namespace.clone(),
        )
        .await?;
        self.lineage.write(&self.storage, lineage).await?;
        Ok(commit)
    }

    pub async fn set_remote(
        &self,
        bucket: String,
        origin: Option<Host>,
        workflow: WorkflowIntent,
    ) -> Res<SetRemoteOutcome> {
        Box::pin(self.lock().await?.set_remote(bucket, origin, workflow)).await
    }

    async fn set_remote_unlocked(
        &self,
        bucket: String,
        origin: Option<Host>,
        workflow: WorkflowIntent,
    ) -> Res<SetRemoteOutcome> {
        if bucket.is_empty() {
            return Err(Error::PackageOp(PackageOpError::Push(
                "Bucket cannot be empty".to_string(),
            )));
        }
        let (_, mut lineage) = self.lineage.read(&self.storage).await?;
        if let Some(existing) = &lineage.remote_uri
            && !existing.hash.is_empty()
        {
            let same_remote = existing.bucket == bucket && existing.origin == origin;
            if same_remote {
                return Ok(SetRemoteOutcome::default());
            }
            return Err(Error::PackageOp(PackageOpError::Push(
                "Cannot change remote on a package that has already been pushed".to_string(),
            )));
        }
        // Validate the bucket up front so a typo surfaces here instead of
        // later at push time as an opaque S3 routing error. This is an
        // unauthenticated HEAD against s3.amazonaws.com — works even when
        // the user hasn't logged into the catalog yet.
        self.remote.verify_bucket(&bucket).await?;
        lineage.remote_uri = Some(ManifestUri {
            origin: origin.clone(),
            bucket: bucket.clone(),
            namespace: self.namespace.clone(),
            hash: String::new(),
        });

        // An explicit workflow gesture (`Named`/`NoWorkflow`) must not be
        // silently dropped: if the recommit that stamps it fails, surface the
        // error. The no-gesture `BucketDefault` path stays best-effort for
        // *resolution* failures only — validity is never best-effort.
        let explicit_workflow = !matches!(&workflow, WorkflowIntent::BucketDefault);

        // Re-commit with the remote's host_config and workflow so push works
        // immediately without a manual re-commit. Nothing has been persisted
        // yet: on success the recommit itself writes the lineage (carrying the
        // remote set above) and the new manifest; on failure the kind of error
        // decides what, if anything, is saved.
        if let Some(origin) = origin
            && lineage.commit.is_some()
        {
            return match self
                .recommit_for_remote(lineage.clone(), origin, bucket, workflow)
                .await
            {
                Ok(()) => Ok(SetRemoteOutcome::default()),
                // The workflow gate rejected the committed revision. The
                // package's previous state must stay fully intact, so the
                // remote is NOT saved either — set_remote fails as a whole
                // and nothing is persisted.
                Err(err @ Error::WorkflowValidation(_)) => Err(err),
                Err(err) => {
                    // A resolution or transient failure (e.g. not logged in
                    // yet, or an unknown workflow id): persist the remote so
                    // the user can fix the problem and retry.
                    self.lineage.write(&self.storage, lineage).await?;
                    if explicit_workflow {
                        // The remote is persisted, but the chosen workflow
                        // could not be applied. Fail loudly so the user can
                        // fix the workflow id or log in and re-run Set Remote
                        // instead of pushing with the wrong workflow.
                        return Err(err);
                    }
                    // Best-effort BucketDefault path: the remote is saved but
                    // the bucket's default workflow could not be resolved. The
                    // operation succeeds without a workflow stamp; carry the
                    // reason back so the caller can surface it rather than
                    // leaving the user silently ungoverned until push time.
                    // Unwrap to the inner message so user-facing callers (the
                    // CLI's stderr warning, the Set-remote popup) don't show
                    // the "Remote catalog error: …" wrapper chain, matching
                    // how the selector's Invalid notice surfaces these.
                    let reason = match &err {
                        Error::RemoteCatalog(inner) => inner.to_string(),
                        _ => err.to_string(),
                    };
                    log::warn!(
                        "Remote saved but recommit failed ({reason}); re-run Set Remote (e.g. after logging in) to complete it before pushing."
                    );
                    Ok(SetRemoteOutcome {
                        resolution_warning: Some(reason),
                    })
                }
            };
        }

        // No origin or no local commit — nothing to recommit or validate.
        self.lineage.write(&self.storage, lineage).await?;

        Ok(SetRemoteOutcome::default())
    }

    async fn recommit_for_remote(
        &self,
        lineage: lineage::PackageLineage,
        origin: Host,
        bucket: String,
        workflow: WorkflowIntent,
    ) -> Res {
        let host = Some(origin);
        let host_config = self.remote.host_config(host.as_ref()).await?;
        let workflows_config_uri = S3Uri {
            key: WORKFLOWS_CONFIG_KEY.to_string(),
            bucket,
            version: None,
        };
        // Fetch the bucket's workflows config exactly once, then reuse the
        // parsed value for both resolution and the recommit gate below — the
        // gate would otherwise re-download the same config via the header's
        // pinned URI.
        let (config_uri, workflows_config) =
            fetch_workflows_config(&*self.remote, host.as_ref(), &workflows_config_uri).await?;
        // Publish later pushes this pending recommit *without* re-resolving the
        // workflow, so recommit must stamp the caller's chosen workflow now.
        // With `WorkflowIntent::BucketDefault` (the no-gesture path) this picks
        // up the bucket's `default_workflow`, so a locally-created package's
        // first publish is governed even when the user expresses no choice.
        let workflow = resolve_workflow_from_config(
            &*self.remote,
            host.as_ref(),
            workflow,
            config_uri,
            workflows_config.as_ref(),
        )
        .await?;
        let manifest = self.manifest_from_lineage(&lineage).await?;
        let lineage = flow::recommit(
            lineage,
            &manifest,
            &self.paths,
            &self.storage,
            &*self.remote,
            host.as_ref(),
            self.namespace.clone(),
            host_config,
            workflow,
            workflows_config.as_ref(),
        )
        .await?;
        self.lineage.write(&self.storage, lineage).await?;
        Ok(())
    }

    /// The remote host and the `.quilt/workflows/config.yml` address for this
    /// package's bucket, or `None` when there is no usable remote. Builds the
    /// address from the package's own remote for the two read paths that need
    /// it ([`Self::resolve_workflow`] and [`Self::workflows_config`]); the key
    /// itself is the shared [`WORKFLOWS_CONFIG_KEY`].
    async fn workflows_config_location(&self) -> Res<Option<(Option<Host>, S3Uri)>> {
        let (_, lineage) = self.lineage.read(&self.storage).await?;
        let remote_uri = match lineage.remote_uri.as_ref() {
            Some(uri) if !uri.bucket.is_empty() => uri.clone(),
            _ => return Ok(None),
        };
        let config_uri = S3Uri {
            key: WORKFLOWS_CONFIG_KEY.to_string(),
            ..S3Uri::from(remote_uri.clone())
        };
        Ok(Some((remote_uri.origin, config_uri)))
    }

    pub async fn resolve_workflow(&self, intent: WorkflowIntent) -> Res<Option<Workflow>> {
        let Some((origin, config_uri)) = self.workflows_config_location().await? else {
            return Ok(None);
        };
        resolve_workflow(&*self.remote, origin.as_ref(), intent, &config_uri).await
    }

    /// Fetch and parse the bucket's `.quilt/workflows/config.yml` for this
    /// package's remote, returning the typed [`WorkflowsConfig`].
    ///
    /// Returns `Ok(None)` when the package has no remote or the bucket has no
    /// config, so callers building UI can degrade gracefully. This is the same
    /// fetch [`Self::resolve_workflow`] performs, exposed as a read-only view of
    /// the declared workflows rather than a resolution outcome.
    pub async fn workflows_config(&self) -> Res<Option<WorkflowsConfig>> {
        let Some((origin, config_uri)) = self.workflows_config_location().await? else {
            return Ok(None);
        };
        let (_, config) =
            fetch_workflows_config(&*self.remote, origin.as_ref(), &config_uri).await?;
        Ok(config)
    }

    /// Fetch and compile the pure-validator [`WorkflowRules`] for a named
    /// workflow declared in this package's bucket config, for live commit-dialog
    /// validation.
    ///
    /// Returns `Ok(None)` when the package has no remote or the bucket has no
    /// config — an ungoverned package has no rules to validate against. This is
    /// the same config fetch [`Self::resolve_workflow`] and
    /// [`Self::workflows_config`] perform, followed by [`fetch_workflow_rules`]
    /// to load the workflow's schema documents; the resulting rules feed
    /// [`crate::workflow::validate_candidate_fields`], mirroring how the commit
    /// gate calls [`fetch_workflow_rules`] + `validate_package`. A schema fetch
    /// failure or an unknown `workflow_id` surfaces as the underlying error, so
    /// the advisory caller can decide to skip validation rather than block.
    pub async fn workflow_rules(&self, workflow_id: &str) -> Res<Option<WorkflowRules>> {
        let Some((origin, config_uri)) = self.workflows_config_location().await? else {
            return Ok(None);
        };
        let (_, config) =
            fetch_workflows_config(&*self.remote, origin.as_ref(), &config_uri).await?;
        let Some(config) = config else {
            return Ok(None);
        };
        Ok(Some(
            fetch_workflow_rules(&*self.remote, origin.as_ref(), &config, workflow_id).await?,
        ))
    }
}

/// The package's writers on the held lock. Each is the [`InstalledPackage`]
/// method of the same name, without taking the lock.
impl<S: Storage + Clone + Sync, R: Remote> LockedPackage<S, R> {
    #[must_use]
    pub fn namespace(&self) -> &Namespace {
        &self.package.namespace
    }

    /// [`InstalledPackage::status`], read under the lock.
    pub async fn status(&self, host_config_opt: Option<HostConfig>) -> Res<InstalledPackageStatus> {
        self.package.status(host_config_opt).await
    }

    /// [`InstalledPackage::pull_outcome`], read under the lock.
    pub async fn pull_outcome(
        &self,
        host_config_opt: Option<HostConfig>,
    ) -> Res<flow::PullPreview> {
        self.package.pull_outcome(host_config_opt).await
    }

    /// [`InstalledPackage::pull_verdict`], read under the lock.
    pub async fn pull_verdict(
        &self,
        host_config_opt: Option<HostConfig>,
    ) -> Res<flow::PullOutcome> {
        self.package.pull_verdict(host_config_opt).await
    }

    pub async fn install_paths(&self, paths: &[PathBuf]) -> Res<flow::InstallPathsReport> {
        self.package.install_paths_unlocked(paths).await
    }

    pub async fn uninstall_paths(&self, paths: &Vec<PathBuf>) -> Res<LineagePaths> {
        self.package.uninstall_paths_unlocked(paths).await
    }

    pub async fn commit(
        &self,
        message: String,
        user_meta: UserMeta,
        workflow: Option<Workflow>,
        host_config_opt: Option<HostConfig>,
    ) -> Res<CommitState> {
        self.package
            .commit_unlocked(message, user_meta, workflow, host_config_opt)
            .await
    }

    /// [`InstalledPackage::publish`]. `status_opt` is a status the caller
    /// walked on this same handle; when `Some`, the publish reuses it rather
    /// than walking the tree again. The lock held since then means nothing
    /// else changed the package in between.
    pub async fn publish(
        &self,
        message: String,
        user_meta: UserMeta,
        workflow: Option<Workflow>,
        host_config_opt: Option<HostConfig>,
        status_opt: Option<InstalledPackageStatus>,
    ) -> Res<PublishOutcome> {
        self.package
            .publish_unlocked(message, user_meta, workflow, host_config_opt, status_opt)
            .await
    }

    pub async fn push(&self, host_config_opt: Option<HostConfig>) -> Res<PushOutcome> {
        self.package.push_unlocked(host_config_opt).await
    }

    pub async fn set_sync_scope(&self, scope: SyncScope) -> Res<()> {
        self.package.set_sync_scope_unlocked(scope).await
    }

    pub async fn pull(
        &self,
        host_config_opt: Option<HostConfig>,
        scope: SyncScope,
    ) -> Res<flow::PullReport> {
        self.package.pull_unlocked(host_config_opt, scope).await
    }

    pub async fn certify_latest(&self) -> Res<ManifestUri> {
        self.package.certify_latest_unlocked().await
    }

    pub async fn reset_to_latest(&self) -> Res<ManifestUri> {
        self.package.reset_to_latest_unlocked().await
    }

    pub async fn undo_commit(&self) -> Res<CommitState> {
        self.package.undo_commit_unlocked().await
    }

    pub async fn set_remote(
        &self,
        bucket: String,
        origin: Option<Host>,
        workflow: WorkflowIntent,
    ) -> Res<SetRemoteOutcome> {
        self.package
            .set_remote_unlocked(bucket, origin, workflow)
            .await
    }
}

#[cfg(test)]
mod current_revision_tests;
#[cfg(test)]
mod older_revision_tests;
#[cfg(test)]
mod ordered_writers_tests;
#[cfg(test)]
mod resolve_comparison_tests;
#[cfg(test)]
mod revision_history_tests;
#[cfg(test)]
mod set_remote_tests;
#[cfg(test)]
mod sync_flow_tests;