aidens_receipts/

lib.rs

//! Thin persistence wiring for canonical receipt crates.
//!
//! AiDENs does not define receipt semantics here. This crate only gives the
//! product layer a small JSONL sink for receipts whose schemas are owned by
//! canonical libraries, plus orchestration reports that are explicitly labeled
//! as AiDENs-owned reports rather than stack receipts.

use async_trait::async_trait;
use chrono::Utc;
pub use llm_tool_runtime::{
    ToolError, ToolErrorClass, ToolReceipt as CanonicalRuntimeToolReceipt, ToolReceiptSink,
};
pub use semantic_memory_forge::ForgeToolReceiptV2 as CanonicalForgeToolReceiptV2;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use stack_ids::ContentDigest;
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::thread;
use std::time::{Duration, Instant};
use thiserror::Error;
pub use verification_control::ControlReceipt as CanonicalControlReceipt;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunBundleStoreConfig {
    pub root_path: PathBuf,
    pub bundles_path: PathBuf,
    pub index_path: PathBuf,
}

impl RunBundleStoreConfig {
    pub fn for_receipt_root(root_path: impl Into<PathBuf>) -> Self {
        let root_path = root_path.into();
        let bundles_path = root_path.join("run-bundles");
        Self {
            index_path: bundles_path.join("index.ndjson"),
            bundles_path,
            root_path,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunBundleStoreRecord {
    pub artifact_kind: String,
    pub ownership: String,
    pub support_tier: String,
    pub semantic_status: String,
    pub run_id: String,
    pub bundle_schema: String,
    pub recorded_at: String,
    pub bundle_path: PathBuf,
    pub content_digest: ContentDigest,
    pub canonical_event_log_path: PathBuf,
    pub known_limits: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunBundleStoreInspection {
    pub record: RunBundleStoreRecord,
    pub bundle: Value,
    pub digest_verified: bool,
}

#[derive(Debug, Error)]
pub enum RunBundleStoreError {
    #[error("run bundle store io error at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    #[error("run bundle store json error at {path}: {source}")]
    Json {
        path: PathBuf,
        #[source]
        source: serde_json::Error,
    },
    #[error("run bundle digest error: {source}")]
    Digest {
        #[source]
        source: stack_ids::DigestError,
    },
    #[error("run bundle store requires AiDENsRunBundleV3, got {0}")]
    UnsupportedSchema(String),
    #[error("run bundle missing string run_id")]
    MissingRunId,
    #[error("run bundle not found: {0}")]
    NotFound(String),
    #[error("run bundle store has multiple bundles; pass a specific run output directory or bundle path")]
    AmbiguousBundle,
    #[error("run bundle already exists at {0}")]
    AlreadyExists(String),
    #[error("run bundle index chain verification failed at sequence {0}")]
    ChainVerificationFailed(u64),
}

#[derive(Debug, Clone)]
pub struct RunBundleStore {
    config: RunBundleStoreConfig,
}

impl RunBundleStore {
    pub fn open(config: RunBundleStoreConfig) -> Result<Self, RunBundleStoreError> {
        std::fs::create_dir_all(&config.bundles_path).map_err(|source| {
            RunBundleStoreError::Io {
                path: config.bundles_path.clone(),
                source,
            }
        })?;
        ensure_run_bundle_file(&config.index_path)?;
        Ok(Self { config })
    }

    pub fn config(&self) -> &RunBundleStoreConfig {
        &self.config
    }

    pub fn bundle_path_for_run_id(&self, run_id: &str) -> PathBuf {
        self.config
            .bundles_path
            .join(receipt_store_segment(run_id))
            .join("run-bundle.json")
    }

    pub fn bundle_path_for_run_id_and_digest(
        &self,
        run_id: &str,
        content_digest: &ContentDigest,
    ) -> PathBuf {
        self.config
            .bundles_path
            .join(receipt_store_segment(run_id))
            .join(content_digest.hex())
            .join("run-bundle.json")
    }

    pub fn write_bundle_value(
        &self,
        bundle: &Value,
    ) -> Result<RunBundleStoreRecord, RunBundleStoreError> {
        let schema = bundle
            .get("schema")
            .and_then(Value::as_str)
            .unwrap_or_default();
        if schema != "AiDENsRunBundleV3" {
            return Err(RunBundleStoreError::UnsupportedSchema(schema.into()));
        }
        let run_id = bundle
            .get("run_id")
            .and_then(Value::as_str)
            .ok_or(RunBundleStoreError::MissingRunId)?;
        let content_digest = ContentDigest::compute_json(bundle)
            .map_err(|source| RunBundleStoreError::Digest { source })?;
        let bundle_path = self.bundle_path_for_run_id_and_digest(run_id, &content_digest);
        if bundle_path.exists() {
            return Err(RunBundleStoreError::AlreadyExists(
                bundle_path.display().to_string(),
            ));
        }
        if let Some(parent) = bundle_path.parent() {
            std::fs::create_dir_all(parent).map_err(|source| RunBundleStoreError::Io {
                path: parent.to_path_buf(),
                source,
            })?;
        }
        let body =
            serde_json::to_string_pretty(bundle).map_err(|source| RunBundleStoreError::Json {
                path: bundle_path.clone(),
                source,
            })?;
        write_atomic_file(&bundle_path, &(body + "\n")).map_err(|source| {
            RunBundleStoreError::Io {
                path: bundle_path.clone(),
                source,
            }
        })?;
        let record = RunBundleStoreRecord {
            artifact_kind: "local_operator_run_bundle_store_record".into(),
            ownership:
                "AiDENs-local operator evidence; canonical receipt and trace semantics remain in owner crates.".into(),
            support_tier: bundle
                .pointer("/support/support_tier")
                .and_then(Value::as_str)
                .unwrap_or("partial")
                .into(),
            semantic_status: if bundle
                .pointer("/failure/degraded")
                .and_then(Value::as_bool)
                .unwrap_or(false)
            {
                "degraded_exact_check".into()
            } else {
                "exact_check".into()
            },
            run_id: run_id.into(),
            bundle_schema: schema.into(),
            recorded_at: Utc::now().to_rfc3339(),
            bundle_path: bundle_path.clone(),
            content_digest,
            canonical_event_log_path: self.config.root_path.join("canonical-receipts.ndjson"),
            known_limits: vec![
                "Stores AiDENsRunBundleV3 operator evidence only; it is not a canonical memory or verification truth store.".into(),
            ],
        };
        if let Err(error) = self.append_record(&record) {
            let _ = std::fs::remove_file(&bundle_path);
            return Err(error);
        }
        Ok(record)
    }

    pub fn write_bundle<T: Serialize>(
        &self,
        bundle: &T,
    ) -> Result<RunBundleStoreRecord, RunBundleStoreError> {
        let value = serde_json::to_value(bundle).map_err(|source| RunBundleStoreError::Json {
            path: self.config.index_path.clone(),
            source,
        })?;
        self.write_bundle_value(&value)
    }

    pub fn list_records(&self) -> Result<Vec<RunBundleStoreRecord>, RunBundleStoreError> {
        read_run_bundle_records(&self.config.index_path)
    }

    pub fn inspect(&self, run_id: &str) -> Result<RunBundleStoreInspection, RunBundleStoreError> {
        let record = self
            .list_records()?
            .into_iter()
            .rev()
            .find(|record| record.run_id == run_id)
            .ok_or_else(|| RunBundleStoreError::NotFound(run_id.into()))?;
        let bundle_text = std::fs::read_to_string(&record.bundle_path).map_err(|source| {
            RunBundleStoreError::Io {
                path: record.bundle_path.clone(),
                source,
            }
        })?;
        let bundle: Value =
            serde_json::from_str(&bundle_text).map_err(|source| RunBundleStoreError::Json {
                path: record.bundle_path.clone(),
                source,
            })?;
        let digest_verified = ContentDigest::compute_json(&bundle)
            .map(|digest| digest == record.content_digest)
            .unwrap_or(false);
        Ok(RunBundleStoreInspection {
            record,
            bundle,
            digest_verified,
        })
    }

    pub fn single_bundle_path(&self) -> Result<PathBuf, RunBundleStoreError> {
        let records = self.list_records()?;
        let mut paths = records
            .into_iter()
            .map(|record| record.bundle_path)
            .collect::<Vec<_>>();
        paths.sort();
        paths.dedup();
        match paths.len() {
            0 => Err(RunBundleStoreError::NotFound(
                "run-bundles/index.ndjson".into(),
            )),
            1 => Ok(paths.remove(0)),
            _ => Err(RunBundleStoreError::AmbiguousBundle),
        }
    }

    fn append_record(&self, record: &RunBundleStoreRecord) -> Result<(), RunBundleStoreError> {
        let _lock = acquire_exclusive_lock(&self.config.index_path).map_err(|source| {
            RunBundleStoreError::Io {
                path: lock_path_for(&self.config.index_path),
                source,
            }
        })?;
        let mut file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.config.index_path)
            .map_err(|source| RunBundleStoreError::Io {
                path: self.config.index_path.clone(),
                source,
            })?;
        serde_json::to_writer(&mut file, record).map_err(|source| RunBundleStoreError::Json {
            path: self.config.index_path.clone(),
            source,
        })?;
        file.write_all(b"\n")
            .map_err(|source| RunBundleStoreError::Io {
                path: self.config.index_path.clone(),
                source,
            })?;
        file.flush().map_err(|source| RunBundleStoreError::Io {
            path: self.config.index_path.clone(),
            source,
        })?;
        file.sync_all().map_err(|source| RunBundleStoreError::Io {
            path: self.config.index_path.clone(),
            source,
        })?;
        Ok(())
    }
}

pub fn forge_tool_receipt_from_runtime(
    receipt: &CanonicalRuntimeToolReceipt,
    raw_payload: Value,
) -> CanonicalForgeToolReceiptV2 {
    receipt.to_forge_tool_receipt_v2(raw_payload)
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CanonicalEventLogConfig {
    pub root_path: PathBuf,
    pub records_path: PathBuf,
    #[serde(default)]
    pub append_only: bool,
}

impl CanonicalEventLogConfig {
    pub fn for_root(root_path: impl Into<PathBuf>) -> Self {
        let root_path = root_path.into();
        Self {
            records_path: root_path.join("canonical-receipts.ndjson"),
            root_path,
            append_only: true,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CanonicalEventLogEntry {
    #[serde(default)]
    pub sequence_number: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub previous_record_digest: Option<ContentDigest>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub record_digest: Option<ContentDigest>,
    pub owner_crate: String,
    pub schema_name: String,
    pub receipt_id: String,
    pub recorded_at: String,
    pub content_digest: ContentDigest,
    pub body: Value,
}

impl CanonicalEventLogEntry {
    pub fn new(
        owner_crate: impl Into<String>,
        schema_name: impl Into<String>,
        receipt_id: impl Into<String>,
        body: Value,
    ) -> Result<Self, CanonicalEventLogError> {
        Self::new_with_chain(owner_crate, schema_name, receipt_id, body, 0, None)
    }

    pub fn new_with_chain(
        owner_crate: impl Into<String>,
        schema_name: impl Into<String>,
        receipt_id: impl Into<String>,
        body: Value,
        sequence_number: u64,
        previous_record_digest: Option<ContentDigest>,
    ) -> Result<Self, CanonicalEventLogError> {
        let content_digest = ContentDigest::compute_json(&body)
            .map_err(|source| CanonicalEventLogError::Digest { source })?;
        let mut entry = Self {
            sequence_number,
            previous_record_digest,
            record_digest: None,
            owner_crate: owner_crate.into(),
            schema_name: schema_name.into(),
            receipt_id: receipt_id.into(),
            recorded_at: Utc::now().to_rfc3339(),
            content_digest,
            body,
        };
        entry.record_digest = Some(entry.compute_record_digest()?);
        Ok(entry)
    }

    pub fn verify_digest(&self) -> bool {
        ContentDigest::compute_json(&self.body)
            .map(|digest| digest == self.content_digest)
            .unwrap_or(false)
    }

    pub fn compute_record_digest(&self) -> Result<ContentDigest, CanonicalEventLogError> {
        let payload = serde_json::json!({
            "sequence_number": self.sequence_number,
            "previous_record_digest": self.previous_record_digest,
            "owner_crate": self.owner_crate,
            "schema_name": self.schema_name,
            "receipt_id": self.receipt_id,
            "recorded_at": self.recorded_at,
            "content_digest": self.content_digest,
            "body": self.body,
        });
        ContentDigest::compute_json(&payload)
            .map_err(|source| CanonicalEventLogError::Digest { source })
    }

    pub fn verify_record_digest(&self) -> bool {
        self.record_digest.as_ref().is_some_and(|expected| {
            self.compute_record_digest()
                .is_ok_and(|actual| actual == *expected)
        })
    }
}

#[derive(Debug, Error)]
pub enum CanonicalEventLogError {
    #[error("canonical receipt log io error at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    #[error("canonical receipt log json error at {path}: {source}")]
    Json {
        path: PathBuf,
        #[source]
        source: serde_json::Error,
    },
    #[error("canonical receipt digest error: {source}")]
    Digest {
        #[source]
        source: stack_ids::DigestError,
    },
    #[error("canonical receipt record not found: {0}")]
    NotFound(String),
    #[error("canonical receipt log chain verification failed at sequence {0}")]
    ChainVerificationFailed(u64),
    #[error("canonical receipt log duplicate receipt id: {0}")]
    DuplicateReceiptId(String),
}

#[derive(Debug, Clone)]
pub struct CanonicalEventLog {
    config: CanonicalEventLogConfig,
}

impl CanonicalEventLog {
    pub fn open(config: CanonicalEventLogConfig) -> Result<Self, CanonicalEventLogError> {
        std::fs::create_dir_all(&config.root_path).map_err(|source| {
            CanonicalEventLogError::Io {
                path: config.root_path.clone(),
                source,
            }
        })?;
        ensure_file(&config.records_path)?;
        Ok(Self { config })
    }

    pub fn config(&self) -> &CanonicalEventLogConfig {
        &self.config
    }

    pub fn append_runtime_tool_receipt(
        &self,
        receipt: &CanonicalRuntimeToolReceipt,
    ) -> Result<CanonicalEventLogEntry, CanonicalEventLogError> {
        self.append_json(
            "llm-tool-runtime",
            "tool-receipt",
            receipt.receipt_id.clone(),
            serde_json::to_value(receipt).map_err(|source| CanonicalEventLogError::Json {
                path: self.config.records_path.clone(),
                source,
            })?,
        )
    }

    pub fn append_forge_tool_receipt(
        &self,
        receipt: &CanonicalForgeToolReceiptV2,
    ) -> Result<CanonicalEventLogEntry, CanonicalEventLogError> {
        let event_log_receipt_id = format!(
            "semantic-memory-forge:forge-tool-receipt-v2:{}",
            receipt.receipt_id
        );
        self.append_json(
            "semantic-memory-forge",
            "forge-tool-receipt-v2",
            event_log_receipt_id,
            serde_json::to_value(receipt).map_err(|source| CanonicalEventLogError::Json {
                path: self.config.records_path.clone(),
                source,
            })?,
        )
    }

    pub fn append_control_receipt(
        &self,
        receipt: &CanonicalControlReceipt,
    ) -> Result<CanonicalEventLogEntry, CanonicalEventLogError> {
        self.append_json(
            "verification-control",
            "control-receipt",
            receipt.receipt_id.to_string(),
            serde_json::to_value(receipt).map_err(|source| CanonicalEventLogError::Json {
                path: self.config.records_path.clone(),
                source,
            })?,
        )
    }

    pub fn append_orchestration_report(
        &self,
        schema_name: impl Into<String>,
        report_id: impl Into<String>,
        body: Value,
    ) -> Result<CanonicalEventLogEntry, CanonicalEventLogError> {
        self.append_json("aidens-orchestration", schema_name, report_id, body)
    }

    pub fn append_json(
        &self,
        owner_crate: impl Into<String>,
        schema_name: impl Into<String>,
        receipt_id: impl Into<String>,
        body: Value,
    ) -> Result<CanonicalEventLogEntry, CanonicalEventLogError> {
        let owner_crate = owner_crate.into();
        let schema_name = schema_name.into();
        let receipt_id = receipt_id.into();
        let _lock = acquire_exclusive_lock(&self.config.records_path).map_err(|source| {
            CanonicalEventLogError::Io {
                path: lock_path_for(&self.config.records_path),
                source,
            }
        })?;
        let records = read_records(&self.config.records_path)?;
        if records.iter().any(|record| record.receipt_id == receipt_id) {
            return Err(CanonicalEventLogError::DuplicateReceiptId(receipt_id));
        }
        let sequence_number = records
            .last()
            .map(|record| record.sequence_number + 1)
            .unwrap_or(0);
        let previous_record_digest = records
            .last()
            .and_then(|record| record.record_digest.clone());
        let record = CanonicalEventLogEntry::new_with_chain(
            owner_crate,
            schema_name,
            receipt_id,
            body,
            sequence_number,
            previous_record_digest,
        )?;
        self.append_record_locked(&record)?;
        Ok(record)
    }

    pub fn append_record(
        &self,
        record: &CanonicalEventLogEntry,
    ) -> Result<(), CanonicalEventLogError> {
        let _lock = acquire_exclusive_lock(&self.config.records_path).map_err(|source| {
            CanonicalEventLogError::Io {
                path: lock_path_for(&self.config.records_path),
                source,
            }
        })?;
        let records = read_records(&self.config.records_path)?;
        if records
            .iter()
            .any(|existing| existing.receipt_id == record.receipt_id)
        {
            return Err(CanonicalEventLogError::DuplicateReceiptId(
                record.receipt_id.clone(),
            ));
        }
        let expected_sequence = records
            .last()
            .map(|existing| existing.sequence_number + 1)
            .unwrap_or(0);
        let expected_previous = records
            .last()
            .and_then(|existing| existing.record_digest.clone());
        if record.sequence_number != expected_sequence
            || record.previous_record_digest != expected_previous
            || !record.verify_digest()
            || !record.verify_record_digest()
        {
            return Err(CanonicalEventLogError::ChainVerificationFailed(
                record.sequence_number,
            ));
        }
        self.append_record_locked(record)
    }

    fn append_record_locked(
        &self,
        record: &CanonicalEventLogEntry,
    ) -> Result<(), CanonicalEventLogError> {
        let mut file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.config.records_path)
            .map_err(|source| CanonicalEventLogError::Io {
                path: self.config.records_path.clone(),
                source,
            })?;
        serde_json::to_writer(&mut file, record).map_err(|source| {
            CanonicalEventLogError::Json {
                path: self.config.records_path.clone(),
                source,
            }
        })?;
        file.write_all(b"\n")
            .map_err(|source| CanonicalEventLogError::Io {
                path: self.config.records_path.clone(),
                source,
            })?;
        file.flush().map_err(|source| CanonicalEventLogError::Io {
            path: self.config.records_path.clone(),
            source,
        })?;
        file.sync_all()
            .map_err(|source| CanonicalEventLogError::Io {
                path: self.config.records_path.clone(),
                source,
            })?;
        Ok(())
    }

    pub fn list_records(&self) -> Result<Vec<CanonicalEventLogEntry>, CanonicalEventLogError> {
        read_records(&self.config.records_path)
    }

    pub fn inspect(
        &self,
        receipt_id: &str,
    ) -> Result<CanonicalEventLogEntry, CanonicalEventLogError> {
        self.list_records()?
            .into_iter()
            .find(|record| record.receipt_id == receipt_id)
            .ok_or_else(|| CanonicalEventLogError::NotFound(receipt_id.into()))
    }

    pub fn verify_digest(&self, receipt_id: &str) -> Result<bool, CanonicalEventLogError> {
        Ok(self.inspect(receipt_id)?.verify_digest())
    }

    pub fn verify_chain(&self) -> Result<bool, CanonicalEventLogError> {
        let records = self.list_records()?;
        let mut previous_digest: Option<ContentDigest> = None;
        for (index, record) in records.iter().enumerate() {
            if record.sequence_number != index as u64
                || record.previous_record_digest != previous_digest
                || !record.verify_digest()
                || !record.verify_record_digest()
            {
                return Ok(false);
            }
            previous_digest = record.record_digest.clone();
        }
        Ok(true)
    }
}

#[async_trait]
impl ToolReceiptSink for CanonicalEventLog {
    async fn persist(&self, receipt: &CanonicalRuntimeToolReceipt) -> Result<(), ToolError> {
        self.append_runtime_tool_receipt(receipt)
            .map(|_| ())
            .map_err(|error| {
                ToolError::new(
                    ToolErrorClass::ReceiptPersistence,
                    format!("canonical receipt sink failed: {error}"),
                )
            })
    }
}

fn ensure_file(path: &Path) -> Result<(), CanonicalEventLogError> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).map_err(|source| CanonicalEventLogError::Io {
            path: parent.to_path_buf(),
            source,
        })?;
    }
    OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .map_err(|source| CanonicalEventLogError::Io {
            path: path.to_path_buf(),
            source,
        })?;
    Ok(())
}

fn ensure_run_bundle_file(path: &Path) -> Result<(), RunBundleStoreError> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).map_err(|source| RunBundleStoreError::Io {
            path: parent.to_path_buf(),
            source,
        })?;
    }
    OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .map_err(|source| RunBundleStoreError::Io {
            path: path.to_path_buf(),
            source,
        })?;
    Ok(())
}

fn read_run_bundle_records(path: &Path) -> Result<Vec<RunBundleStoreRecord>, RunBundleStoreError> {
    ensure_run_bundle_file(path)?;
    let file = File::open(path).map_err(|source| RunBundleStoreError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    let reader = BufReader::new(file);
    let mut records = Vec::new();
    for (index, line) in reader.lines().enumerate() {
        let line = line.map_err(|source| RunBundleStoreError::Io {
            path: path.to_path_buf(),
            source,
        })?;
        if line.trim().is_empty() {
            continue;
        }
        match serde_json::from_str(&line) {
            Ok(record) => records.push(record),
            Err(source) => {
                quarantine_corrupt_line(path, index + 1, &line, &source).map_err(|source| {
                    RunBundleStoreError::Io {
                        path: quarantine_path_for(path),
                        source,
                    }
                })?;
            }
        }
    }
    Ok(records)
}

fn write_atomic_file(path: &Path, body: &str) -> std::io::Result<()> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let tmp_path = path.with_file_name(format!(
        ".{}.tmp-{}-{}",
        path.file_name()
            .and_then(|name| name.to_str())
            .unwrap_or("receipt"),
        std::process::id(),
        Utc::now()
            .timestamp_nanos_opt()
            .unwrap_or_else(|| Utc::now().timestamp_micros())
    ));
    {
        let mut file = OpenOptions::new()
            .create_new(true)
            .write(true)
            .open(&tmp_path)?;
        file.write_all(body.as_bytes())?;
        file.flush()?;
        file.sync_all()?;
    }
    std::fs::rename(&tmp_path, path)?;
    if let Some(parent) = path.parent() {
        sync_directory(parent)?;
    }
    Ok(())
}

fn sync_directory(path: &Path) -> std::io::Result<()> {
    File::open(path)?.sync_all()
}

fn acquire_exclusive_lock(path: &Path) -> std::io::Result<ExclusiveFileLock> {
    let lock_path = lock_path_for(path);
    if let Some(parent) = lock_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let start = Instant::now();
    loop {
        match OpenOptions::new()
            .create_new(true)
            .write(true)
            .open(&lock_path)
        {
            Ok(mut file) => {
                writeln!(
                    file,
                    "pid={} acquired_at={}",
                    std::process::id(),
                    Utc::now().to_rfc3339()
                )?;
                file.sync_all()?;
                return Ok(ExclusiveFileLock { path: lock_path });
            }
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                if start.elapsed() >= Duration::from_secs(10) {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::TimedOut,
                        format!("receipt log lock still held at {}", lock_path.display()),
                    ));
                }
                thread::sleep(Duration::from_millis(10));
            }
            Err(error) => return Err(error),
        }
    }
}

fn lock_path_for(path: &Path) -> PathBuf {
    path.with_file_name(format!(
        "{}.lock",
        path.file_name()
            .and_then(|name| name.to_str())
            .unwrap_or("receipt-log")
    ))
}

struct ExclusiveFileLock {
    path: PathBuf,
}

impl Drop for ExclusiveFileLock {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.path);
    }
}

fn quarantine_path_for(path: &Path) -> PathBuf {
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("receipt-log");
    path.parent()
        .unwrap_or_else(|| Path::new("."))
        .join("quarantine")
        .join(format!("{file_name}.corrupt.ndjson"))
}

fn quarantine_corrupt_line(
    path: &Path,
    line_number: usize,
    raw_line: &str,
    error: &serde_json::Error,
) -> std::io::Result<()> {
    let quarantine_path = quarantine_path_for(path);
    if let Some(parent) = quarantine_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let _lock = acquire_exclusive_lock(&quarantine_path)?;
    let mut file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(&quarantine_path)?;
    let record = serde_json::json!({
        "artifact_kind": "aidens_local_corrupt_receipt_log_line_quarantine",
        "source_path": path.display().to_string(),
        "line_number": line_number,
        "observed_at": Utc::now().to_rfc3339(),
        "error": error.to_string(),
        "raw_line": raw_line,
    });
    serde_json::to_writer(&mut file, &record)?;
    file.write_all(b"\n")?;
    file.flush()?;
    file.sync_all()
}

fn read_records(path: &Path) -> Result<Vec<CanonicalEventLogEntry>, CanonicalEventLogError> {
    ensure_file(path)?;
    let file = File::open(path).map_err(|source| CanonicalEventLogError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    let reader = BufReader::new(file);
    let mut records = Vec::new();
    for (index, line) in reader.lines().enumerate() {
        let line = line.map_err(|source| CanonicalEventLogError::Io {
            path: path.to_path_buf(),
            source,
        })?;
        if line.trim().is_empty() {
            continue;
        }
        match serde_json::from_str(&line) {
            Ok(record) => records.push(record),
            Err(source) => {
                quarantine_corrupt_line(path, index + 1, &line, &source).map_err(|source| {
                    CanonicalEventLogError::Io {
                        path: quarantine_path_for(path),
                        source,
                    }
                })?;
            }
        }
    }
    Ok(records)
}

fn receipt_store_segment(value: &str) -> String {
    let mut out = String::new();
    for ch in value.chars() {
        if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
            out.push(ch.to_ascii_lowercase());
        } else {
            out.push('-');
        }
    }
    let out = out.trim_matches('-').to_string();
    if out.is_empty() {
        "run".into()
    } else {
        out
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn canonical_log_persists_owner_labeled_records() {
        let root = std::env::temp_dir().join(format!(
            "aidens-canonical-receipts-{}",
            uuid::Uuid::new_v4()
        ));
        let log = CanonicalEventLog::open(CanonicalEventLogConfig::for_root(&root)).unwrap();

        let record = log
            .append_orchestration_report(
                "operator-status-report",
                "report:test",
                serde_json::json!({"status": "ok"}),
            )
            .unwrap();

        assert_eq!(record.owner_crate, "aidens-orchestration");
        assert_eq!(record.sequence_number, 0);
        assert!(record.verify_digest());
        assert!(record.verify_record_digest());
        assert_eq!(log.inspect("report:test").unwrap(), record);
        assert!(log.verify_digest("report:test").unwrap());
        assert!(log.verify_chain().unwrap());
    }

    #[test]
    fn p28_canonical_log_digest_chain_detects_tampering() {
        let root = std::env::temp_dir().join(format!(
            "aidens-canonical-receipts-chain-{}",
            uuid::Uuid::new_v4()
        ));
        let log = CanonicalEventLog::open(CanonicalEventLogConfig::for_root(&root)).unwrap();
        let first = log
            .append_orchestration_report(
                "operator-status-report",
                "report:first",
                serde_json::json!({"status": "first"}),
            )
            .unwrap();
        let second = log
            .append_orchestration_report(
                "operator-status-report",
                "report:second",
                serde_json::json!({"status": "second"}),
            )
            .unwrap();

        assert_eq!(second.sequence_number, 1);
        assert_eq!(second.previous_record_digest, first.record_digest);
        assert!(log.verify_chain().unwrap());

        let path = log.config().records_path.clone();
        let mut lines = std::fs::read_to_string(&path)
            .unwrap()
            .lines()
            .map(str::to_string)
            .collect::<Vec<_>>();
        let mut tampered: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
        tampered["body"]["status"] = serde_json::json!("tampered");
        lines[0] = serde_json::to_string(&tampered).unwrap();
        std::fs::write(&path, lines.join("\n") + "\n").unwrap();

        let reopened = CanonicalEventLog::open(CanonicalEventLogConfig::for_root(&root)).unwrap();
        assert!(!reopened.verify_chain().unwrap());
        let _ = std::fs::remove_dir_all(root);
    }

    #[test]
    fn phase01_concurrent_canonical_appends_keep_single_digest_chain() {
        let root = std::env::temp_dir().join(format!(
            "aidens-canonical-receipts-concurrent-{}",
            uuid::Uuid::new_v4()
        ));
        let log = CanonicalEventLog::open(CanonicalEventLogConfig::for_root(&root)).unwrap();
        let mut handles = Vec::new();

        for index in 0..24 {
            let log = log.clone();
            handles.push(std::thread::spawn(move || {
                log.append_orchestration_report(
                    "operator-status-report",
                    format!("report:concurrent:{index}"),
                    serde_json::json!({ "index": index }),
                )
                .unwrap();
            }));
        }

        for handle in handles {
            handle.join().unwrap();
        }

        let records = log.list_records().unwrap();
        assert_eq!(records.len(), 24);
        assert!(log.verify_chain().unwrap());
        for (index, record) in records.iter().enumerate() {
            assert_eq!(record.sequence_number, index as u64);
            assert!(record.verify_record_digest());
        }
        let _ = std::fs::remove_dir_all(root);
    }

    #[test]
    fn phase01_corrupt_trailing_record_is_quarantined_not_history_poisoning() {
        let root = std::env::temp_dir().join(format!(
            "aidens-canonical-receipts-corrupt-{}",
            uuid::Uuid::new_v4()
        ));
        let log = CanonicalEventLog::open(CanonicalEventLogConfig::for_root(&root)).unwrap();
        log.append_orchestration_report(
            "operator-status-report",
            "report:before-corruption",
            serde_json::json!({"status": "ok"}),
        )
        .unwrap();
        {
            let mut file = OpenOptions::new()
                .append(true)
                .open(log.config().records_path.clone())
                .unwrap();
            file.write_all(b"{not-json}\n").unwrap();
            file.sync_all().unwrap();
        }

        let reopened = CanonicalEventLog::open(CanonicalEventLogConfig::for_root(&root)).unwrap();
        let records = reopened.list_records().unwrap();
        assert_eq!(records.len(), 1);
        assert!(reopened.verify_chain().unwrap());
        let quarantine = quarantine_path_for(&reopened.config().records_path);
        let quarantine_text = std::fs::read_to_string(quarantine).unwrap();
        assert!(quarantine_text.contains("aidens_local_corrupt_receipt_log_line_quarantine"));
        assert!(quarantine_text.contains("not-json"));
        let appended = reopened
            .append_orchestration_report(
                "operator-status-report",
                "report:after-corruption",
                serde_json::json!({"status": "continued"}),
            )
            .unwrap();
        assert_eq!(appended.sequence_number, 1);
        assert!(reopened.verify_chain().unwrap());
        let _ = std::fs::remove_dir_all(root);
    }

    #[test]
    fn phase01_duplicate_receipt_ids_are_rejected() {
        let root = std::env::temp_dir().join(format!(
            "aidens-canonical-receipts-duplicate-{}",
            uuid::Uuid::new_v4()
        ));
        let log = CanonicalEventLog::open(CanonicalEventLogConfig::for_root(&root)).unwrap();
        log.append_orchestration_report(
            "operator-status-report",
            "report:duplicate",
            serde_json::json!({"status": "first"}),
        )
        .unwrap();

        let err = log
            .append_orchestration_report(
                "operator-status-report",
                "report:duplicate",
                serde_json::json!({"status": "second"}),
            )
            .unwrap_err();
        assert!(matches!(
            err,
            CanonicalEventLogError::DuplicateReceiptId(id) if id == "report:duplicate"
        ));
        assert!(log.verify_chain().unwrap());
        let _ = std::fs::remove_dir_all(root);
    }

    #[test]
    fn run_bundle_store_persists_v3_operator_evidence() {
        let root =
            std::env::temp_dir().join(format!("aidens-run-bundle-store-{}", uuid::Uuid::new_v4()));
        let store = RunBundleStore::open(RunBundleStoreConfig::for_receipt_root(&root)).unwrap();
        let bundle = serde_json::json!({
            "schema": "AiDENsRunBundleV3",
            "run_id": "agent:fixture/run",
            "support": {
                "support_tier": "supported-local"
            },
            "failure": {
                "degraded": false
            }
        });

        let record = store.write_bundle_value(&bundle).unwrap();
        assert_eq!(record.support_tier, "supported-local");
        assert_eq!(record.semantic_status, "exact_check");
        assert!(record.bundle_path.exists());
        assert!(record
            .bundle_path
            .display()
            .to_string()
            .contains(record.content_digest.hex()));
        assert!(matches!(
            store.write_bundle_value(&bundle),
            Err(RunBundleStoreError::AlreadyExists(_))
        ));

        let reopened = RunBundleStore::open(RunBundleStoreConfig::for_receipt_root(&root)).unwrap();
        let inspection = reopened.inspect("agent:fixture/run").unwrap();
        assert!(inspection.digest_verified);
        assert_eq!(inspection.bundle["schema"], "AiDENsRunBundleV3");
        assert_eq!(reopened.single_bundle_path().unwrap(), record.bundle_path);
        let _ = std::fs::remove_dir_all(root);
    }
}