//! Runtime-owned execution of Backend-resolved Worker retention dispositions.
//!
//! This boundary deliberately accepts only stable ids and resolved dispositions.
//! Host paths and provider handles never cross it.

use crate::error::RuntimeError;
use crate::identity::WorkerId;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::BTreeSet;
use std::fs::{self, File, OpenOptions};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

const ARCHIVE_SCHEMA_VERSION: u32 = 1;
const OPERATION_SCHEMA_VERSION: u32 = 1;
const RETENTION_LOCK: &str = ".worker-retention.lock";
static NEXT_TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(1);

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionDisposition {
    Archive,
    Purge,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DiagnosticsDisposition {
    Purge,
    /// The Backend catalog owns the expiry. Runtime keeps only bounded stdout/stderr
    /// evidence and never mixes it into the Session archive.
    Retain,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerRetentionInventory {
    pub workspace_id: String,
    pub runtime_id: String,
    pub worker_id: WorkerId,
    pub run_generation: u64,
    pub session_id: Option<String>,
    pub segment_ids: Vec<String>,
    pub session_bytes: u64,
    pub diagnostics_bytes: u64,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct RuntimeWorkerAggregateDiagnostic {
    worker_id: String,
    category: String,
    detail: String,
}

impl RuntimeWorkerAggregateDiagnostic {
    pub fn worker_id(&self) -> &str {
        &self.worker_id
    }

    pub fn category(&self) -> &str {
        &self.category
    }

    pub fn detail(&self) -> &str {
        &self.detail
    }
}

/// Opaque host-derived inventory snapshot. Callers can inspect but cannot
/// construct or alter the Runtime/Workspace scope used by reconciliation.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct WorkerRetentionInventorySnapshot {
    workspace_id: String,
    runtime_id: String,
    workers: Vec<WorkerRetentionInventory>,
    diagnostics: Vec<RuntimeWorkerAggregateDiagnostic>,
}

impl WorkerRetentionInventorySnapshot {
    pub(crate) fn new(
        workspace_id: String,
        runtime_id: String,
        workers: Vec<WorkerRetentionInventory>,
        diagnostics: Vec<RuntimeWorkerAggregateDiagnostic>,
    ) -> Self {
        Self {
            workspace_id,
            runtime_id,
            workers,
            diagnostics,
        }
    }

    pub fn workspace_id(&self) -> &str {
        &self.workspace_id
    }

    pub fn runtime_id(&self) -> &str {
        &self.runtime_id
    }

    pub fn workers(&self) -> &[WorkerRetentionInventory] {
        &self.workers
    }

    pub fn diagnostics(&self) -> &[RuntimeWorkerAggregateDiagnostic] {
        &self.diagnostics
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerRetentionExecutionRequest {
    pub operation_id: String,
    pub input_fingerprint: String,
    pub archive_id: Option<String>,
    pub workspace_id: String,
    pub source_runtime_id: String,
    pub worker_id: WorkerId,
    pub expected_worker_revision: String,
    pub expected_run_generation: u64,
    pub source_created_at: String,
    pub removed_at: String,
    pub effective_profile: Option<String>,
    pub retention_class: Option<String>,
    pub policy_id: String,
    pub policy_revision: u64,
    pub session_disposition: SessionDisposition,
    pub diagnostics_disposition: DiagnosticsDisposition,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerSessionArchiveManifest {
    pub schema_version: u32,
    pub archive_id: String,
    pub workspace_id: String,
    pub source_runtime_id: String,
    pub source_worker_id: WorkerId,
    pub source_session_id: String,
    pub segment_ids: Vec<String>,
    pub source_created_at: String,
    pub removed_at: String,
    pub archived_at_unix_seconds: u64,
    pub effective_profile: Option<String>,
    pub retention_class: Option<String>,
    pub content_checksum_sha256: String,
    pub content_bytes: u64,
    pub content_file_count: u64,
    pub policy_id: String,
    pub policy_revision: u64,
    pub operation_id: String,
    pub input_fingerprint: String,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerRetentionExecutionResult {
    pub operation_id: String,
    pub input_fingerprint: String,
    pub expected_worker_revision: String,
    pub worker_id: WorkerId,
    pub session_disposition: SessionDisposition,
    pub diagnostics_disposition: DiagnosticsDisposition,
    pub archive: Option<WorkerSessionArchiveManifest>,
    pub source_removed: bool,
    pub diagnostics_retained: bool,
}

pub(crate) trait WorkerRetentionProvider: Send + Sync {
    fn inventory(
        &self,
        workspace_id: &str,
        runtime_id: &str,
        worker_id: WorkerId,
        run_generation: u64,
    ) -> Result<WorkerRetentionInventory, RuntimeError>;

    fn execute(
        &self,
        request: &WorkerRetentionExecutionRequest,
    ) -> Result<WorkerRetentionExecutionResult, RuntimeError>;
}

/// Filesystem provider for the canonical Runtime Worker aggregate.
#[derive(Clone, Debug)]
pub(crate) struct FsWorkerRetentionProvider {
    runtime_root: PathBuf,
}

impl FsWorkerRetentionProvider {
    pub(crate) fn new(runtime_root: impl Into<PathBuf>) -> Self {
        Self {
            runtime_root: runtime_root.into(),
        }
    }

    pub(crate) fn completed_for(
        &self,
        request: &WorkerRetentionExecutionRequest,
    ) -> Result<Option<WorkerRetentionExecutionResult>, RuntimeError> {
        let path = self.operation_path(&request.operation_id)?;
        let Some(receipt) = read_operation_receipt(&path)? else {
            return Ok(None);
        };
        validate_receipt_request(&receipt, request)?;
        Ok(receipt.result.source_removed.then_some(receipt.result))
    }

    pub(crate) fn snapshot(
        &self,
        workspace_id: &str,
        runtime_id: &str,
    ) -> Result<WorkerRetentionInventorySnapshot, RuntimeError> {
        fs::create_dir_all(&self.runtime_root).map_err(|source| RuntimeError::StoreIo {
            operation: "prepare Worker retention inventory",
            path: self.runtime_root.clone(),
            source,
        })?;
        let lock_path = self.runtime_root.join(RETENTION_LOCK);
        let lock = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .open(&lock_path)
            .map_err(|source| RuntimeError::StoreIo {
                operation: "lock Worker retention inventory",
                path: lock_path.clone(),
                source,
            })?;
        lock.lock().map_err(|source| RuntimeError::StoreIo {
            operation: "lock Worker retention inventory",
            path: lock_path,
            source,
        })?;
        let workers_root = self.runtime_root.join("workers");
        if !workers_root.is_dir() {
            return Ok(WorkerRetentionInventorySnapshot::new(
                workspace_id.to_string(),
                runtime_id.to_string(),
                Vec::new(),
                Vec::new(),
            ));
        }
        let mut entries = fs::read_dir(&workers_root)
            .map_err(|source| RuntimeError::StoreIo {
                operation: "scan Worker retention inventory",
                path: workers_root.clone(),
                source,
            })?
            .collect::<Result<Vec<_>, _>>()
            .map_err(|source| RuntimeError::StoreIo {
                operation: "scan Worker retention inventory",
                path: workers_root.clone(),
                source,
            })?;
        entries.sort_by_key(|entry| entry.file_name());
        let mut workers = Vec::new();
        let mut diagnostics = Vec::new();
        for entry in entries {
            let raw_id = entry.file_name().to_string_lossy().to_string();
            let bounded_id = bounded_diagnostic_id(&raw_id);
            let file_type = match entry.file_type() {
                Ok(file_type) => file_type,
                Err(_) => {
                    diagnostics.push(runtime_aggregate_diagnostic(
                        &bounded_id,
                        "aggregate_metadata_unreadable",
                    ));
                    continue;
                }
            };
            if !file_type.is_dir() || file_type.is_symlink() {
                diagnostics.push(runtime_aggregate_diagnostic(
                    &bounded_id,
                    "aggregate_entry_unsupported",
                ));
                continue;
            }
            let Ok(worker_number) = raw_id.parse::<u64>() else {
                diagnostics.push(runtime_aggregate_diagnostic(
                    &bounded_id,
                    "aggregate_worker_id_invalid",
                ));
                continue;
            };
            let worker_id = WorkerId::new(worker_number);
            let worker_dir = self.worker_dir(worker_id);
            let snapshot: WorkerGenerationSnapshot = match read_json(
                &worker_dir.join("worker.json"),
                "scan Worker retention inventory",
            ) {
                Ok(snapshot) => snapshot,
                Err(_) => {
                    diagnostics.push(runtime_aggregate_diagnostic(
                        &bounded_id,
                        "aggregate_worker_record_corrupt",
                    ));
                    continue;
                }
            };
            if snapshot.workspace_id.as_deref() != Some(workspace_id) {
                diagnostics.push(runtime_aggregate_diagnostic(
                    &bounded_id,
                    "aggregate_workspace_mismatch",
                ));
                continue;
            }
            match self.inventory(workspace_id, runtime_id, worker_id, snapshot.run_generation) {
                Ok(item) => workers.push(item),
                Err(_) => diagnostics.push(runtime_aggregate_diagnostic(
                    &bounded_id,
                    "aggregate_session_inventory_failed",
                )),
            }
        }
        workers.sort_by_key(|item| item.worker_id);
        diagnostics.sort_by(|left, right| {
            (&left.worker_id, &left.category).cmp(&(&right.worker_id, &right.category))
        });
        Ok(WorkerRetentionInventorySnapshot::new(
            workspace_id.to_string(),
            runtime_id.to_string(),
            workers,
            diagnostics,
        ))
    }

    pub(crate) fn recover_after_source_removal(
        &self,
        request: &WorkerRetentionExecutionRequest,
    ) -> Result<WorkerRetentionExecutionResult, RuntimeError> {
        if self.worker_dir(request.worker_id).exists() {
            return Err(RuntimeError::WorkerNotFound {
                worker_id: request.worker_id,
            });
        }
        self.execute(request)
    }

    fn worker_dir(&self, worker_id: WorkerId) -> PathBuf {
        self.runtime_root
            .join("workers")
            .join(worker_id.to_string())
    }

    fn operation_path(&self, operation_id: &str) -> Result<PathBuf, RuntimeError> {
        validate_id("operation_id", operation_id)?;
        Ok(self
            .runtime_root
            .join("retention")
            .join("operations")
            .join(format!("{operation_id}.json")))
    }

    fn archive_dir(&self, archive_id: &str) -> Result<PathBuf, RuntimeError> {
        validate_id("archive_id", archive_id)?;
        Ok(self
            .runtime_root
            .join("archives")
            .join("workers")
            .join(archive_id))
    }

    fn diagnostics_dir(&self, operation_id: &str) -> Result<PathBuf, RuntimeError> {
        validate_id("operation_id", operation_id)?;
        Ok(self
            .runtime_root
            .join("archives")
            .join("diagnostics")
            .join(operation_id))
    }
}

impl WorkerRetentionProvider for FsWorkerRetentionProvider {
    fn inventory(
        &self,
        workspace_id: &str,
        runtime_id: &str,
        worker_id: WorkerId,
        run_generation: u64,
    ) -> Result<WorkerRetentionInventory, RuntimeError> {
        let worker_dir = self.worker_dir(worker_id);
        if !worker_dir.is_dir() {
            return Err(RuntimeError::WorkerNotFound { worker_id });
        }
        let worker: WorkerGenerationSnapshot = read_json(
            &worker_dir.join("worker.json"),
            "inventory Worker retention",
        )?;
        if worker.workspace_id.as_deref() != Some(workspace_id) {
            return Err(RuntimeError::WorkerNotFound { worker_id });
        }
        if worker.run_generation != run_generation {
            return Err(RuntimeError::InvalidRequest(format!(
                "Worker retention inventory expected generation {run_generation}, current generation is {}",
                worker.run_generation
            )));
        }
        let session_dir = worker_dir.join("session");
        let (session_id, segment_ids, session_bytes) = if session_dir.is_dir() {
            let manifest: CanonicalSessionManifest = read_json(
                &session_dir.join("session.json"),
                "inventory Worker retention",
            )?;
            let files = collect_files(&session_dir, "inventory Worker retention")?;
            let mut segment_ids = BTreeSet::new();
            let mut bytes = 0_u64;
            for (relative, path) in files {
                bytes = bytes.saturating_add(file_len(&path, "inventory Worker retention")?);
                if let Some(name) = relative.file_name().and_then(|value| value.to_str()) {
                    let segment = name
                        .strip_suffix(".trace.jsonl")
                        .or_else(|| name.strip_suffix(".jsonl"));
                    if let Some(segment) = segment {
                        segment_ids.insert(segment.to_string());
                    }
                }
            }
            (
                Some(manifest.session_id),
                segment_ids.into_iter().collect(),
                bytes,
            )
        } else {
            (None, Vec::new(), 0)
        };
        let diagnostics_bytes = diagnostics_files(&worker_dir, "inventory Worker retention")?
            .into_iter()
            .try_fold(0_u64, |total, path| {
                file_len(&path, "inventory Worker retention").map(|size| total.saturating_add(size))
            })?;
        Ok(WorkerRetentionInventory {
            workspace_id: workspace_id.to_string(),
            runtime_id: runtime_id.to_string(),
            worker_id,
            run_generation,
            session_id,
            segment_ids,
            session_bytes,
            diagnostics_bytes,
        })
    }

    fn execute(
        &self,
        request: &WorkerRetentionExecutionRequest,
    ) -> Result<WorkerRetentionExecutionResult, RuntimeError> {
        validate_request(request)?;
        fs::create_dir_all(&self.runtime_root).map_err(|source| RuntimeError::StoreIo {
            operation: "prepare Worker retention",
            path: self.runtime_root.clone(),
            source,
        })?;
        let lock_path = self.runtime_root.join(RETENTION_LOCK);
        let lock = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .open(&lock_path)
            .map_err(|source| RuntimeError::StoreIo {
                operation: "lock Worker retention",
                path: lock_path.clone(),
                source,
            })?;
        lock.lock().map_err(|source| RuntimeError::StoreIo {
            operation: "lock Worker retention",
            path: lock_path,
            source,
        })?;

        let pending = read_operation_receipt(&self.operation_path(&request.operation_id)?)?;
        if let Some(receipt) = &pending {
            validate_receipt_request(receipt, request)?;
            if receipt.result.source_removed {
                return Ok(receipt.result.clone());
            }
        }

        let worker_dir = self.worker_dir(request.worker_id);
        if !worker_dir.is_dir() {
            if let Some(mut receipt) = pending {
                // A prior attempt durably committed all disposition evidence and
                // removed the aggregate, then stopped before finalizing its receipt.
                receipt.result.source_removed = true;
                atomic_write_json(
                    &self.operation_path(&request.operation_id)?,
                    &receipt,
                    "recover Worker retention receipt",
                )?;
                return Ok(receipt.result);
            }
            return Err(RuntimeError::WorkerNotFound {
                worker_id: request.worker_id,
            });
        }
        let snapshot: WorkerGenerationSnapshot =
            read_json(&worker_dir.join("worker.json"), "execute Worker retention")?;
        if snapshot.workspace_id.as_deref() != Some(request.workspace_id.as_str()) {
            return Err(RuntimeError::WorkerNotFound {
                worker_id: request.worker_id,
            });
        }
        if snapshot.run_generation != request.expected_run_generation {
            return Err(RuntimeError::InvalidRequest(format!(
                "Worker retention plan expected generation {}, current generation is {}",
                request.expected_run_generation, snapshot.run_generation
            )));
        }

        let archive = match request.session_disposition {
            SessionDisposition::Archive => {
                Some(commit_session_archive(self, request, &worker_dir)?)
            }
            SessionDisposition::Purge => None,
        };
        let diagnostics_retained = match request.diagnostics_disposition {
            DiagnosticsDisposition::Purge => false,
            DiagnosticsDisposition::Retain => {
                commit_diagnostics_archive(self, request, &worker_dir)?;
                true
            }
        };

        let mut result = WorkerRetentionExecutionResult {
            operation_id: request.operation_id.clone(),
            input_fingerprint: request.input_fingerprint.clone(),
            expected_worker_revision: request.expected_worker_revision.clone(),
            worker_id: request.worker_id,
            session_disposition: request.session_disposition,
            diagnostics_disposition: request.diagnostics_disposition,
            archive,
            source_removed: false,
            diagnostics_retained,
        };
        let mut receipt = RetentionOperationReceipt {
            schema_version: OPERATION_SCHEMA_VERSION,
            request: request.clone(),
            result: result.clone(),
        };
        // Pending receipt makes the delete/final-receipt crash window
        // recoverable without treating an uncommitted archive as completion.
        atomic_write_json(
            &self.operation_path(&request.operation_id)?,
            &receipt,
            "commit pending Worker retention receipt",
        )?;

        fs::remove_dir_all(&worker_dir).map_err(|source| RuntimeError::StoreIo {
            operation: "remove retained Worker aggregate",
            path: worker_dir.clone(),
            source,
        })?;
        sync_directory(
            worker_dir.parent().unwrap_or(&self.runtime_root),
            "remove retained Worker aggregate",
        )?;

        result.source_removed = true;
        receipt.result = result.clone();
        atomic_write_json(
            &self.operation_path(&request.operation_id)?,
            &receipt,
            "commit Worker retention receipt",
        )?;
        Ok(result)
    }
}

#[derive(Deserialize)]
struct WorkerGenerationSnapshot {
    #[serde(default)]
    workspace_id: Option<String>,
    #[serde(default)]
    run_generation: u64,
}

#[derive(Deserialize)]
struct CanonicalSessionManifest {
    session_id: String,
}

#[derive(Serialize, Deserialize)]
struct RetentionOperationReceipt {
    schema_version: u32,
    request: WorkerRetentionExecutionRequest,
    result: WorkerRetentionExecutionResult,
}

fn read_operation_receipt(path: &Path) -> Result<Option<RetentionOperationReceipt>, RuntimeError> {
    if !path.is_file() {
        return Ok(None);
    }
    let receipt: RetentionOperationReceipt = read_json(path, "read Worker retention receipt")?;
    if receipt.schema_version != OPERATION_SCHEMA_VERSION {
        return Err(RuntimeError::StoreCorrupt {
            operation: "read Worker retention receipt",
            path: path.to_path_buf(),
            message: format!(
                "unsupported operation receipt schema {}",
                receipt.schema_version
            ),
        });
    }
    Ok(Some(receipt))
}

fn validate_receipt_request(
    receipt: &RetentionOperationReceipt,
    request: &WorkerRetentionExecutionRequest,
) -> Result<(), RuntimeError> {
    if &receipt.request != request
        || receipt.result.operation_id != request.operation_id
        || receipt.result.input_fingerprint != request.input_fingerprint
    {
        return Err(RuntimeError::InvalidRequest(format!(
            "retention operation {} was already used with different input",
            request.operation_id
        )));
    }
    Ok(())
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
struct DiagnosticsArchiveManifest {
    schema_version: u32,
    operation_id: String,
    workspace_id: String,
    source_runtime_id: String,
    source_worker_id: WorkerId,
    input_fingerprint: String,
    content_checksum_sha256: String,
    content_bytes: u64,
    content_file_count: u64,
}

fn runtime_aggregate_diagnostic(
    worker_id: &str,
    category: &str,
) -> RuntimeWorkerAggregateDiagnostic {
    RuntimeWorkerAggregateDiagnostic {
        worker_id: worker_id.to_string(),
        category: category.to_string(),
        detail: "Canonical Runtime Worker aggregate requires diagnostic reconciliation".to_string(),
    }
}

fn bounded_diagnostic_id(value: &str) -> String {
    let sanitized = value
        .chars()
        .take(64)
        .map(|character| {
            if character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.') {
                character
            } else {
                '_'
            }
        })
        .collect::<String>();
    if sanitized.is_empty() {
        "unknown".to_string()
    } else {
        sanitized
    }
}

fn validate_request(request: &WorkerRetentionExecutionRequest) -> Result<(), RuntimeError> {
    validate_id("operation_id", &request.operation_id)?;
    validate_id("workspace_id", &request.workspace_id)?;
    validate_id("source_runtime_id", &request.source_runtime_id)?;
    validate_id("policy_id", &request.policy_id)?;
    if request.input_fingerprint.trim().is_empty() {
        return Err(RuntimeError::InvalidRequest(
            "retention input fingerprint must not be empty".to_string(),
        ));
    }
    match (request.session_disposition, request.archive_id.as_deref()) {
        (SessionDisposition::Archive, Some(id)) => validate_id("archive_id", id),
        (SessionDisposition::Archive, None) => Err(RuntimeError::InvalidRequest(
            "archive disposition requires archive_id".to_string(),
        )),
        (SessionDisposition::Purge, None) => Ok(()),
        (SessionDisposition::Purge, Some(_)) => Err(RuntimeError::InvalidRequest(
            "purge disposition must not include archive_id".to_string(),
        )),
    }
}

fn validate_id(kind: &str, value: &str) -> Result<(), RuntimeError> {
    if value.is_empty()
        || value.len() > 160
        || value == "."
        || value == ".."
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
    {
        return Err(RuntimeError::InvalidRequest(format!(
            "invalid retention {kind}"
        )));
    }
    Ok(())
}

fn commit_session_archive(
    provider: &FsWorkerRetentionProvider,
    request: &WorkerRetentionExecutionRequest,
    worker_dir: &Path,
) -> Result<WorkerSessionArchiveManifest, RuntimeError> {
    let archive_id = request.archive_id.as_deref().ok_or_else(|| {
        RuntimeError::InvalidRequest("archive disposition requires archive_id".to_string())
    })?;
    let session_dir = worker_dir.join("session");
    if !session_dir.is_dir() {
        return Err(RuntimeError::StoreMissing {
            operation: "archive Worker Session",
            path: session_dir,
        });
    }
    let session: CanonicalSessionManifest =
        read_json(&session_dir.join("session.json"), "archive Worker Session")?;
    let source_files = collect_files(&session_dir, "archive Worker Session")?;
    let (checksum, bytes, count) = checksum_files(&source_files, "archive Worker Session")?;
    let mut segment_ids = BTreeSet::new();
    for (relative, _) in &source_files {
        if let Some(name) = relative.file_name().and_then(|name| name.to_str()) {
            if let Some(segment) = name
                .strip_suffix(".trace.jsonl")
                .or_else(|| name.strip_suffix(".jsonl"))
            {
                segment_ids.insert(segment.to_string());
            }
        }
    }
    let manifest = WorkerSessionArchiveManifest {
        schema_version: ARCHIVE_SCHEMA_VERSION,
        archive_id: archive_id.to_string(),
        workspace_id: request.workspace_id.clone(),
        source_runtime_id: request.source_runtime_id.clone(),
        source_worker_id: request.worker_id,
        source_session_id: session.session_id,
        segment_ids: segment_ids.into_iter().collect(),
        source_created_at: request.source_created_at.clone(),
        removed_at: request.removed_at.clone(),
        archived_at_unix_seconds: std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs(),
        effective_profile: request.effective_profile.clone(),
        retention_class: request.retention_class.clone(),
        content_checksum_sha256: checksum,
        content_bytes: bytes,
        content_file_count: count,
        policy_id: request.policy_id.clone(),
        policy_revision: request.policy_revision,
        operation_id: request.operation_id.clone(),
        input_fingerprint: request.input_fingerprint.clone(),
    };
    let archive_dir = provider.archive_dir(archive_id)?;
    if archive_dir.exists() {
        return validate_existing_archive(&archive_dir, &manifest);
    }
    let parent = archive_dir
        .parent()
        .ok_or_else(|| RuntimeError::StoreCorrupt {
            operation: "archive Worker Session",
            path: archive_dir.clone(),
            message: "archive target has no parent".to_string(),
        })?;
    fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo {
        operation: "archive Worker Session",
        path: parent.to_path_buf(),
        source,
    })?;
    let staging = temporary_path(&archive_dir);
    let result = (|| {
        fs::create_dir(&staging).map_err(|source| RuntimeError::StoreIo {
            operation: "archive Worker Session",
            path: staging.clone(),
            source,
        })?;
        let target_session = staging.join("session");
        fs::create_dir(&target_session).map_err(|source| RuntimeError::StoreIo {
            operation: "archive Worker Session",
            path: target_session.clone(),
            source,
        })?;
        copy_files(&source_files, &target_session, "archive Worker Session")?;
        atomic_write_json(
            &staging.join("manifest.json"),
            &manifest,
            "archive Worker Session",
        )?;
        sync_tree(&staging, "archive Worker Session")?;
        fs::rename(&staging, &archive_dir).map_err(|source| RuntimeError::StoreIo {
            operation: "archive Worker Session",
            path: archive_dir.clone(),
            source,
        })?;
        sync_directory(parent, "archive Worker Session")?;
        validate_existing_archive(&archive_dir, &manifest)
    })();
    if result.is_err() {
        let _ = fs::remove_dir_all(&staging);
    }
    result
}

fn validate_existing_archive(
    archive_dir: &Path,
    expected: &WorkerSessionArchiveManifest,
) -> Result<WorkerSessionArchiveManifest, RuntimeError> {
    let existing: WorkerSessionArchiveManifest = read_json(
        &archive_dir.join("manifest.json"),
        "verify Worker Session archive",
    )?;
    let mut comparable_expected = expected.clone();
    comparable_expected.archived_at_unix_seconds = existing.archived_at_unix_seconds;
    if existing != comparable_expected {
        return Err(RuntimeError::StoreCorrupt {
            operation: "verify Worker Session archive",
            path: archive_dir.join("manifest.json"),
            message: "archive id collision or manifest mismatch".to_string(),
        });
    }
    let files = collect_files(
        &archive_dir.join("session"),
        "verify Worker Session archive",
    )?;
    let (checksum, bytes, count) = checksum_files(&files, "verify Worker Session archive")?;
    if checksum != existing.content_checksum_sha256
        || bytes != existing.content_bytes
        || count != existing.content_file_count
    {
        return Err(RuntimeError::StoreCorrupt {
            operation: "verify Worker Session archive",
            path: archive_dir.to_path_buf(),
            message: "archive checksum or content summary mismatch".to_string(),
        });
    }
    Ok(existing)
}

fn commit_diagnostics_archive(
    provider: &FsWorkerRetentionProvider,
    request: &WorkerRetentionExecutionRequest,
    worker_dir: &Path,
) -> Result<(), RuntimeError> {
    let files = diagnostics_files(worker_dir, "archive Worker diagnostics")?;
    let target = provider.diagnostics_dir(&request.operation_id)?;
    if target.exists() {
        validate_existing_diagnostics_archive(&target, request)?;
        return Ok(());
    }
    let parent = target.parent().ok_or_else(|| RuntimeError::StoreCorrupt {
        operation: "archive Worker diagnostics",
        path: target.clone(),
        message: "diagnostics archive target has no parent".to_string(),
    })?;
    fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo {
        operation: "archive Worker diagnostics",
        path: parent.to_path_buf(),
        source,
    })?;
    let staging = temporary_path(&target);
    let result = (|| {
        fs::create_dir(&staging).map_err(|source| RuntimeError::StoreIo {
            operation: "archive Worker diagnostics",
            path: staging.clone(),
            source,
        })?;
        let source_files = files
            .iter()
            .map(|path| {
                let relative =
                    path.strip_prefix(worker_dir)
                        .map_err(|_| RuntimeError::StoreCorrupt {
                            operation: "archive Worker diagnostics",
                            path: path.clone(),
                            message: "diagnostics path escaped Worker aggregate".to_string(),
                        })?;
                Ok((relative.to_path_buf(), path.clone()))
            })
            .collect::<Result<Vec<_>, RuntimeError>>()?;
        copy_files(&source_files, &staging, "archive Worker diagnostics")?;
        let (checksum, bytes, count) = checksum_files(&source_files, "archive Worker diagnostics")?;
        let manifest = DiagnosticsArchiveManifest {
            schema_version: ARCHIVE_SCHEMA_VERSION,
            operation_id: request.operation_id.clone(),
            workspace_id: request.workspace_id.clone(),
            source_runtime_id: request.source_runtime_id.clone(),
            source_worker_id: request.worker_id,
            input_fingerprint: request.input_fingerprint.clone(),
            content_checksum_sha256: checksum,
            content_bytes: bytes,
            content_file_count: count,
        };
        atomic_write_json(
            &staging.join("manifest.json"),
            &manifest,
            "archive Worker diagnostics",
        )?;
        sync_tree(&staging, "archive Worker diagnostics")?;
        fs::rename(&staging, &target).map_err(|source| RuntimeError::StoreIo {
            operation: "archive Worker diagnostics",
            path: target.clone(),
            source,
        })?;
        sync_directory(parent, "archive Worker diagnostics")?;
        validate_existing_diagnostics_archive(&target, request)
    })();
    if result.is_err() {
        let _ = fs::remove_dir_all(&staging);
    }
    result
}

fn validate_existing_diagnostics_archive(
    target: &Path,
    request: &WorkerRetentionExecutionRequest,
) -> Result<(), RuntimeError> {
    let manifest: DiagnosticsArchiveManifest = read_json(
        &target.join("manifest.json"),
        "verify Worker diagnostics archive",
    )?;
    if manifest.schema_version != ARCHIVE_SCHEMA_VERSION
        || manifest.operation_id != request.operation_id
        || manifest.workspace_id != request.workspace_id
        || manifest.source_runtime_id != request.source_runtime_id
        || manifest.source_worker_id != request.worker_id
        || manifest.input_fingerprint != request.input_fingerprint
    {
        return Err(RuntimeError::InvalidRequest(format!(
            "diagnostics archive operation {} does not match the retention request",
            request.operation_id
        )));
    }
    let mut files = collect_files(target, "verify Worker diagnostics archive")?;
    files.retain(|(relative, _)| relative != Path::new("manifest.json"));
    let (checksum, bytes, count) = checksum_files(&files, "verify Worker diagnostics archive")?;
    if checksum != manifest.content_checksum_sha256
        || bytes != manifest.content_bytes
        || count != manifest.content_file_count
    {
        return Err(RuntimeError::StoreCorrupt {
            operation: "verify Worker diagnostics archive",
            path: target.to_path_buf(),
            message: "diagnostics archive checksum or content summary mismatch".to_string(),
        });
    }
    Ok(())
}

fn diagnostics_files(
    worker_dir: &Path,
    operation: &'static str,
) -> Result<Vec<PathBuf>, RuntimeError> {
    let runs = worker_dir.join("runs");
    if !runs.is_dir() {
        return Ok(Vec::new());
    }
    let mut files = Vec::new();
    for (_, path) in collect_files(&runs, operation)? {
        let name = path.file_name().and_then(|name| name.to_str());
        if matches!(name, Some("worker.out.log" | "worker.err.log")) {
            files.push(path);
        }
    }
    files.sort();
    Ok(files)
}

fn collect_files(
    root: &Path,
    operation: &'static str,
) -> Result<Vec<(PathBuf, PathBuf)>, RuntimeError> {
    fn visit(
        root: &Path,
        current: &Path,
        operation: &'static str,
        files: &mut Vec<(PathBuf, PathBuf)>,
    ) -> Result<(), RuntimeError> {
        let mut entries = fs::read_dir(current)
            .map_err(|source| RuntimeError::StoreIo {
                operation,
                path: current.to_path_buf(),
                source,
            })?
            .collect::<Result<Vec<_>, _>>()
            .map_err(|source| RuntimeError::StoreIo {
                operation,
                path: current.to_path_buf(),
                source,
            })?;
        entries.sort_by_key(|entry| entry.file_name());
        for entry in entries {
            let path = entry.path();
            let file_type = entry.file_type().map_err(|source| RuntimeError::StoreIo {
                operation,
                path: path.clone(),
                source,
            })?;
            if file_type.is_symlink() {
                return Err(RuntimeError::StoreCorrupt {
                    operation,
                    path,
                    message: "symlinks are not allowed in retained Worker evidence".to_string(),
                });
            }
            if file_type.is_dir() {
                visit(root, &path, operation, files)?;
            } else if file_type.is_file() {
                let relative = path
                    .strip_prefix(root)
                    .map_err(|_| RuntimeError::StoreCorrupt {
                        operation,
                        path: path.clone(),
                        message: "retention source escaped its aggregate root".to_string(),
                    })?
                    .to_path_buf();
                files.push((relative, path));
            } else {
                return Err(RuntimeError::StoreCorrupt {
                    operation,
                    path,
                    message: "unsupported retained Worker evidence entry".to_string(),
                });
            }
        }
        Ok(())
    }
    if !root.is_dir() {
        return Ok(Vec::new());
    }
    let mut files = Vec::new();
    visit(root, root, operation, &mut files)?;
    files.sort_by(|left, right| left.0.cmp(&right.0));
    Ok(files)
}

fn checksum_files(
    files: &[(PathBuf, PathBuf)],
    operation: &'static str,
) -> Result<(String, u64, u64), RuntimeError> {
    let mut hasher = Sha256::new();
    let mut total = 0_u64;
    for (relative, path) in files {
        let relative = relative.to_string_lossy();
        hasher.update((relative.len() as u64).to_be_bytes());
        hasher.update(relative.as_bytes());
        let mut file = File::open(path).map_err(|source| RuntimeError::StoreIo {
            operation,
            path: path.clone(),
            source,
        })?;
        let length = file_len(path, operation)?;
        hasher.update(length.to_be_bytes());
        let mut buffer = [0_u8; 8192];
        loop {
            let read = file
                .read(&mut buffer)
                .map_err(|source| RuntimeError::StoreIo {
                    operation,
                    path: path.clone(),
                    source,
                })?;
            if read == 0 {
                break;
            }
            hasher.update(&buffer[..read]);
        }
        total = total.saturating_add(length);
    }
    let digest = hasher.finalize();
    let checksum = digest
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<String>();
    Ok((checksum, total, files.len() as u64))
}

fn copy_files(
    files: &[(PathBuf, PathBuf)],
    target_root: &Path,
    operation: &'static str,
) -> Result<(), RuntimeError> {
    for (relative, source_path) in files {
        let target = target_root.join(relative);
        let parent = target.parent().ok_or_else(|| RuntimeError::StoreCorrupt {
            operation,
            path: target.clone(),
            message: "retention copy target has no parent".to_string(),
        })?;
        fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo {
            operation,
            path: parent.to_path_buf(),
            source,
        })?;
        let bytes = fs::read(source_path).map_err(|source| RuntimeError::StoreIo {
            operation,
            path: source_path.clone(),
            source,
        })?;
        let mut file = OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&target)
            .map_err(|source| RuntimeError::StoreIo {
                operation,
                path: target.clone(),
                source,
            })?;
        file.write_all(&bytes)
            .and_then(|_| file.sync_all())
            .map_err(|source| RuntimeError::StoreIo {
                operation,
                path: target,
                source,
            })?;
    }
    Ok(())
}

fn atomic_write_json<T: Serialize>(
    path: &Path,
    value: &T,
    operation: &'static str,
) -> Result<(), RuntimeError> {
    let parent = path.parent().ok_or_else(|| RuntimeError::StoreCorrupt {
        operation,
        path: path.to_path_buf(),
        message: "retention record has no parent".to_string(),
    })?;
    fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo {
        operation,
        path: parent.to_path_buf(),
        source,
    })?;
    let temporary = temporary_path(path);
    let result = (|| {
        let mut file = OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&temporary)
            .map_err(|source| RuntimeError::StoreIo {
                operation,
                path: temporary.clone(),
                source,
            })?;
        serde_json::to_writer_pretty(&mut file, value).map_err(|source| {
            RuntimeError::StoreCorrupt {
                operation,
                path: temporary.clone(),
                message: source.to_string(),
            }
        })?;
        file.write_all(b"\n")
            .and_then(|_| file.sync_all())
            .map_err(|source| RuntimeError::StoreIo {
                operation,
                path: temporary.clone(),
                source,
            })?;
        drop(file);
        fs::rename(&temporary, path).map_err(|source| RuntimeError::StoreIo {
            operation,
            path: path.to_path_buf(),
            source,
        })?;
        sync_directory(parent, operation)
    })();
    if result.is_err() {
        let _ = fs::remove_file(temporary);
    }
    result
}

fn read_json<T: for<'de> Deserialize<'de>>(
    path: &Path,
    operation: &'static str,
) -> Result<T, RuntimeError> {
    let file = File::open(path).map_err(|source| match source.kind() {
        std::io::ErrorKind::NotFound => RuntimeError::StoreMissing {
            operation,
            path: path.to_path_buf(),
        },
        _ => RuntimeError::StoreIo {
            operation,
            path: path.to_path_buf(),
            source,
        },
    })?;
    serde_json::from_reader(file).map_err(|source| RuntimeError::StoreCorrupt {
        operation,
        path: path.to_path_buf(),
        message: source.to_string(),
    })
}

fn sync_tree(path: &Path, operation: &'static str) -> Result<(), RuntimeError> {
    let mut directories = vec![path.to_path_buf()];
    let mut index = 0;
    while index < directories.len() {
        let current = directories[index].clone();
        index += 1;
        for entry in fs::read_dir(&current).map_err(|source| RuntimeError::StoreIo {
            operation,
            path: current.clone(),
            source,
        })? {
            let entry = entry.map_err(|source| RuntimeError::StoreIo {
                operation,
                path: current.clone(),
                source,
            })?;
            if entry.path().is_dir() {
                directories.push(entry.path());
            }
        }
    }
    for directory in directories.into_iter().rev() {
        sync_directory(&directory, operation)?;
    }
    Ok(())
}

fn sync_directory(path: &Path, operation: &'static str) -> Result<(), RuntimeError> {
    File::open(path)
        .and_then(|file| file.sync_all())
        .map_err(|source| RuntimeError::StoreIo {
            operation,
            path: path.to_path_buf(),
            source,
        })
}

fn file_len(path: &Path, operation: &'static str) -> Result<u64, RuntimeError> {
    fs::metadata(path)
        .map(|metadata| metadata.len())
        .map_err(|source| RuntimeError::StoreIo {
            operation,
            path: path.to_path_buf(),
            source,
        })
}

fn temporary_path(path: &Path) -> PathBuf {
    let sequence = NEXT_TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
    let name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("retention");
    path.with_file_name(format!(".{name}.tmp-{}-{sequence}", std::process::id()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::{Arc, Barrier};

    fn write_json(path: &Path, value: &impl Serialize) {
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(path, serde_json::to_vec_pretty(value).unwrap()).unwrap();
    }

    fn source(root: &Path, worker_id: WorkerId, generation: u64) {
        let worker = root.join("workers").join(worker_id.to_string());
        write_json(
            &worker.join("worker.json"),
            &serde_json::json!({"workspace_id": "workspace-a", "run_generation": generation}),
        );
        write_json(
            &worker.join("session/session.json"),
            &serde_json::json!({"schema_version": 1, "session_id": "session-a"}),
        );
        fs::create_dir_all(worker.join("session/segments")).unwrap();
        fs::write(worker.join("session/segments/segment-a.jsonl"), b"one\n").unwrap();
        fs::create_dir_all(worker.join(format!("runs/{generation}"))).unwrap();
        fs::write(
            worker.join(format!("runs/{generation}/worker.out.log")),
            b"diagnostic\n",
        )
        .unwrap();
        fs::write(
            worker.join(format!("runs/{generation}/worker.sock")),
            b"not retained",
        )
        .unwrap();
    }

    fn request(
        worker_id: WorkerId,
        generation: u64,
        disposition: SessionDisposition,
    ) -> WorkerRetentionExecutionRequest {
        WorkerRetentionExecutionRequest {
            operation_id: "operation-a".to_string(),
            input_fingerprint: "fingerprint-a".to_string(),
            expected_worker_revision: "revision-a".to_string(),
            archive_id: (disposition == SessionDisposition::Archive)
                .then(|| "archive-a".to_string()),
            workspace_id: "workspace-a".to_string(),
            source_runtime_id: "runtime-a".to_string(),
            worker_id,
            expected_run_generation: generation,
            source_created_at: "2026-01-01T00:00:00Z".to_string(),
            removed_at: "2026-01-02T00:00:00Z".to_string(),
            effective_profile: Some("builtin:coder".to_string()),
            retention_class: None,
            policy_id: "policy-a".to_string(),
            policy_revision: 3,
            session_disposition: disposition,
            diagnostics_disposition: DiagnosticsDisposition::Purge,
        }
    }

    #[test]
    fn archive_is_verified_before_source_removal_and_retry_converges() {
        let temp = tempfile::tempdir().unwrap();
        let worker_id = WorkerId::new(7);
        source(temp.path(), worker_id, 4);
        let provider = FsWorkerRetentionProvider::new(temp.path());
        let request = request(worker_id, 4, SessionDisposition::Archive);

        let first = provider.execute(&request).unwrap();
        assert!(first.source_removed);
        let archive = first.archive.as_ref().unwrap();
        assert_eq!(archive.source_session_id, "session-a");
        assert_eq!(archive.segment_ids, vec!["segment-a"]);
        assert!(!temp.path().join("workers/7").exists());
        assert!(
            temp.path()
                .join("archives/workers/archive-a/session/segments/segment-a.jsonl")
                .is_file()
        );
        assert!(!temp.path().join("archives/workers/archive-a/runs").exists());

        let retry = provider.execute(&request).unwrap();
        assert_eq!(retry, first);
        assert_eq!(
            fs::read_dir(temp.path().join("archives/workers"))
                .unwrap()
                .count(),
            1
        );
    }

    #[test]
    fn archive_failure_keeps_live_source_for_retry() {
        let temp = tempfile::tempdir().unwrap();
        let worker_id = WorkerId::new(8);
        source(temp.path(), worker_id, 2);
        let collision = temp.path().join("archives/workers/archive-a");
        fs::create_dir_all(&collision).unwrap();
        fs::write(collision.join("manifest.json"), b"not-json").unwrap();
        let provider = FsWorkerRetentionProvider::new(temp.path());

        assert!(
            provider
                .execute(&request(worker_id, 2, SessionDisposition::Archive))
                .is_err()
        );
        assert!(temp.path().join("workers/8/session").is_dir());
        assert!(
            !temp
                .path()
                .join("retention/operations/operation-a.json")
                .exists()
        );
    }

    #[test]
    fn target_inventory_and_execute_reject_cross_workspace_aggregate() {
        let temp = tempfile::tempdir().unwrap();
        let worker_id = WorkerId::new(16);
        source(temp.path(), worker_id, 3);
        let provider = FsWorkerRetentionProvider::new(temp.path());
        assert!(matches!(
            provider.inventory("other-workspace", "runtime-a", worker_id, 3),
            Err(RuntimeError::WorkerNotFound { .. })
        ));
        let mut request = request(worker_id, 3, SessionDisposition::Purge);
        request.workspace_id = "other-workspace".to_string();
        assert!(matches!(
            provider.execute(&request),
            Err(RuntimeError::WorkerNotFound { .. })
        ));
        assert!(temp.path().join("workers/16/session").is_dir());
        assert!(
            !temp
                .path()
                .join("retention/operations/operation-a.json")
                .exists()
        );

        request.workspace_id = "workspace-a".to_string();
        provider.execute(&request).unwrap();
        request.workspace_id = "other-workspace".to_string();
        assert!(provider.execute(&request).is_err());
    }

    #[test]
    fn purge_removes_aggregate_and_rejects_stale_generation() {
        let temp = tempfile::tempdir().unwrap();
        let provider = FsWorkerRetentionProvider::new(temp.path());
        let worker_id = WorkerId::new(9);
        source(temp.path(), worker_id, 5);
        let stale = request(worker_id, 4, SessionDisposition::Purge);
        assert!(provider.execute(&stale).is_err());
        assert!(temp.path().join("workers/9/session").is_dir());

        let mut current = request(worker_id, 5, SessionDisposition::Purge);
        current.operation_id = "operation-current".to_string();
        current.input_fingerprint = "fingerprint-current".to_string();
        let result = provider.execute(&current).unwrap();
        assert!(result.archive.is_none());
        assert!(!temp.path().join("workers/9").exists());
        assert!(
            temp.path()
                .join("retention/operations/operation-current.json")
                .is_file()
        );
    }

    #[test]
    fn pending_receipt_recovers_delete_to_receipt_crash_window() {
        let temp = tempfile::tempdir().unwrap();
        let worker_id = WorkerId::new(11);
        source(temp.path(), worker_id, 1);
        let provider = FsWorkerRetentionProvider::new(temp.path());
        let request = request(worker_id, 1, SessionDisposition::Archive);
        let completed = provider.execute(&request).unwrap();
        let receipt_path = temp.path().join("retention/operations/operation-a.json");
        let mut receipt: RetentionOperationReceipt =
            serde_json::from_slice(&fs::read(&receipt_path).unwrap()).unwrap();
        receipt.result.source_removed = false;
        fs::write(&receipt_path, serde_json::to_vec_pretty(&receipt).unwrap()).unwrap();

        let recovered = provider.execute(&request).unwrap();
        assert_eq!(recovered, completed);
        assert!(provider.completed_for(&request).unwrap().is_some());
    }

    #[test]
    fn provider_snapshot_scans_aggregate_storage_independent_of_runtime_catalog() {
        let temp = tempfile::tempdir().unwrap();
        source(temp.path(), WorkerId::new(13), 2);
        source(temp.path(), WorkerId::new(14), 1);
        write_json(
            &temp.path().join("workers/14/worker.json"),
            &serde_json::json!({"workspace_id": "other-workspace", "run_generation": 1}),
        );
        fs::create_dir_all(temp.path().join("workers/not-a-worker")).unwrap();
        fs::write(
            temp.path().join("workers/not-a-worker/worker.json"),
            b"not-json",
        )
        .unwrap();
        fs::create_dir_all(temp.path().join("workers/15")).unwrap();
        fs::write(temp.path().join("workers/15/worker.json"), b"not-json").unwrap();
        let provider = FsWorkerRetentionProvider::new(temp.path());

        let snapshot = provider.snapshot("workspace-a", "runtime-a").unwrap();
        assert_eq!(snapshot.workers().len(), 1);
        assert_eq!(snapshot.workers()[0].worker_id, WorkerId::new(13));
        assert!(snapshot.diagnostics().iter().any(|diagnostic| {
            diagnostic.worker_id() == "14"
                && diagnostic.category() == "aggregate_workspace_mismatch"
        }));
        assert!(snapshot.diagnostics().iter().any(|diagnostic| {
            diagnostic.worker_id() == "not-a-worker"
                && diagnostic.category() == "aggregate_worker_id_invalid"
        }));
        assert!(snapshot.diagnostics().iter().any(|diagnostic| {
            diagnostic.worker_id() == "15"
                && diagnostic.category() == "aggregate_worker_record_corrupt"
        }));
    }

    #[test]
    fn diagnostics_retry_rejects_corrupt_existing_archive_before_source_delete() {
        let temp = tempfile::tempdir().unwrap();
        let worker_id = WorkerId::new(12);
        source(temp.path(), worker_id, 1);
        let provider = FsWorkerRetentionProvider::new(temp.path());
        let mut request = request(worker_id, 1, SessionDisposition::Archive);
        request.diagnostics_disposition = DiagnosticsDisposition::Retain;
        provider.execute(&request).unwrap();

        let receipt_path = temp.path().join("retention/operations/operation-a.json");
        let mut receipt: RetentionOperationReceipt =
            serde_json::from_slice(&fs::read(&receipt_path).unwrap()).unwrap();
        receipt.result.source_removed = false;
        fs::write(&receipt_path, serde_json::to_vec_pretty(&receipt).unwrap()).unwrap();
        source(temp.path(), worker_id, 1);
        fs::write(
            temp.path()
                .join("archives/diagnostics/operation-a/runs/1/worker.out.log"),
            b"corrupt\n",
        )
        .unwrap();

        assert!(provider.execute(&request).is_err());
        assert!(temp.path().join("workers/12/session").is_dir());
        assert!(provider.completed_for(&request).unwrap().is_none());
    }

    #[test]
    fn concurrent_retry_produces_one_archive() {
        let temp = tempfile::tempdir().unwrap();
        let worker_id = WorkerId::new(10);
        source(temp.path(), worker_id, 1);
        let provider = Arc::new(FsWorkerRetentionProvider::new(temp.path()));
        let request = Arc::new(request(worker_id, 1, SessionDisposition::Archive));
        let barrier = Arc::new(Barrier::new(3));
        let handles = (0..2)
            .map(|_| {
                let provider = provider.clone();
                let request = request.clone();
                let barrier = barrier.clone();
                std::thread::spawn(move || {
                    barrier.wait();
                    provider.execute(&request)
                })
            })
            .collect::<Vec<_>>();
        barrier.wait();
        let results = handles
            .into_iter()
            .map(|handle| handle.join().unwrap().unwrap())
            .collect::<Vec<_>>();
        assert_eq!(results[0], results[1]);
        assert_eq!(
            fs::read_dir(temp.path().join("archives/workers"))
                .unwrap()
                .count(),
            1
        );
    }
}