harn-hostlib 0.10.139

use std::collections::{BTreeMap, VecDeque};
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::sync::{LazyLock, Mutex};
use std::time::{Duration, Instant, SystemTime};

use sha2::{Digest, Sha256};

use crate::error::HostlibError;

static ARTIFACTS: LazyLock<Mutex<ArtifactRegistry>> =
    LazyLock::new(|| Mutex::new(ArtifactRegistry::default()));
static ACTIVE_ARTIFACT_LEASES: LazyLock<Mutex<BTreeMap<PathBuf, File>>> =
    LazyLock::new(|| Mutex::new(BTreeMap::new()));
static LAST_RETENTION_SWEEP: LazyLock<Mutex<Option<Instant>>> = LazyLock::new(|| Mutex::new(None));
/// One descriptor per artifact namespace, held for the life of the process,
/// standing in for the one this process used to hold per completed command.
/// In production a process uses a single namespace, so this is one descriptor.
static SESSION_LEASES: LazyLock<Mutex<BTreeMap<PathBuf, File>>> =
    LazyLock::new(|| Mutex::new(BTreeMap::new()));

const RETENTION_ENV: &str = "HARN_COMMAND_ARTIFACT_RETENTION_SECS";
const MAX_DIRS_ENV: &str = "HARN_COMMAND_ARTIFACT_MAX_DIRS";
const DEFAULT_RETENTION: Duration = Duration::from_hours(168);
// Completed registrations are bounded per process, so N live processes may
// retain up to N * this value. The shared sweep removes oldest unleased
// directories under pressure; it never revokes another process's live lease.
const DEFAULT_MAX_DIRS: usize = 512;
const SWEEP_INTERVAL: Duration = Duration::from_hours(1);
const ARTIFACT_PREFIX: &str = "harn-command-cmd_";
const ACTIVE_LEASE_FILE: &str = ".active.lock";
const SESSION_LEASE_PREFIX: &str = ".session-";
const ARTIFACT_NAMESPACE_PREFIX: &str = "harn-command-artifacts";
const LEGACY_NAMESPACE_LEASE_PREFIX: &str = ".harn-command-artifacts";
const NAMESPACE_LEASE_FILE: &str = ".namespace.lock";
const RUN_COMMAND_BUILTIN: &str = "hostlib_tools_run_command";
const READ_COMMAND_OUTPUT_BUILTIN: &str = "hostlib_tools_read_command_output";
// Command IDs are unique. Contention therefore indicates a stale process or
// an identity collision, not useful work whose duration should be inherited.
const ACTIVE_LEASE_LOCK_TIMEOUT: Duration = Duration::from_secs(30);

#[derive(Clone, Debug)]
struct ArtifactDir {
    path: PathBuf,
    modified: SystemTime,
}

#[derive(Default)]
struct ArtifactRegistry {
    by_id: BTreeMap<String, CommandArtifacts>,
    // Completed results remain leased and repeat-readable while registered.
    // Count or age retirement removes every alias before releasing the lease.
    completed: VecDeque<CompletedArtifact>,
}

struct CompletedArtifact {
    path: PathBuf,
    completed_at: SystemTime,
}

#[derive(Clone, Copy)]
enum ArtifactLeaseCleanup {
    NamespaceHeld,
    LeaseOnly,
}

/// Releases a newly acquired active lease unless ownership is transferred to
/// the completed or live artifact registry. Drop deliberately does not take
/// the namespace lock: it is the last-resort cleanup when that lock caused the
/// registration failure in the first place.
struct ActiveArtifactLeaseGuard {
    dir: Option<PathBuf>,
}

impl ActiveArtifactLeaseGuard {
    fn new(artifacts: &CommandArtifacts) -> Self {
        Self {
            dir: artifact_dir(artifacts),
        }
    }

    fn keep_registered(mut self) {
        self.dir = None;
    }
}

impl Drop for ActiveArtifactLeaseGuard {
    fn drop(&mut self) {
        if let Some(dir) = self.dir.take() {
            release_artifact_lease(&dir);
        }
    }
}

#[derive(Clone, Debug)]
pub(crate) struct CommandArtifacts {
    pub(crate) output_path: PathBuf,
    pub(crate) stdout_path: PathBuf,
    pub(crate) stderr_path: PathBuf,
    pub(crate) line_count: u64,
    pub(crate) byte_count: u64,
    pub(crate) output_sha256: String,
}

pub(crate) struct CommandArtifactRead {
    pub(crate) path: PathBuf,
    pub(crate) offset: u64,
    pub(crate) bytes: Vec<u8>,
    pub(crate) total_bytes: u64,
}

pub(crate) fn persist_artifacts(
    command_id: &str,
    stdout: &[u8],
    stderr: &[u8],
    handle_id: Option<&str>,
) -> Result<CommandArtifacts, HostlibError> {
    let artifacts = planned_artifact_paths(command_id);
    create_and_mark_artifacts_active(&artifacts)?;
    let active_lease = ActiveArtifactLeaseGuard::new(&artifacts);
    let persisted = (|| -> Result<CommandArtifacts, HostlibError> {
        std::fs::write(&artifacts.stdout_path, stdout).map_err(|e| HostlibError::Backend {
            builtin: "hostlib_tools_run_command",
            message: format!("failed to write stdout artifact: {e}"),
        })?;
        std::fs::write(&artifacts.stderr_path, stderr).map_err(|e| HostlibError::Backend {
            builtin: "hostlib_tools_run_command",
            message: format!("failed to write stderr artifact: {e}"),
        })?;
        let mut combined = Vec::with_capacity(stdout.len() + stderr.len());
        combined.extend_from_slice(stdout);
        combined.extend_from_slice(stderr);
        std::fs::write(&artifacts.output_path, &combined).map_err(|e| HostlibError::Backend {
            builtin: "hostlib_tools_run_command",
            message: format!("failed to write combined output artifact: {e}"),
        })?;
        Ok(CommandArtifacts {
            output_path: artifacts.output_path.clone(),
            stdout_path: artifacts.stdout_path.clone(),
            stderr_path: artifacts.stderr_path.clone(),
            line_count: crate::text::count_lines(&combined),
            byte_count: combined.len() as u64,
            output_sha256: format!("sha256:{}", hex::encode(Sha256::digest(&combined))),
        })
    })();
    let artifacts = persisted?;
    register_completed_artifacts_with_guard(command_id, handle_id, &artifacts, active_lease)?;
    let current_dir = artifact_dir(&artifacts);
    maybe_sweep_stale_artifacts(current_dir.as_deref());
    Ok(artifacts)
}

pub(crate) fn register_live_artifacts(
    command_id: &str,
    handle_id: Option<&str>,
) -> Result<CommandArtifacts, HostlibError> {
    let artifacts = planned_artifact_paths(command_id);
    create_and_mark_artifacts_active(&artifacts)?;
    let active_lease = ActiveArtifactLeaseGuard::new(&artifacts);
    let created = (|| -> Result<(), HostlibError> {
        std::fs::File::create(&artifacts.stdout_path).map_err(|e| HostlibError::Backend {
            builtin: "hostlib_tools_run_command",
            message: format!("failed to create stdout artifact: {e}"),
        })?;
        std::fs::File::create(&artifacts.stderr_path).map_err(|e| HostlibError::Backend {
            builtin: "hostlib_tools_run_command",
            message: format!("failed to create stderr artifact: {e}"),
        })?;
        std::fs::File::create(&artifacts.output_path).map_err(|e| HostlibError::Backend {
            builtin: "hostlib_tools_run_command",
            message: format!("failed to create combined output artifact: {e}"),
        })?;
        Ok(())
    })();
    created?;
    register_artifacts(command_id, handle_id, &artifacts);
    active_lease.keep_registered();
    let current_dir = artifact_dir(&artifacts);
    maybe_sweep_stale_artifacts(current_dir.as_deref());
    Ok(artifacts)
}

pub(crate) fn planned_artifact_paths(command_id: &str) -> CommandArtifacts {
    let dir = command_artifact_root().join(format!("harn-command-{command_id}"));
    CommandArtifacts {
        output_path: dir.join("combined.txt"),
        stdout_path: dir.join("stdout.txt"),
        stderr_path: dir.join("stderr.txt"),
        line_count: 0,
        byte_count: 0,
        output_sha256: String::new(),
    }
}

/// Build the terminal artifact metadata from captured bytes when persistence
/// is unavailable. The command result still needs a complete, truthful
/// terminal shape so waiters are not left without a result merely because
/// artifact storage had a transient failure.
pub(crate) fn summarize_artifacts(
    command_id: &str,
    stdout: &[u8],
    stderr: &[u8],
    handle_id: Option<&str>,
) -> CommandArtifacts {
    let mut combined = Vec::with_capacity(stdout.len() + stderr.len());
    combined.extend_from_slice(stdout);
    combined.extend_from_slice(stderr);
    let artifacts = CommandArtifacts {
        line_count: crate::text::count_lines(&combined),
        byte_count: combined.len() as u64,
        output_sha256: format!("sha256:{}", hex::encode(Sha256::digest(&combined))),
        ..planned_artifact_paths(command_id)
    };
    register_fallback_artifacts(command_id, handle_id, &artifacts);
    artifacts
}

fn resolve_output_path(command_id: Option<&str>, handle_id: Option<&str>) -> Option<PathBuf> {
    let artifacts = ARTIFACTS.lock().expect("command artifact store poisoned");
    command_id
        .and_then(|id| artifacts.by_id.get(id))
        .or_else(|| handle_id.and_then(|id| artifacts.by_id.get(id)))
        .map(|a| a.output_path.clone())
}

pub(crate) fn read_output(
    command_id: Option<&str>,
    handle_id: Option<&str>,
    path: Option<&Path>,
    offset: u64,
    length: u64,
) -> Result<Option<CommandArtifactRead>, HostlibError> {
    let explicit = path
        .map(|path| {
            let namespace = command_artifact_namespace(path).ok_or_else(invalid_artifact_path)?;
            let artifact_dir = path.parent().ok_or_else(invalid_artifact_path)?;
            let file_is_regular = std::fs::symlink_metadata(path)
                .map(|metadata| metadata.file_type().is_file())
                .unwrap_or(false);
            let directory_is_real = std::fs::symlink_metadata(artifact_dir)
                .map(|metadata| metadata.file_type().is_dir())
                .unwrap_or(false);
            let namespace_is_established =
                std::fs::symlink_metadata(artifact_namespace_lease_path(&namespace))
                    .map(|metadata| metadata.file_type().is_file())
                    .unwrap_or(false);
            if !file_is_regular || !directory_is_real || !namespace_is_established {
                return Err(invalid_artifact_path());
            }
            Ok((path.to_path_buf(), namespace))
        })
        .transpose()?;
    let candidate = explicit
        .as_ref()
        .map(|(path, _)| path.clone())
        .or_else(|| resolve_output_path(command_id, handle_id));
    let Some(candidate) = candidate else {
        return Ok(None);
    };
    let namespace = explicit
        .as_ref()
        .map(|(_, namespace)| namespace.clone())
        .or_else(|| command_artifact_namespace(&candidate))
        .ok_or_else(|| HostlibError::Backend {
            builtin: READ_COMMAND_OUTPUT_BUILTIN,
            message: "registered command output has an invalid artifact path".to_string(),
        })?;
    with_artifact_namespace_lock(
        &namespace,
        ACTIVE_LEASE_LOCK_TIMEOUT,
        READ_COMMAND_OUTPUT_BUILTIN,
        NamespaceLockCreation::ExistingOnly,
        || {
            let path = if explicit.is_some() {
                candidate.clone()
            } else {
                let Some(resolved) = resolve_output_path(command_id, handle_id) else {
                    return Ok(None);
                };
                if resolved != candidate {
                    return Ok(None);
                }
                resolved
            };
            let mut file = File::open(&path).map_err(|error| HostlibError::Backend {
                builtin: READ_COMMAND_OUTPUT_BUILTIN,
                message: format!(
                    "failed to open command output '{}': {error}",
                    path.display()
                ),
            })?;
            let total_bytes = file.metadata().map(|metadata| metadata.len()).unwrap_or(0);
            file.seek(SeekFrom::Start(offset))
                .map_err(|error| HostlibError::Backend {
                    builtin: READ_COMMAND_OUTPUT_BUILTIN,
                    message: format!(
                        "failed to seek command output '{}': {error}",
                        path.display()
                    ),
                })?;
            let mut bytes = vec![
                0_u8;
                usize::try_from(length)
                    .unwrap_or(usize::MAX)
                    .min(1024 * 1024)
            ];
            let bytes_read = file
                .read(&mut bytes)
                .map_err(|error| HostlibError::Backend {
                    builtin: READ_COMMAND_OUTPUT_BUILTIN,
                    message: format!(
                        "failed to read command output '{}': {error}",
                        path.display()
                    ),
                })?;
            bytes.truncate(bytes_read);
            Ok(Some(CommandArtifactRead {
                path,
                offset,
                bytes,
                total_bytes,
            }))
        },
    )
}

fn invalid_artifact_path() -> HostlibError {
    HostlibError::InvalidParameter {
        builtin: READ_COMMAND_OUTPUT_BUILTIN,
        param: "path",
        message: "path must point at an existing harn-command artifact file".to_string(),
    }
}

fn command_artifact_namespace(path: &Path) -> Option<PathBuf> {
    let file_name = path.file_name()?.to_str()?;
    if !matches!(file_name, "combined.txt" | "stdout.txt" | "stderr.txt") {
        return None;
    }
    let artifact_dir = path.parent()?;
    parse_command_artifact_dir_name(artifact_dir.file_name()?.to_str()?)?;
    artifact_dir.parent().map(Path::to_path_buf)
}

pub(crate) fn live_artifact_snapshot(
    command_id: Option<&str>,
    handle_id: Option<&str>,
) -> Option<CommandArtifacts> {
    let mut artifacts = lookup_artifacts(command_id, handle_id)?;
    artifacts.byte_count = std::fs::metadata(&artifacts.output_path)
        .map(|metadata| metadata.len())
        .unwrap_or(0);
    Some(artifacts)
}

pub(crate) fn live_artifact_tail(
    command_id: Option<&str>,
    handle_id: Option<&str>,
    max_bytes: u64,
) -> Option<String> {
    let artifacts = lookup_artifacts(command_id, handle_id)?;
    let mut file = std::fs::File::open(&artifacts.output_path).ok()?;
    let len = file.metadata().ok()?.len();
    let offset = len.saturating_sub(max_bytes);
    file.seek(SeekFrom::Start(offset)).ok()?;

    let mut bytes = Vec::with_capacity(len.saturating_sub(offset) as usize);
    file.take(max_bytes).read_to_end(&mut bytes).ok()?;
    Some(String::from_utf8_lossy(&bytes).into_owned())
}

fn register_artifacts(command_id: &str, handle_id: Option<&str>, artifacts: &CommandArtifacts) {
    let mut store = ARTIFACTS.lock().expect("command artifact store poisoned");
    register_artifact_aliases(&mut store, command_id, handle_id, artifacts);
}

fn register_artifact_aliases(
    store: &mut ArtifactRegistry,
    command_id: &str,
    handle_id: Option<&str>,
    artifacts: &CommandArtifacts,
) {
    store
        .by_id
        .insert(command_id.to_string(), artifacts.clone());
    if let Some(handle_id) = handle_id {
        store.by_id.insert(handle_id.to_string(), artifacts.clone());
    }
}

fn register_fallback_artifacts(
    command_id: &str,
    handle_id: Option<&str>,
    artifacts: &CommandArtifacts,
) {
    let mut store = ARTIFACTS.lock().expect("command artifact store poisoned");
    register_completed_artifacts_in_store(
        &mut store,
        command_id,
        handle_id,
        artifacts,
        max_artifact_dirs(),
        ArtifactLeaseCleanup::LeaseOnly,
    );
}

fn register_completed_artifacts_in_store(
    store: &mut ArtifactRegistry,
    command_id: &str,
    handle_id: Option<&str>,
    artifacts: &CommandArtifacts,
    max_dirs: usize,
    lease_cleanup: ArtifactLeaseCleanup,
) {
    register_artifact_aliases(store, command_id, handle_id, artifacts);
    let Some(dir) = artifact_dir(artifacts) else {
        return;
    };
    if !store.completed.iter().any(|entry| entry.path == dir) {
        store.completed.push_back(CompletedArtifact {
            path: dir,
            completed_at: SystemTime::now(),
        });
    }
    retire_completed_artifacts(store, max_dirs, None, lease_cleanup);
}

fn register_completed_artifacts_with_guard(
    command_id: &str,
    handle_id: Option<&str>,
    artifacts: &CommandArtifacts,
    active_lease: ActiveArtifactLeaseGuard,
) -> Result<(), HostlibError> {
    register_completed_artifacts_with_guard_options(
        command_id,
        handle_id,
        artifacts,
        active_lease,
        max_artifact_dirs(),
        ACTIVE_LEASE_LOCK_TIMEOUT,
    )
}

fn register_completed_artifacts_with_guard_options(
    command_id: &str,
    handle_id: Option<&str>,
    artifacts: &CommandArtifacts,
    active_lease: ActiveArtifactLeaseGuard,
    max_dirs: usize,
    timeout: Duration,
) -> Result<(), HostlibError> {
    register_completed_artifacts_with_options(command_id, handle_id, artifacts, max_dirs, timeout)?;
    active_lease.keep_registered();
    Ok(())
}

fn register_completed_artifacts_with_options(
    command_id: &str,
    handle_id: Option<&str>,
    artifacts: &CommandArtifacts,
    max_dirs: usize,
    timeout: Duration,
) -> Result<(), HostlibError> {
    let Some(dir) = artifact_dir(artifacts) else {
        register_artifacts(command_id, handle_id, artifacts);
        return Ok(());
    };
    let temp_dir = dir.parent().unwrap_or_else(|| Path::new("."));
    with_artifact_namespace_lock(
        temp_dir,
        timeout,
        RUN_COMMAND_BUILTIN,
        NamespaceLockCreation::Create,
        || {
            let mut store = ARTIFACTS.lock().expect("command artifact store poisoned");
            register_completed_artifacts_in_store(
                &mut store,
                command_id,
                handle_id,
                artifacts,
                max_dirs,
                ArtifactLeaseCleanup::NamespaceHeld,
            );
            Ok(())
        },
    )?;
    // The active lease answers one question: is a command running in this
    // directory right now. A completed one is not, and holding its descriptor
    // until the directory is retired spends one per command for the life of
    // the session. That is how a long session ran out of descriptors: the
    // retention cap is measured in directories, 512 of them by default, while
    // the cost was paid in descriptors against a per-process limit of 256, so
    // the process died of exhaustion before retention ever trimmed anything.
    //
    // Nothing is weakened by releasing here. The sweep skips any directory
    // younger than the retention window before it ever consults the lease, so
    // a completed artifact is protected by that window, which is the policy
    // that is supposed to govern it. The lease still protects a command that
    // is genuinely running, whose directory can go stale while it works, and
    // the marker file stays on disk until the artifact is retired.
    release_artifact_lease(&dir);
    Ok(())
}

fn retire_completed_artifacts_under_namespace(
    store: &mut ArtifactRegistry,
    max_dirs: usize,
    expired_before: Option<SystemTime>,
) {
    retire_completed_artifacts(
        store,
        max_dirs,
        expired_before,
        ArtifactLeaseCleanup::NamespaceHeld,
    );
}

fn retire_completed_artifacts(
    store: &mut ArtifactRegistry,
    max_dirs: usize,
    expired_before: Option<SystemTime>,
    lease_cleanup: ArtifactLeaseCleanup,
) {
    loop {
        let over_limit = max_dirs != 0 && store.completed.len() > max_dirs;
        let expired = expired_before
            .zip(store.completed.front())
            .is_some_and(|(cutoff, artifact)| artifact.completed_at <= cutoff);
        if !over_limit && !expired {
            break;
        }
        let Some(retired) = store.completed.pop_front() else {
            break;
        };
        store
            .by_id
            .retain(|_, artifacts| artifact_dir(artifacts).as_ref() != Some(&retired.path));
        match lease_cleanup {
            ArtifactLeaseCleanup::NamespaceHeld => {
                mark_artifact_dir_inactive_under_namespace(&retired.path);
            }
            ArtifactLeaseCleanup::LeaseOnly => {
                // A persistence failure can itself be caused by namespace-lock
                // contention. Bound the process registry immediately and leave
                // the unlocked marker for a later namespace sweep to remove.
                release_artifact_lease(&retired.path);
            }
        }
    }
}

fn artifact_dir(artifacts: &CommandArtifacts) -> Option<PathBuf> {
    artifacts.output_path.parent().map(Path::to_path_buf)
}

fn create_and_mark_artifacts_active(artifacts: &CommandArtifacts) -> Result<(), HostlibError> {
    create_and_mark_artifacts_active_with_timeout(artifacts, ACTIVE_LEASE_LOCK_TIMEOUT)
}

fn create_and_mark_artifacts_active_with_timeout(
    artifacts: &CommandArtifacts,
    timeout: Duration,
) -> Result<(), HostlibError> {
    let Some(dir) = artifact_dir(artifacts) else {
        return Ok(());
    };
    let temp_dir = dir.parent().unwrap_or_else(|| Path::new("."));
    ensure_artifact_namespace(temp_dir)?;
    with_artifact_namespace_lock(
        temp_dir,
        timeout,
        RUN_COMMAND_BUILTIN,
        NamespaceLockCreation::Create,
        || {
            std::fs::create_dir_all(&dir).map_err(|error| HostlibError::Backend {
                builtin: "hostlib_tools_run_command",
                message: format!("failed to create command artifact dir: {error}"),
            })?;
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                let _ = std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700));
            }
            mark_artifacts_active_under_namespace(artifacts, timeout)
        },
    )
}

#[cfg(test)]
fn mark_artifacts_active(artifacts: &CommandArtifacts) -> Result<(), HostlibError> {
    mark_artifacts_active_with_timeout(artifacts, ACTIVE_LEASE_LOCK_TIMEOUT)
}

#[cfg(test)]
fn mark_artifacts_active_with_timeout(
    artifacts: &CommandArtifacts,
    timeout: Duration,
) -> Result<(), HostlibError> {
    let Some(dir) = artifact_dir(artifacts) else {
        return Ok(());
    };
    let temp_dir = dir.parent().unwrap_or_else(|| Path::new("."));
    with_artifact_namespace_lock(
        temp_dir,
        timeout,
        RUN_COMMAND_BUILTIN,
        NamespaceLockCreation::Create,
        || mark_artifacts_active_under_namespace(artifacts, timeout),
    )
}

fn mark_artifacts_active_under_namespace(
    artifacts: &CommandArtifacts,
    timeout: Duration,
) -> Result<(), HostlibError> {
    let Some(dir) = artifact_dir(artifacts) else {
        return Ok(());
    };
    let mut active_leases = ACTIVE_ARTIFACT_LEASES
        .lock()
        .expect("active command artifact lease store poisoned");
    if active_leases.contains_key(&dir) {
        return Ok(());
    }
    if let Some(namespace) = dir.parent() {
        hold_session_lease(namespace);
    }
    let lease_path = dir.join(ACTIVE_LEASE_FILE);
    let lease = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&lease_path)
        .map_err(|error| HostlibError::Backend {
            builtin: "hostlib_tools_run_command",
            message: format!("failed to open command artifact lease: {error}"),
        })?;
    harn_flock::lock_with_deadline(
        &lease,
        &lease_path,
        harn_flock::LockMode::Exclusive,
        timeout,
    )
    .map_err(|error| HostlibError::Backend {
        builtin: "hostlib_tools_run_command",
        message: format!("failed to lock command artifact lease: {error}"),
    })?;
    active_leases.insert(dir, lease);
    Ok(())
}

#[cfg(test)]
fn mark_artifacts_inactive(artifacts: &CommandArtifacts) {
    if let Some(dir) = artifact_dir(artifacts) {
        let temp_dir = dir.parent().unwrap_or_else(|| Path::new("."));
        let _ = with_artifact_namespace_lock(
            temp_dir,
            ACTIVE_LEASE_LOCK_TIMEOUT,
            RUN_COMMAND_BUILTIN,
            NamespaceLockCreation::Create,
            || {
                mark_artifact_dir_inactive_under_namespace(&dir);
                Ok(())
            },
        );
    }
}

fn mark_artifact_dir_inactive_under_namespace(dir: &Path) {
    release_artifact_lease(dir);
    let _ = std::fs::remove_file(dir.join(ACTIVE_LEASE_FILE));
}

fn release_artifact_lease(dir: &Path) {
    if let Some(lease) = ACTIVE_ARTIFACT_LEASES
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .remove(dir)
    {
        let _ = lease.unlock();
    }
}

fn with_artifact_namespace_lock<T>(
    temp_dir: &Path,
    timeout: Duration,
    builtin: &'static str,
    creation: NamespaceLockCreation,
    operation: impl FnOnce() -> Result<T, HostlibError>,
) -> Result<T, HostlibError> {
    let lease_path = artifact_namespace_lease_path(temp_dir);
    let mut options = OpenOptions::new();
    options.read(true).write(true).truncate(false);
    if matches!(creation, NamespaceLockCreation::Create) {
        options.create(true);
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(0o600);
    }
    let lease = options
        .open(&lease_path)
        .map_err(|error| HostlibError::Backend {
            builtin,
            message: format!("failed to open command artifact namespace lease: {error}"),
        })?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        if lease
            .metadata()
            .map(|metadata| metadata.uid() != unsafe { libc::geteuid() })
            .unwrap_or(true)
        {
            return Err(HostlibError::Backend {
                builtin,
                message: "command artifact namespace lease is not owned by the current user"
                    .to_string(),
            });
        }
    }
    harn_flock::lock_with_deadline(
        &lease,
        &lease_path,
        harn_flock::LockMode::Exclusive,
        timeout,
    )
    .map_err(|error| HostlibError::Backend {
        builtin,
        message: format!("failed to lock command artifact namespace: {error}"),
    })?;
    let result = operation();
    let _ = lease.unlock();
    result
}

#[derive(Clone, Copy)]
enum NamespaceLockCreation {
    Create,
    ExistingOnly,
}

fn artifact_namespace_lease_path(temp_dir: &Path) -> PathBuf {
    if temp_dir.file_name().and_then(|name| name.to_str())
        == Some(artifact_namespace_dir_name().as_str())
    {
        return temp_dir.join(NAMESPACE_LEASE_FILE);
    }
    temp_dir.join(format!(
        "{LEGACY_NAMESPACE_LEASE_PREFIX}-{}.lock",
        artifact_owner_suffix()
    ))
}

fn artifact_owner_suffix() -> String {
    #[cfg(unix)]
    let suffix = unsafe { libc::geteuid() }.to_string();
    #[cfg(not(unix))]
    let suffix = "user".to_string();
    suffix
}

fn artifact_namespace_dir_name() -> String {
    format!("{ARTIFACT_NAMESPACE_PREFIX}-{}", artifact_owner_suffix())
}

fn command_artifact_root() -> PathBuf {
    std::env::temp_dir().join(artifact_namespace_dir_name())
}

fn ensure_artifact_namespace(namespace: &Path) -> Result<(), HostlibError> {
    match std::fs::create_dir(namespace) {
        Ok(()) => {}
        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
        Err(error) => {
            return Err(HostlibError::Backend {
                builtin: RUN_COMMAND_BUILTIN,
                message: format!("failed to create command artifact namespace: {error}"),
            });
        }
    }
    let metadata = std::fs::symlink_metadata(namespace).map_err(|error| HostlibError::Backend {
        builtin: RUN_COMMAND_BUILTIN,
        message: format!("failed to inspect command artifact namespace: {error}"),
    })?;
    if !metadata.file_type().is_dir() {
        return Err(HostlibError::Backend {
            builtin: RUN_COMMAND_BUILTIN,
            message: "command artifact namespace must be a real directory".to_string(),
        });
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::{MetadataExt, PermissionsExt};
        if metadata.uid() != unsafe { libc::geteuid() } {
            return Err(HostlibError::Backend {
                builtin: RUN_COMMAND_BUILTIN,
                message: "command artifact namespace is not owned by the current user".to_string(),
            });
        }
        std::fs::set_permissions(namespace, std::fs::Permissions::from_mode(0o700)).map_err(
            |error| HostlibError::Backend {
                builtin: RUN_COMMAND_BUILTIN,
                message: format!("failed to secure command artifact namespace: {error}"),
            },
        )?;
    }
    Ok(())
}

fn lookup_artifacts(command_id: Option<&str>, handle_id: Option<&str>) -> Option<CommandArtifacts> {
    let store = ARTIFACTS.lock().expect("command artifact store poisoned");
    command_id
        .and_then(|id| store.by_id.get(id))
        .or_else(|| handle_id.and_then(|id| store.by_id.get(id)))
        .cloned()
}

fn maybe_sweep_stale_artifacts(current_dir: Option<&Path>) {
    let Some(retention) = retention_duration() else {
        return;
    };
    let temp_dir = command_artifact_root();
    let max_dirs = max_artifact_dirs();
    let now = Instant::now();
    {
        let mut last = LAST_RETENTION_SWEEP
            .lock()
            .expect("command artifact retention state poisoned");
        if last
            .map(|last_run| now.duration_since(last_run) < SWEEP_INTERVAL)
            .unwrap_or(false)
            && !command_artifact_dir_count_exceeds(&temp_dir, max_dirs)
        {
            return;
        }
        *last = Some(now);
    }
    let _ = with_artifact_namespace_lock(
        &temp_dir,
        ACTIVE_LEASE_LOCK_TIMEOUT,
        RUN_COMMAND_BUILTIN,
        NamespaceLockCreation::Create,
        || {
            let expired_before = SystemTime::now().checked_sub(retention);
            retire_completed_artifacts_under_namespace(
                &mut ARTIFACTS.lock().expect("command artifact store poisoned"),
                max_dirs,
                expired_before,
            );
            sweep_command_artifact_dirs_except(
                &temp_dir,
                retention,
                max_dirs,
                SystemTime::now(),
                current_dir,
            );
            Ok(())
        },
    );
}

fn retention_duration() -> Option<Duration> {
    let secs = std::env::var(RETENTION_ENV)
        .ok()
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(DEFAULT_RETENTION.as_secs());
    if secs == 0 {
        None
    } else {
        Some(Duration::from_secs(secs))
    }
}

fn max_artifact_dirs() -> usize {
    std::env::var(MAX_DIRS_ENV)
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .unwrap_or(DEFAULT_MAX_DIRS)
}

fn command_artifact_dir_count_exceeds(temp_dir: &Path, max_dirs: usize) -> bool {
    if max_dirs == 0 {
        return false;
    }
    let Ok(entries) = std::fs::read_dir(temp_dir) else {
        return false;
    };
    let mut count = 0;
    for entry in entries.flatten() {
        let path = entry.path();
        let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        if parse_command_artifact_dir_name(name).is_none() {
            continue;
        }
        let Ok(metadata) = std::fs::symlink_metadata(&path) else {
            continue;
        };
        if !metadata.file_type().is_dir() {
            continue;
        }
        count += 1;
        if count > max_dirs {
            return true;
        }
    }
    false
}

fn collect_command_artifact_dirs(temp_dir: &Path) -> Vec<ArtifactDir> {
    let Ok(entries) = std::fs::read_dir(temp_dir) else {
        return Vec::new();
    };
    let mut dirs = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        if parse_command_artifact_dir_name(name).is_none() {
            continue;
        }
        let Ok(metadata) = std::fs::symlink_metadata(&path) else {
            continue;
        };
        if !metadata.file_type().is_dir() {
            continue;
        }
        let Ok(modified) = metadata.modified() else {
            continue;
        };
        dirs.push(ArtifactDir { path, modified });
    }
    dirs
}

#[cfg(test)]
fn sweep_command_artifact_dirs(
    temp_dir: &Path,
    retention: Duration,
    max_dirs: usize,
    now: SystemTime,
) {
    sweep_command_artifact_dirs_except(temp_dir, retention, max_dirs, now, None);
}

fn sweep_command_artifact_dirs_except(
    temp_dir: &Path,
    retention: Duration,
    max_dirs: usize,
    now: SystemTime,
    current_dir: Option<&Path>,
) {
    sweep_stale_session_leases(temp_dir);
    let mut dirs = collect_command_artifact_dirs(temp_dir);
    dirs.sort_by_key(|dir| dir.modified);
    let mut live_count = dirs.len();
    for dir in &dirs {
        if current_dir == Some(dir.path.as_path()) {
            continue;
        }
        if now
            .duration_since(dir.modified)
            .map(|age| age < retention)
            .unwrap_or(true)
        {
            continue;
        }
        if should_preserve_artifact_dir(dir) {
            continue;
        }
        if remove_artifact_dir(&dir.path) {
            live_count = live_count.saturating_sub(1);
        }
    }
    if max_dirs == 0 || live_count <= max_dirs {
        return;
    }
    for dir in &dirs {
        if live_count <= max_dirs {
            break;
        }
        if !dir.path.exists()
            || current_dir == Some(dir.path.as_path())
            || should_preserve_artifact_dir(dir)
        {
            continue;
        }
        if remove_artifact_dir(&dir.path) {
            live_count = live_count.saturating_sub(1);
        }
    }
}

/// The session lease a process holds while it can still serve its own
/// completed results.
fn session_lease_path(namespace: &Path, pid: u32) -> PathBuf {
    namespace.join(format!("{SESSION_LEASE_PREFIX}{pid}.lock"))
}

/// Take this process's session lease once, under the namespace lock.
///
/// A completed command's own lock used to be what told another process that
/// its results were still being served, at the cost of one descriptor per
/// command. This says the same thing once, for the whole session.
fn hold_session_lease(namespace: &Path) {
    let mut held = SESSION_LEASES
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if held.contains_key(namespace) {
        return;
    }
    let path = session_lease_path(namespace, std::process::id());
    let Ok(lease) = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&path)
    else {
        return;
    };
    if harn_flock::lock_with_deadline(
        &lease,
        &path,
        harn_flock::LockMode::Exclusive,
        ACTIVE_LEASE_LOCK_TIMEOUT,
    )
    .is_ok()
    {
        held.insert(namespace.to_path_buf(), lease);
    }
}

/// Whether the process that owns `pid`'s artifacts is still serving them.
///
/// Only a process that took a session lease counts. An artifact directory
/// named after some unrelated live process is not claimed by anything and
/// stays evictable, which is what keeps a live stranger from starving fresh
/// output.
fn owner_session_is_live(namespace: &Path, pid: u32) -> bool {
    let path = session_lease_path(namespace, pid);
    let Ok(lease) = OpenOptions::new().read(true).write(true).open(&path) else {
        return false;
    };
    match lease.try_lock() {
        Ok(()) => {
            let _ = lease.unlock();
            false
        }
        // Held, or unreadable: either way, assume the owner is still serving.
        Err(_) => true,
    }
}

/// Remove session leases whose owning process is gone.
///
/// A lease file is one per process, not one per command, but a machine that
/// runs many sessions would still accumulate them forever. Both this and the
/// acquisition run under the namespace lock, so a lease that can be taken here
/// belongs to no live session and is safe to delete.
fn sweep_stale_session_leases(namespace: &Path) {
    let Ok(entries) = std::fs::read_dir(namespace) else {
        return;
    };
    let own = session_lease_path(namespace, std::process::id());
    for entry in entries.flatten() {
        let path = entry.path();
        if path == own {
            continue;
        }
        let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        if !name.starts_with(SESSION_LEASE_PREFIX) || !name.ends_with(".lock") {
            continue;
        }
        let Ok(lease) = OpenOptions::new().read(true).write(true).open(&path) else {
            continue;
        };
        if lease.try_lock().is_ok() {
            let _ = lease.unlock();
            let _ = std::fs::remove_file(&path);
        }
    }
}

/// Whether this process still has the directory in its completed registry.
fn dir_is_registered(path: &Path) -> bool {
    ARTIFACTS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .completed
        .iter()
        .any(|artifact| artifact.path == path)
}

fn should_preserve_artifact_dir(dir: &ArtifactDir) -> bool {
    if ACTIVE_ARTIFACT_LEASES
        .lock()
        .expect("active command artifact lease store poisoned")
        .contains_key(&dir.path)
    {
        return true;
    }
    // A completed result stays readable while its owner is still serving it.
    // This is the guarantee the per-command descriptor used to provide, now
    // answered once per process instead of once per command.
    if let (Some(namespace), Some(name)) = (
        dir.path.parent(),
        dir.path.file_name().and_then(|name| name.to_str()),
    ) {
        if let Some(owner) = parse_command_artifact_dir_name(name) {
            if owner == std::process::id() {
                // Our own directories answer from the registry, so the ones we
                // have already evicted stay evictable.
                if dir_is_registered(&dir.path) {
                    return true;
                }
            } else if owner_session_is_live(namespace, owner) {
                return true;
            }
        }
    }
    let lease_path = dir.path.join(ACTIVE_LEASE_FILE);
    let Ok(metadata) = std::fs::symlink_metadata(&lease_path) else {
        return false;
    };
    if !metadata.file_type().is_file() {
        return false;
    }
    let Ok(lease) = OpenOptions::new().read(true).write(true).open(lease_path) else {
        return true;
    };
    match lease.try_lock() {
        Ok(()) => {
            let _ = lease.unlock();
            false
        }
        // Contended or unreadable: either way, treat the lease as live.
        Err(_) => true,
    }
}

fn remove_artifact_dir(dir: &Path) -> bool {
    if std::fs::remove_dir_all(dir).is_err() {
        return false;
    }
    ARTIFACTS
        .lock()
        .expect("command artifact store poisoned")
        .by_id
        .retain(|_, artifacts| artifact_dir(artifacts).as_deref() != Some(dir));
    true
}

fn parse_command_artifact_dir_name(name: &str) -> Option<u32> {
    let suffix = name.strip_prefix(ARTIFACT_PREFIX)?;
    let mut parts = suffix.split('_');
    let pid = parts.next()?.parse::<u32>().ok()?;
    parts.next()?.parse::<u128>().ok()?;
    parts.next()?.parse::<u64>().ok()?;
    if parts.next().is_some() {
        return None;
    }
    Some(pid)
}

#[cfg(test)]
#[path = "artifacts/tests.rs"]
mod tests;