aidens_queue_kit/

lib.rs

//! Append-only queue substrate for P11 daemon execution.

use aidens_contracts::{
    ArtifactId, CanonicalToolSideEffectClass, DaemonNamespaceV1, DuplicateSuppressionReportV1,
    JobStateV1, JobV1, QueueHopKindV1, QueueHopReportV1, QueueLeaseV1, SafeModeOperationV1,
    SafeModeReportV1, ScheduleOccurrenceV1, WakeSignalV1,
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::thread;
use std::time::{Duration as StdDuration, Instant};
use thiserror::Error;

#[derive(Debug, Error)]
pub enum QueueError {
    #[error("queue io error at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    #[error("queue json error at {path}: {source}")]
    Json {
        path: PathBuf,
        #[source]
        source: serde_json::Error,
    },
    #[error("queue writer owner mismatch: expected {expected}, got {actual}")]
    WriterOwnerMismatch { expected: String, actual: String },
    #[error("queue opened read-only")]
    ReadOnly,
    #[error("job not found: {0}")]
    JobNotFound(String),
    #[error("job is terminal and cannot transition: {0}")]
    TerminalJob(String),
    #[error("job completion requires an active lease: {0}")]
    LeaseRequired(String),
    #[error("job lease mismatch for {job_id}: expected {expected}, got {actual}")]
    LeaseMismatch {
        job_id: String,
        expected: String,
        actual: String,
    },
    #[error("job lease expired for {job_id}: {lease_id}")]
    LeaseExpired { job_id: String, lease_id: String },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueLogEntryV1 {
    pub sequence: u64,
    pub namespace_id: ArtifactId,
    pub job: Option<JobV1>,
    pub lease: Option<QueueLeaseV1>,
    pub safe_mode_enabled: Option<bool>,
    pub queue_hop_receipt: Option<QueueHopReportV1>,
    pub duplicate_suppression_receipt: Option<DuplicateSuppressionReportV1>,
    pub safe_mode_receipt: Option<SafeModeReportV1>,
    pub recorded_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueSnapshotV1 {
    pub namespace: DaemonNamespaceV1,
    pub jobs: Vec<JobV1>,
    pub leases: Vec<QueueLeaseV1>,
    pub safe_mode_enabled: bool,
    pub records_seen: usize,
}

impl QueueSnapshotV1 {
    pub fn job(&self, job_id: &ArtifactId) -> Option<&JobV1> {
        self.jobs.iter().find(|job| &job.job_id == job_id)
    }

    pub fn logical_job_count_for(&self, idempotency_key: &str) -> usize {
        self.jobs
            .iter()
            .filter(|job| job.idempotency_key == idempotency_key)
            .count()
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueEnqueueOutcomeV1 {
    pub enqueued: bool,
    pub job: Option<JobV1>,
    pub existing_job: Option<JobV1>,
    pub queue_hop_receipt: QueueHopReportV1,
    pub duplicate_suppression_receipt: Option<DuplicateSuppressionReportV1>,
    pub safe_mode_receipt: Option<SafeModeReportV1>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueLeaseOutcomeV1 {
    pub job: JobV1,
    pub lease: QueueLeaseV1,
    pub queue_hop_receipt: QueueHopReportV1,
}

#[derive(Debug, Clone)]
pub struct DurableQueueLogV1 {
    namespace: DaemonNamespaceV1,
    root_path: PathBuf,
    log_path: PathBuf,
    writer_owner: Option<String>,
}

impl DurableQueueLogV1 {
    pub fn open(
        root_path: impl AsRef<Path>,
        namespace: DaemonNamespaceV1,
        writer_owner: impl Into<String>,
    ) -> Result<Self, QueueError> {
        let writer_owner = writer_owner.into();
        if writer_owner != namespace.daemon_owner {
            return Err(QueueError::WriterOwnerMismatch {
                expected: namespace.daemon_owner.clone(),
                actual: writer_owner,
            });
        }
        Self::open_inner(root_path, namespace, Some(writer_owner))
    }

    pub fn open_read_only(
        root_path: impl AsRef<Path>,
        namespace: DaemonNamespaceV1,
    ) -> Result<Self, QueueError> {
        Self::open_inner(root_path, namespace, None)
    }

    fn open_inner(
        root_path: impl AsRef<Path>,
        namespace: DaemonNamespaceV1,
        writer_owner: Option<String>,
    ) -> Result<Self, QueueError> {
        let root_path = root_path.as_ref().to_path_buf();
        std::fs::create_dir_all(&root_path).map_err(|source| QueueError::Io {
            path: root_path.clone(),
            source,
        })?;
        let log_path = root_path.join("queue.ndjson");
        ensure_file(&log_path)?;
        Ok(Self {
            namespace,
            root_path,
            log_path,
            writer_owner,
        })
    }

    pub fn namespace(&self) -> &DaemonNamespaceV1 {
        &self.namespace
    }

    pub fn root_path(&self) -> &Path {
        &self.root_path
    }

    pub fn enqueue_from_schedule(
        &self,
        occurrence: ScheduleOccurrenceV1,
    ) -> Result<QueueEnqueueOutcomeV1, QueueError> {
        let job = JobV1::new(
            occurrence.namespace_id.clone(),
            occurrence.idempotency_key.clone(),
            "schedule",
            occurrence.payload,
            occurrence.risk,
            Some(occurrence.occurrence_id),
            None,
        );
        self.enqueue_job(job)
    }

    pub fn enqueue_from_wake(
        &self,
        signal: WakeSignalV1,
    ) -> Result<QueueEnqueueOutcomeV1, QueueError> {
        let job = JobV1::new(
            signal.namespace_id.clone(),
            signal.idempotency_key.clone(),
            "wake",
            signal.payload,
            signal.risk,
            None,
            Some(signal.signal_id),
        );
        self.enqueue_job(job)
    }

    pub fn enqueue_job(&self, job: JobV1) -> Result<QueueEnqueueOutcomeV1, QueueError> {
        self.ensure_writer()?;
        let _lock = acquire_exclusive_lock(&self.log_path)?;
        let records = self.read_records()?;
        let snapshot = self.snapshot_from_records(&records);
        if snapshot.safe_mode_enabled && job.risk != CanonicalToolSideEffectClass::ReadOnly {
            let safe = SafeModeReportV1::new(
                self.namespace.namespace_id.clone(),
                SafeModeOperationV1::BlockedRiskyJob,
                true,
                Some(job.job_id.clone()),
                "safe-mode-blocked-new-risky-job",
            );
            let hop = QueueHopReportV1::new(
                self.namespace.namespace_id.clone(),
                job.job_id.clone(),
                None,
                QueueHopKindV1::SafeModeBlocked,
                None,
                JobStateV1::Queued,
                "safe-mode-blocked-new-risky-job",
            );
            self.append_record_locked(QueueLogEntryV1 {
                sequence: 0,
                namespace_id: self.namespace.namespace_id.clone(),
                job: None,
                lease: None,
                safe_mode_enabled: Some(true),
                queue_hop_receipt: Some(hop.clone()),
                duplicate_suppression_receipt: None,
                safe_mode_receipt: Some(safe.clone()),
                recorded_at: Utc::now(),
            })?;
            return Ok(QueueEnqueueOutcomeV1 {
                enqueued: false,
                job: Some(job),
                existing_job: None,
                queue_hop_receipt: hop,
                duplicate_suppression_receipt: None,
                safe_mode_receipt: Some(safe),
            });
        }

        if let Some(existing) = snapshot
            .jobs
            .iter()
            .find(|existing| existing.idempotency_key == job.idempotency_key)
            .cloned()
        {
            let duplicate = DuplicateSuppressionReportV1::new(
                self.namespace.namespace_id.clone(),
                job.idempotency_key.clone(),
                existing.job_id.clone(),
                job.source_kind,
            );
            let hop = QueueHopReportV1::new(
                self.namespace.namespace_id.clone(),
                existing.job_id.clone(),
                existing.lease_id.clone(),
                QueueHopKindV1::DuplicateSuppressed,
                Some(existing.state.clone()),
                existing.state.clone(),
                "duplicate-logical-job-suppressed",
            );
            self.append_record_locked(QueueLogEntryV1 {
                sequence: 0,
                namespace_id: self.namespace.namespace_id.clone(),
                job: None,
                lease: None,
                safe_mode_enabled: None,
                queue_hop_receipt: Some(hop.clone()),
                duplicate_suppression_receipt: Some(duplicate.clone()),
                safe_mode_receipt: None,
                recorded_at: Utc::now(),
            })?;
            return Ok(QueueEnqueueOutcomeV1 {
                enqueued: false,
                job: None,
                existing_job: Some(existing),
                queue_hop_receipt: hop,
                duplicate_suppression_receipt: Some(duplicate),
                safe_mode_receipt: None,
            });
        }

        let hop = QueueHopReportV1::new(
            self.namespace.namespace_id.clone(),
            job.job_id.clone(),
            None,
            QueueHopKindV1::Enqueued,
            None,
            JobStateV1::Queued,
            "job-enqueued",
        );
        self.append_record_locked(QueueLogEntryV1 {
            sequence: 0,
            namespace_id: self.namespace.namespace_id.clone(),
            job: Some(job.clone()),
            lease: None,
            safe_mode_enabled: None,
            queue_hop_receipt: Some(hop.clone()),
            duplicate_suppression_receipt: None,
            safe_mode_receipt: None,
            recorded_at: Utc::now(),
        })?;
        Ok(QueueEnqueueOutcomeV1 {
            enqueued: true,
            job: Some(job),
            existing_job: None,
            queue_hop_receipt: hop,
            duplicate_suppression_receipt: None,
            safe_mode_receipt: None,
        })
    }

    pub fn acquire_next_lease(
        &self,
        owner: impl Into<String>,
        ttl_seconds: i64,
    ) -> Result<Option<QueueLeaseOutcomeV1>, QueueError> {
        self.acquire_next_lease_at(owner, ttl_seconds, Utc::now())
    }

    pub fn acquire_next_lease_at(
        &self,
        owner: impl Into<String>,
        ttl_seconds: i64,
        now: DateTime<Utc>,
    ) -> Result<Option<QueueLeaseOutcomeV1>, QueueError> {
        self.ensure_writer()?;
        let owner = owner.into();
        let _lock = acquire_exclusive_lock(&self.log_path)?;
        let records = self.read_records()?;
        let snapshot = self.snapshot_from_records(&records);
        let leases_by_id = snapshot
            .leases
            .iter()
            .map(|lease| (lease.lease_id.clone(), lease.clone()))
            .collect::<BTreeMap<_, _>>();

        let mut jobs = snapshot.jobs;
        jobs.sort_by_key(|job| (job.created_at, job.job_id.0.clone()));
        for job in jobs {
            let stale_lease = job
                .lease_id
                .as_ref()
                .and_then(|lease_id| leases_by_id.get(lease_id))
                .filter(|lease| lease.is_expired_at(now))
                .cloned();
            let can_lease = matches!(job.state, JobStateV1::Queued | JobStateV1::Retrying)
                || stale_lease.is_some();
            if !can_lease || job.state.is_terminal() {
                continue;
            }
            let from_state = job.state.clone();
            let lease = QueueLeaseV1::new(
                &job,
                owner.clone(),
                ttl_seconds,
                stale_lease.as_ref().map(|lease| lease.lease_id.clone()),
            );
            let mut leased_job = job.with_state(
                JobStateV1::Leased,
                Some(lease.lease_id.clone()),
                if stale_lease.is_some() {
                    "lease-expired-stolen"
                } else {
                    "lease-acquired"
                },
            );
            leased_job.updated_at = now;
            let hop = QueueHopReportV1::new(
                self.namespace.namespace_id.clone(),
                leased_job.job_id.clone(),
                Some(lease.lease_id.clone()),
                if stale_lease.is_some() {
                    QueueHopKindV1::LeaseStolen
                } else {
                    QueueHopKindV1::LeaseAcquired
                },
                Some(from_state),
                JobStateV1::Leased,
                if stale_lease.is_some() {
                    "lease-expired-stolen"
                } else {
                    "lease-acquired"
                },
            );
            self.append_record_locked(QueueLogEntryV1 {
                sequence: 0,
                namespace_id: self.namespace.namespace_id.clone(),
                job: Some(leased_job.clone()),
                lease: Some(lease.clone()),
                safe_mode_enabled: None,
                queue_hop_receipt: Some(hop.clone()),
                duplicate_suppression_receipt: None,
                safe_mode_receipt: None,
                recorded_at: Utc::now(),
            })?;
            return Ok(Some(QueueLeaseOutcomeV1 {
                job: leased_job,
                lease,
                queue_hop_receipt: hop,
            }));
        }
        Ok(None)
    }

    pub fn cancel_job(
        &self,
        job_id: &ArtifactId,
        reason: impl Into<String>,
    ) -> Result<QueueHopReportV1, QueueError> {
        self.transition_job(
            job_id,
            JobStateV1::Cancelled,
            QueueHopKindV1::Cancelled,
            None,
            reason,
            Utc::now(),
        )
    }

    pub fn complete_job(
        &self,
        job_id: &ArtifactId,
        lease_id: Option<ArtifactId>,
    ) -> Result<QueueHopReportV1, QueueError> {
        self.complete_job_at(job_id, lease_id, Utc::now())
    }

    pub fn complete_job_at(
        &self,
        job_id: &ArtifactId,
        lease_id: Option<ArtifactId>,
        now: DateTime<Utc>,
    ) -> Result<QueueHopReportV1, QueueError> {
        self.transition_job(
            job_id,
            JobStateV1::Completed,
            QueueHopKindV1::Executed,
            lease_id,
            "job-executed",
            now,
        )
    }

    pub fn poison_job(
        &self,
        job_id: &ArtifactId,
        reason: impl Into<String>,
    ) -> Result<QueueHopReportV1, QueueError> {
        self.transition_job(
            job_id,
            JobStateV1::Poisoned,
            QueueHopKindV1::Poisoned,
            None,
            reason,
            Utc::now(),
        )
    }

    pub fn set_safe_mode(
        &self,
        enabled: bool,
        reason: impl Into<String>,
    ) -> Result<SafeModeReportV1, QueueError> {
        self.ensure_writer()?;
        let _lock = acquire_exclusive_lock(&self.log_path)?;
        let records = self.read_records()?;
        let snapshot = self.snapshot_from_records(&records);
        let receipt = SafeModeReportV1::new(
            self.namespace.namespace_id.clone(),
            if enabled {
                SafeModeOperationV1::Entered
            } else {
                SafeModeOperationV1::Exited
            },
            enabled,
            None,
            reason,
        );
        self.append_record_locked(QueueLogEntryV1 {
            sequence: 0,
            namespace_id: self.namespace.namespace_id.clone(),
            job: None,
            lease: None,
            safe_mode_enabled: Some(enabled),
            queue_hop_receipt: None,
            duplicate_suppression_receipt: None,
            safe_mode_receipt: Some(receipt.clone()),
            recorded_at: Utc::now(),
        })?;
        if enabled {
            for job in snapshot.jobs {
                if !job.state.is_terminal() && job.risk != CanonicalToolSideEffectClass::ReadOnly {
                    let from_state = job.state.clone();
                    let lease_id = job.lease_id.clone();
                    let next_job = job.with_state(
                        JobStateV1::Cancelled,
                        lease_id,
                        "safe-mode-quarantined-existing-risky-job",
                    );
                    let hop = QueueHopReportV1::new(
                        self.namespace.namespace_id.clone(),
                        next_job.job_id.clone(),
                        next_job.lease_id.clone(),
                        QueueHopKindV1::Drained,
                        Some(from_state),
                        JobStateV1::Cancelled,
                        "safe-mode-quarantined-existing-risky-job",
                    );
                    self.append_record_locked(QueueLogEntryV1 {
                        sequence: 0,
                        namespace_id: self.namespace.namespace_id.clone(),
                        job: Some(next_job),
                        lease: None,
                        safe_mode_enabled: None,
                        queue_hop_receipt: Some(hop),
                        duplicate_suppression_receipt: None,
                        safe_mode_receipt: None,
                        recorded_at: Utc::now(),
                    })?;
                }
            }
        }
        Ok(receipt)
    }

    pub fn drain_non_terminal(
        &self,
        reason: impl Into<String>,
    ) -> Result<Vec<QueueHopReportV1>, QueueError> {
        self.ensure_writer()?;
        let reason = reason.into();
        let jobs = self.snapshot()?.jobs;
        let mut receipts = Vec::new();
        for job in jobs {
            if !job.state.is_terminal() {
                receipts.push(self.transition_job(
                    &job.job_id,
                    JobStateV1::Cancelled,
                    QueueHopKindV1::Drained,
                    job.lease_id.clone(),
                    reason.clone(),
                    Utc::now(),
                )?);
            }
        }
        Ok(receipts)
    }

    pub fn snapshot(&self) -> Result<QueueSnapshotV1, QueueError> {
        let records = self.read_records()?;
        Ok(self.snapshot_from_records(&records))
    }

    fn snapshot_from_records(&self, records: &[QueueLogEntryV1]) -> QueueSnapshotV1 {
        let mut jobs_by_id = BTreeMap::<ArtifactId, JobV1>::new();
        let mut leases_by_id = BTreeMap::<ArtifactId, QueueLeaseV1>::new();
        let mut safe_mode_enabled = self.namespace.safe_mode_enabled;
        for record in records {
            if record.namespace_id != self.namespace.namespace_id {
                continue;
            }
            if let Some(job) = &record.job {
                jobs_by_id.insert(job.job_id.clone(), job.clone());
            }
            if let Some(lease) = &record.lease {
                leases_by_id.insert(lease.lease_id.clone(), lease.clone());
            }
            if let Some(enabled) = record.safe_mode_enabled {
                safe_mode_enabled = enabled;
            }
        }
        QueueSnapshotV1 {
            namespace: self.namespace.clone().with_safe_mode(safe_mode_enabled),
            jobs: jobs_by_id.into_values().collect(),
            leases: leases_by_id.into_values().collect(),
            safe_mode_enabled,
            records_seen: records.len(),
        }
    }

    fn transition_job(
        &self,
        job_id: &ArtifactId,
        to_state: JobStateV1,
        hop_kind: QueueHopKindV1,
        lease_id: Option<ArtifactId>,
        reason: impl Into<String>,
        now: DateTime<Utc>,
    ) -> Result<QueueHopReportV1, QueueError> {
        self.ensure_writer()?;
        let reason = reason.into();
        let _lock = acquire_exclusive_lock(&self.log_path)?;
        let records = self.read_records()?;
        let snapshot = self.snapshot_from_records(&records);
        let job = snapshot
            .job(job_id)
            .cloned()
            .ok_or_else(|| QueueError::JobNotFound(job_id.0.clone()))?;
        if job.state.is_terminal() {
            return Err(QueueError::TerminalJob(job_id.0.clone()));
        }
        if to_state == JobStateV1::Completed {
            let requested_lease_id = lease_id
                .clone()
                .ok_or_else(|| QueueError::LeaseRequired(job_id.0.clone()))?;
            let current_lease_id = job
                .lease_id
                .clone()
                .ok_or_else(|| QueueError::LeaseRequired(job_id.0.clone()))?;
            if requested_lease_id != current_lease_id {
                return Err(QueueError::LeaseMismatch {
                    job_id: job_id.0.clone(),
                    expected: current_lease_id.0,
                    actual: requested_lease_id.0,
                });
            }
            let lease = snapshot
                .leases
                .iter()
                .find(|lease| lease.lease_id == requested_lease_id && lease.active)
                .ok_or_else(|| QueueError::LeaseRequired(job_id.0.clone()))?;
            if lease.is_expired_at(now) {
                return Err(QueueError::LeaseExpired {
                    job_id: job_id.0.clone(),
                    lease_id: lease.lease_id.0.clone(),
                });
            }
        }
        let from_state = job.state.clone();
        let next_job = job.with_state(to_state.clone(), lease_id.clone(), reason.clone());
        let hop = QueueHopReportV1::new(
            self.namespace.namespace_id.clone(),
            next_job.job_id.clone(),
            lease_id,
            hop_kind,
            Some(from_state),
            to_state,
            reason,
        );
        self.append_record_locked(QueueLogEntryV1 {
            sequence: 0,
            namespace_id: self.namespace.namespace_id.clone(),
            job: Some(next_job),
            lease: None,
            safe_mode_enabled: None,
            queue_hop_receipt: Some(hop.clone()),
            duplicate_suppression_receipt: None,
            safe_mode_receipt: None,
            recorded_at: Utc::now(),
        })?;
        Ok(hop)
    }

    fn ensure_writer(&self) -> Result<(), QueueError> {
        match &self.writer_owner {
            Some(owner) if owner == &self.namespace.daemon_owner => Ok(()),
            Some(owner) => Err(QueueError::WriterOwnerMismatch {
                expected: self.namespace.daemon_owner.clone(),
                actual: owner.clone(),
            }),
            None => Err(QueueError::ReadOnly),
        }
    }

    fn append_record_locked(&self, mut record: QueueLogEntryV1) -> Result<(), QueueError> {
        record.sequence = self.read_records()?.len() as u64 + 1;
        let line = serde_json::to_string(&record).map_err(|source| QueueError::Json {
            path: self.log_path.clone(),
            source,
        })?;
        let mut file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.log_path)
            .map_err(|source| QueueError::Io {
                path: self.log_path.clone(),
                source,
            })?;
        writeln!(file, "{line}").map_err(|source| QueueError::Io {
            path: self.log_path.clone(),
            source,
        })?;
        file.sync_data().map_err(|source| QueueError::Io {
            path: self.log_path.clone(),
            source,
        })?;
        Ok(())
    }

    fn read_records(&self) -> Result<Vec<QueueLogEntryV1>, QueueError> {
        let file = File::open(&self.log_path).map_err(|source| QueueError::Io {
            path: self.log_path.clone(),
            source,
        })?;
        let reader = BufReader::new(file);
        let mut records = Vec::new();
        for line in reader.lines() {
            let line = line.map_err(|source| QueueError::Io {
                path: self.log_path.clone(),
                source,
            })?;
            if line.trim().is_empty() {
                continue;
            }
            records.push(
                serde_json::from_str::<QueueLogEntryV1>(&line).map_err(|source| {
                    QueueError::Json {
                        path: self.log_path.clone(),
                        source,
                    }
                })?,
            );
        }
        Ok(records)
    }
}

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

fn acquire_exclusive_lock(path: &Path) -> Result<ExclusiveQueueLock, QueueError> {
    let lock_path = path.with_file_name(format!(
        "{}.lock",
        path.file_name()
            .and_then(|name| name.to_str())
            .unwrap_or("queue")
    ));
    if let Some(parent) = lock_path.parent() {
        std::fs::create_dir_all(parent).map_err(|source| QueueError::Io {
            path: parent.to_path_buf(),
            source,
        })?;
    }
    let started = 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()
                )
                .map_err(|source| QueueError::Io {
                    path: lock_path.clone(),
                    source,
                })?;
                file.sync_all().map_err(|source| QueueError::Io {
                    path: lock_path.clone(),
                    source,
                })?;
                return Ok(ExclusiveQueueLock { lock_path });
            }
            Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => {
                if started.elapsed() > StdDuration::from_secs(10) {
                    return Err(QueueError::Io {
                        path: lock_path,
                        source: std::io::Error::new(
                            std::io::ErrorKind::TimedOut,
                            "queue lock still held",
                        ),
                    });
                }
                thread::sleep(StdDuration::from_millis(5));
            }
            Err(source) => {
                return Err(QueueError::Io {
                    path: lock_path,
                    source,
                });
            }
        }
    }
}

struct ExclusiveQueueLock {
    lock_path: PathBuf,
}

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

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

    fn temp_root(name: &str) -> PathBuf {
        let root = std::env::temp_dir().join(format!("aidens-p11-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        root
    }

    fn namespace(root: &Path) -> DaemonNamespaceV1 {
        DaemonNamespaceV1::new("test-namespace", root.display().to_string(), "daemon-a")
    }

    #[test]
    fn duplicate_schedule_occurrence_is_suppressed_by_idempotency_key() {
        let root = temp_root("duplicate");
        let ns = namespace(&root);
        let queue = DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap();
        let occurrence = ScheduleOccurrenceV1::new(
            ns.namespace_id.clone(),
            "daily",
            "2026-04-27T00:00:00Z",
            Utc::now(),
            serde_json::json!({"task":"refresh"}),
            CanonicalToolSideEffectClass::ReadOnly,
        );
        assert!(occurrence.identity_is_not_timestamp_only());

        let first = queue.enqueue_from_schedule(occurrence.clone()).unwrap();
        let second = queue.enqueue_from_schedule(occurrence.clone()).unwrap();

        assert!(first.enqueued);
        assert!(!second.enqueued);
        assert_eq!(
            second
                .duplicate_suppression_receipt
                .as_ref()
                .unwrap()
                .idempotency_key,
            occurrence.idempotency_key
        );
        assert_eq!(
            queue
                .snapshot()
                .unwrap()
                .logical_job_count_for(&occurrence.idempotency_key),
            1
        );
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn restart_preserves_queued_and_cancelled_jobs_without_resurrection() {
        let root = temp_root("restart");
        let ns = namespace(&root);
        let queue = DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap();
        let job = JobV1::new(
            ns.namespace_id.clone(),
            "manual:job-1",
            "manual",
            serde_json::json!({"task":"one"}),
            CanonicalToolSideEffectClass::ReadOnly,
            None,
            None,
        );
        let enqueued = queue.enqueue_job(job).unwrap().job.unwrap();
        queue
            .cancel_job(&enqueued.job_id, "operator-cancelled")
            .unwrap();

        let reopened = DurableQueueLogV1::open(&root, ns, "daemon-a").unwrap();
        let snapshot = reopened.snapshot().unwrap();
        let restored = snapshot.job(&enqueued.job_id).unwrap();
        assert_eq!(restored.state, JobStateV1::Cancelled);
        assert!(reopened
            .acquire_next_lease("daemon-a", 30)
            .unwrap()
            .is_none());
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn safe_mode_blocks_risky_jobs_and_allows_inspection_and_drain() {
        let root = temp_root("safe-mode");
        let ns = namespace(&root);
        let queue = DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap();
        let safe = queue.set_safe_mode(true, "operator-safe-mode").unwrap();
        assert!(safe.blocks_new_risky_jobs_but_allows_drain());

        let blocked = queue
            .enqueue_job(JobV1::new(
                ns.namespace_id.clone(),
                "shell:1",
                "manual",
                serde_json::json!({"cmd":"cargo check"}),
                CanonicalToolSideEffectClass::Admin,
                None,
                None,
            ))
            .unwrap();
        assert!(!blocked.enqueued);
        assert!(blocked.safe_mode_receipt.is_some());

        let read_only = queue
            .enqueue_job(JobV1::new(
                ns.namespace_id,
                "inspect:1",
                "manual",
                serde_json::json!({"inspect":true}),
                CanonicalToolSideEffectClass::ReadOnly,
                None,
                None,
            ))
            .unwrap();
        assert!(read_only.enqueued);
        assert_eq!(queue.snapshot().unwrap().jobs.len(), 1);
        assert_eq!(
            queue.drain_non_terminal("safe-mode-drain").unwrap().len(),
            1
        );
        assert_eq!(
            queue.snapshot().unwrap().jobs[0].state,
            JobStateV1::Cancelled
        );
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn expired_lease_can_be_stolen_once_without_duplicate_job() {
        let root = temp_root("lease");
        let ns = namespace(&root);
        let queue = DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap();
        let job = JobV1::new(
            ns.namespace_id,
            "lease:job",
            "manual",
            serde_json::json!({"work":true}),
            CanonicalToolSideEffectClass::ReadOnly,
            None,
            None,
        );
        let enqueued = queue.enqueue_job(job).unwrap().job.unwrap();
        let first = queue.acquire_next_lease("daemon-a", 1).unwrap().unwrap();
        let stolen = queue
            .acquire_next_lease_at("daemon-b", 30, Utc::now() + Duration::seconds(5))
            .unwrap()
            .unwrap();
        assert_eq!(stolen.job.job_id, enqueued.job_id);
        assert_eq!(stolen.lease.stolen_from, Some(first.lease.lease_id));
        assert_eq!(queue.snapshot().unwrap().jobs.len(), 1);
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn concurrent_enqueue_is_single_writer_and_idempotent() {
        let root = temp_root("concurrent-enqueue");
        let ns = namespace(&root);
        let queue = Arc::new(DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap());
        let barrier = Arc::new(Barrier::new(16));
        let mut handles = Vec::new();

        for _ in 0..16 {
            let queue = Arc::clone(&queue);
            let barrier = Arc::clone(&barrier);
            let namespace_id = ns.namespace_id.clone();
            handles.push(std::thread::spawn(move || {
                barrier.wait();
                let job = JobV1::new(
                    namespace_id,
                    "concurrent:one-logical-job",
                    "manual",
                    serde_json::json!({"work":true}),
                    CanonicalToolSideEffectClass::ReadOnly,
                    None,
                    None,
                );
                queue.enqueue_job(job).unwrap().enqueued
            }));
        }

        let enqueued_count = handles
            .into_iter()
            .map(|handle| handle.join().unwrap())
            .filter(|enqueued| *enqueued)
            .count();
        let snapshot = queue.snapshot().unwrap();
        let records = queue.read_records().unwrap();

        assert_eq!(enqueued_count, 1);
        assert_eq!(
            snapshot.logical_job_count_for("concurrent:one-logical-job"),
            1
        );
        assert_eq!(snapshot.jobs.len(), 1);
        assert_eq!(records.len(), 16);
        assert_eq!(
            records
                .iter()
                .map(|record| record.sequence)
                .collect::<Vec<_>>(),
            (1..=records.len() as u64).collect::<Vec<_>>()
        );
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn concurrent_lease_acquisition_grants_one_active_lease() {
        let root = temp_root("concurrent-lease");
        let ns = namespace(&root);
        let queue = Arc::new(DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap());
        queue
            .enqueue_job(JobV1::new(
                ns.namespace_id,
                "lease:single-active",
                "manual",
                serde_json::json!({"work":true}),
                CanonicalToolSideEffectClass::ReadOnly,
                None,
                None,
            ))
            .unwrap();

        let barrier = Arc::new(Barrier::new(12));
        let mut handles = Vec::new();
        for worker in 0..12 {
            let queue = Arc::clone(&queue);
            let barrier = Arc::clone(&barrier);
            handles.push(std::thread::spawn(move || {
                barrier.wait();
                queue
                    .acquire_next_lease(format!("daemon-worker-{worker}"), 30)
                    .unwrap()
                    .is_some()
            }));
        }

        let lease_count = handles
            .into_iter()
            .map(|handle| handle.join().unwrap())
            .filter(|leased| *leased)
            .count();
        let snapshot = queue.snapshot().unwrap();

        assert_eq!(lease_count, 1);
        assert_eq!(snapshot.jobs.len(), 1);
        assert_eq!(snapshot.jobs[0].state, JobStateV1::Leased);
        assert_eq!(
            snapshot.leases.iter().filter(|lease| lease.active).count(),
            1
        );
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn late_completion_after_ttl_is_rejected_and_job_stays_leased() {
        let root = temp_root("late-completion");
        let ns = namespace(&root);
        let queue = DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap();
        queue
            .enqueue_job(JobV1::new(
                ns.namespace_id,
                "lease:late-completion",
                "manual",
                serde_json::json!({"work":true}),
                CanonicalToolSideEffectClass::ReadOnly,
                None,
                None,
            ))
            .unwrap();

        let now = Utc::now();
        let leased = queue
            .acquire_next_lease_at("daemon-a", 1, now)
            .unwrap()
            .unwrap();
        let err = queue
            .complete_job_at(
                &leased.job.job_id,
                Some(leased.lease.lease_id.clone()),
                now + Duration::seconds(2),
            )
            .unwrap_err();
        assert!(matches!(err, QueueError::LeaseExpired { .. }));

        let snapshot = queue.snapshot().unwrap();
        let job = snapshot.job(&leased.job.job_id).unwrap();
        assert_eq!(job.state, JobStateV1::Leased);
        assert_eq!(job.lease_id, Some(leased.lease.lease_id));
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn stale_lease_cannot_complete_after_lease_is_stolen() {
        let root = temp_root("stale-completion");
        let ns = namespace(&root);
        let queue = DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap();
        queue
            .enqueue_job(JobV1::new(
                ns.namespace_id,
                "lease:stale-completion",
                "manual",
                serde_json::json!({"work":true}),
                CanonicalToolSideEffectClass::ReadOnly,
                None,
                None,
            ))
            .unwrap();
        let first = queue.acquire_next_lease("daemon-a", 1).unwrap().unwrap();
        let second = queue
            .acquire_next_lease_at("daemon-b", 30, Utc::now() + Duration::seconds(5))
            .unwrap()
            .unwrap();

        let err = queue
            .complete_job_at(
                &first.job.job_id,
                Some(first.lease.lease_id.clone()),
                Utc::now() + Duration::seconds(6),
            )
            .unwrap_err();
        assert!(matches!(err, QueueError::LeaseMismatch { .. }));
        assert_eq!(
            queue
                .snapshot()
                .unwrap()
                .job(&first.job.job_id)
                .unwrap()
                .lease_id,
            Some(second.lease.lease_id)
        );
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn entering_safe_mode_quarantines_existing_risky_jobs() {
        let root = temp_root("safe-mode-existing");
        let ns = namespace(&root);
        let queue = DurableQueueLogV1::open(&root, ns.clone(), "daemon-a").unwrap();
        let risky = queue
            .enqueue_job(JobV1::new(
                ns.namespace_id.clone(),
                "admin:queued-before-safe-mode",
                "manual",
                serde_json::json!({"cmd":"write"}),
                CanonicalToolSideEffectClass::Admin,
                None,
                None,
            ))
            .unwrap()
            .job
            .unwrap();
        let read_only = queue
            .enqueue_job(JobV1::new(
                ns.namespace_id,
                "read:queued-before-safe-mode",
                "manual",
                serde_json::json!({"inspect":true}),
                CanonicalToolSideEffectClass::ReadOnly,
                None,
                None,
            ))
            .unwrap()
            .job
            .unwrap();

        queue.set_safe_mode(true, "operator-safe-mode").unwrap();

        let snapshot = queue.snapshot().unwrap();
        let risky_after = snapshot.job(&risky.job_id).unwrap();
        let read_after = snapshot.job(&read_only.job_id).unwrap();
        assert_eq!(risky_after.state, JobStateV1::Cancelled);
        assert!(risky_after
            .reason_codes
            .contains(&"safe-mode-quarantined-existing-risky-job".to_string()));
        assert_eq!(read_after.state, JobStateV1::Queued);
        let _ = std::fs::remove_dir_all(&root);
    }
}