projectatlas/runtime/

source_observation.rs

//! Process-local source observation for long-lived verified index reads.

use super::{
    ScanRuntimePlan, SourceVerificationWork, WatchChangeSet,
    open_atlas_store_read_only_for_project, open_exact_fresh_atlas_store_for_project_controlled,
    open_exact_saved_source_matches_index_controlled, publication_input_error,
    source_changed_during_derivation, source_inspection_error,
    verify_saved_source_matches_index_controlled,
};
use crate::CliError;
use blake3::Hasher;
use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use projectatlas_core::graph::ProjectInstanceId;
use projectatlas_core::{CanonicalProjectRoot, IndexGeneration, IndexWorkControl, IndexWorkStage};
use projectatlas_db::{AtlasStore, CapturedProjectBinding, IndexPublicationState};
use std::collections::HashMap;
#[cfg(windows)]
use std::ffi::{OsStr, OsString};
use std::fmt;
use std::fs;
use std::fs::File;
use std::hash::Hash;
use std::io::{Read, Take};
#[cfg(windows)]
use std::os::windows::ffi::{OsStrExt, OsStringExt};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{Receiver, SyncSender, TryRecvError, TrySendError, sync_channel};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

/// Maximum independent project bindings observed by one MCP process.
const SOURCE_OBSERVATION_CAPACITY: usize = 16;
/// Maximum unprocessed native watcher events retained per project.
const SOURCE_OBSERVATION_QUEUE_CAPACITY: usize = 1_024;
/// Fixed attempts used to reconcile an edit that overlaps a query or verification.
const VERIFIED_READ_ATTEMPTS: usize = 3;
/// Maximum bytes read from any one external source-selection policy input.
const MAX_POLICY_INPUT_BYTES: u64 = 16 * 1_024 * 1_024;

/// Exact process-local evidence attached to one accepted `SQLite` read snapshot.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct VerifiedReadStamp {
    /// Nonce distinguishing this server process from every restart.
    pub(crate) process_nonce: [u8; 16],
    /// Monotonic verified epoch inside this process and source binding.
    pub(crate) epoch: u64,
    /// Durable complete publication generation read by the `SQLite` snapshot.
    pub(crate) generation: IndexGeneration,
    /// Durable project identity captured by the same root-bound snapshot.
    pub(crate) project_instance_id: ProjectInstanceId,
}

/// Measured freshness work consumed before one result was accepted.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub(crate) struct VerifiedReadWork {
    /// Exact full-source verifications performed for this accepted call.
    pub(crate) exact_verifications: u64,
    /// Repository entries inspected by exact verification.
    pub(crate) filesystem_entries: u64,
    /// Repository source bytes hashed by exact verification.
    pub(crate) filesystem_bytes: u64,
    /// `SQLite` read statements owned directly by freshness verification.
    pub(crate) sqlite_read_statements: u64,
    /// Full indexed-node rows decoded by freshness verification.
    pub(crate) decoded_nodes: u64,
    /// Provisional query results discarded after an overlapping invalidation.
    pub(crate) retries: u64,
    /// Wall time for verification, bounded query construction, and acceptance.
    pub(crate) elapsed: Duration,
    /// Rendered result bytes supplied by the adapter after acceptance.
    pub(crate) output_bytes: u64,
}

impl VerifiedReadWork {
    /// Accumulate one exact source verification into this accepted call's work.
    fn add_exact(&mut self, work: SourceVerificationWork) {
        self.exact_verifications = self.exact_verifications.saturating_add(1);
        self.filesystem_entries = self
            .filesystem_entries
            .saturating_add(work.filesystem_entries);
        self.filesystem_bytes = self.filesystem_bytes.saturating_add(work.filesystem_bytes);
        self.sqlite_read_statements = self
            .sqlite_read_statements
            .saturating_add(work.sqlite_read_statements);
        self.decoded_nodes = self.decoded_nodes.saturating_add(work.decoded_nodes);
    }
}

/// Accepted owned result plus its epoch stamp and measured freshness work.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct VerifiedReadOutcome<T> {
    /// Result built entirely inside the accepted `SQLite` snapshot.
    pub(crate) value: T,
    /// Process epoch and durable generation that accepted the result.
    pub(crate) stamp: VerifiedReadStamp,
    /// Work consumed before the result was accepted.
    pub(crate) work: VerifiedReadWork,
}

impl<T> VerifiedReadOutcome<T> {
    /// Record the already-rendered output size without rescanning or reopening `SQLite`.
    pub(crate) fn with_output_bytes(mut self, output_bytes: usize) -> Self {
        self.work.output_bytes = u64::try_from(output_bytes).unwrap_or(u64::MAX);
        self
    }
}

/// Source authority retained through one purpose mutation commit boundary.
enum MutationSourceWitness {
    /// Native observation plus the accepted source epoch.
    Observed {
        /// Exact source binding and observer continuity retained by this admission.
        entry: Arc<SourceObservationEntry>,
        /// Exact durable generation and policy epoch accepted before mutation.
        epoch: VerifiedSourceEpoch,
    },
    /// Exact-per-call compatibility witness when no observer can be admitted.
    Exact {
        /// Canonical source, database, and configuration identity.
        binding: SourceBinding,
        /// Durable generation and project identity accepted before mutation.
        stamp: VerifiedReadStamp,
        /// Scanner contract accepted around exact source verification.
        contract_fingerprint: String,
        /// Dynamic source-selection policy accepted around exact verification.
        policy_witness: String,
    },
}

/// Saved-source witness retained through one purpose mutation commit boundary.
pub(crate) struct VerifiedMutationAdmission {
    /// Observed or exact-per-call source authority retained by this admission.
    witness: MutationSourceWitness,
    /// Cooperative cancellation and work budget shared through commit admission.
    control: IndexWorkControl,
}

impl VerifiedMutationAdmission {
    /// Revalidate exact saved source, policy, observer continuity, and cancellation before commit.
    pub(crate) fn verify(&self) -> Result<(), CliError> {
        match &self.witness {
            MutationSourceWitness::Observed { entry, epoch } => {
                Self::verify_observed(entry, epoch, &self.control, || {
                    verify_saved_source_matches_index_controlled(
                        &entry.binding.database,
                        &entry.binding.root,
                        entry.binding.config.as_deref(),
                        &self.control,
                    )
                })
            }
            MutationSourceWitness::Exact {
                binding,
                stamp,
                contract_fingerprint,
                policy_witness,
            } => Self::verify_exact(
                binding,
                stamp,
                contract_fingerprint,
                policy_witness,
                &self.control,
            ),
        }
    }

    /// Revalidate an observer-backed mutation witness around exact saved source.
    fn verify_observed(
        entry: &SourceObservationEntry,
        epoch: &VerifiedSourceEpoch,
        control: &IndexWorkControl,
        verify_source: impl FnOnce() -> Result<(), CliError>,
    ) -> Result<(), CliError> {
        if let Err(error) = Self::verify_observation(entry, epoch, control) {
            entry.invalidate_epoch(epoch);
            return Err(error);
        }
        if let Err(error) = verify_source() {
            if matches!(error, CliError::RefreshRequired(_)) {
                entry.invalidate_after_proven_source_mismatch()?;
            } else {
                entry.invalidate_epoch(epoch);
            }
            return Err(error);
        }
        if let Err(error) = Self::verify_observation(entry, epoch, control) {
            entry.invalidate_epoch(epoch);
            return Err(error);
        }
        Ok(())
    }

    /// Require the retained observer epoch to remain current.
    fn verify_observation(
        entry: &SourceObservationEntry,
        epoch: &VerifiedSourceEpoch,
        control: &IndexWorkControl,
    ) -> Result<(), CliError> {
        match SourceObservationRegistry::accepts_observed_result(entry, epoch, control) {
            Ok(ObservedAcceptance::Accepted) => Ok(()),
            Ok(ObservedAcceptance::Superseded | ObservedAcceptance::Invalidated) => {
                Err(source_changed_during_derivation(&entry.binding.root, "."))
            }
            Err(error) => Err(error),
        }
    }

    /// Revalidate exact saved source plus durable identity immediately before commit.
    fn verify_exact(
        binding: &SourceBinding,
        stamp: &VerifiedReadStamp,
        contract_fingerprint: &str,
        policy_witness: &str,
        control: &IndexWorkControl,
    ) -> Result<(), CliError> {
        let before = exact_source_policy(binding, control)?;
        if before.0 != contract_fingerprint || before.1 != policy_witness {
            return Err(source_changed_during_derivation(&binding.root, "."));
        }
        let exact = open_exact_saved_source_matches_index_controlled(
            &binding.database,
            &binding.root,
            binding.config.as_deref(),
            control,
        )?;
        let verification = (|| {
            let publication = exact
                .store
                .index_publication()?
                .ok_or_else(|| source_changed_during_derivation(&binding.root, "."))?;
            let captured = exact.store.captured_project_binding()?;
            let after = exact_source_policy(binding, control)?;
            Ok::<_, CliError>((publication, captured, after))
        })();
        let finished = exact.store.finish_index_read_snapshot();
        let (publication, captured, after) = verification?;
        finished?;
        if before == after
            && after.0 == contract_fingerprint
            && after.1 == policy_witness
            && publication.generation == stamp.generation
            && captured.project_instance_id == stamp.project_instance_id
        {
            Ok(())
        } else {
            Err(source_changed_during_derivation(&binding.root, "."))
        }
    }
}

/// Acceptance of one observed result against the shared source epoch.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ObservedAcceptance {
    /// The retained witness is still current.
    Accepted,
    /// Another reconciliation replaced the witness without invalid source evidence.
    Superseded,
    /// Source, policy, or observation continuity no longer supports the witness.
    Invalidated,
}

/// Exact root/database/config identity used to isolate one observer.
#[derive(Clone, Debug, Eq)]
struct SourceBinding {
    /// Canonical source root watched by this observer.
    root: PathBuf,
    /// Database excluded from source-change invalidation.
    database: PathBuf,
    /// Optional scanner configuration watched with the source root.
    config: Option<PathBuf>,
}

impl PartialEq for SourceBinding {
    fn eq(&self, other: &Self) -> bool {
        self.root == other.root && self.database == other.database && self.config == other.config
    }
}

impl Hash for SourceBinding {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.root.hash(state);
        self.database.hash(state);
        self.config.hash(state);
    }
}

impl SourceBinding {
    /// Configuration directory observed separately when it lies outside the source root.
    fn external_config_parent(&self) -> Option<&Path> {
        self.config
            .as_deref()
            .filter(|config| !config.starts_with(&self.root))
            .and_then(Path::parent)
    }

    /// Resolve one root, database, and optional configuration into a stable binding.
    fn new(database: &Path, root: &Path, config: Option<&Path>) -> Result<Self, CliError> {
        let root = root.canonicalize().map_err(|source| CliError::Io {
            path: root.to_path_buf(),
            source,
        })?;
        let database = absolute_path_from(&root, database);
        let config = config.map(|path| absolute_path_from(&root, path));
        Ok(Self {
            root,
            database,
            config,
        })
    }
}

/// One successfully verified source epoch retained only by this process.
#[derive(Clone, Debug, Eq, PartialEq)]
struct VerifiedSourceEpoch {
    /// Accepted process and durable publication identity.
    stamp: VerifiedReadStamp,
    /// Latest watcher ingress sequence covered by this epoch.
    ingress_sequence: u64,
    /// Source publication contract covered by this epoch.
    contract_fingerprint: String,
    /// Exact external source-selection policy covered by this epoch.
    policy_witness: String,
}

/// Mutable verified-epoch state serialized for one source binding.
#[derive(Debug, Default)]
struct ObservationState {
    /// Monotonic epoch sequence within the owning observer.
    next_epoch: u64,
    /// Last source epoch proven safe for warm reads.
    verified: Option<VerifiedSourceEpoch>,
}

/// Concrete watcher and state for one exact source binding.
struct SourceObservationEntry {
    /// Exact source, database, and configuration identity.
    binding: SourceBinding,
    /// Native watcher kept alive for the entry lifetime.
    watcher: RecommendedWatcher,
    /// Bounded receiver whose lock serializes callback publication with drain acknowledgment.
    receiver: Arc<Mutex<Receiver<Event>>>,
    /// Monotonic count of relevant events accepted by the callback.
    ingress_sequence: Arc<AtomicU64>,
    /// Whether overflow, disconnection, or rescan invalidated event continuity.
    continuity_lost: Arc<AtomicBool>,
    /// Serializes exact verification and epoch installation for this binding.
    reconcile: Mutex<()>,
    /// Current process-local verified epoch.
    state: Mutex<ObservationState>,
    #[cfg(test)]
    /// Deterministic event ingress used only by owning tests.
    test_sender: SyncSender<Event>,
    #[cfg(test)]
    /// One event injected after an acceptance drain for owning race tests.
    acceptance_event: Mutex<Option<Event>>,
    #[cfg(test)]
    /// Continuity loss injected during the next locked event drain for owning race tests.
    drain_continuity_invalidations: AtomicU64,
}

#[cfg(windows)]
impl Drop for SourceObservationEntry {
    fn drop(&mut self) {
        drop(self.watcher.unwatch(&self.binding.root));
        if let Some(parent) = self.binding.external_config_parent() {
            drop(self.watcher.unwatch(parent));
        }
        // notify's Windows Unwatch and Drop only enqueue actions. Configure waits
        // for an acknowledgement after preceding cancellation/completion work,
        // so directory cleanup cannot race an outstanding native notification.
        drop(self.watcher.configure(notify::Config::default()));
    }
}

impl fmt::Debug for SourceObservationEntry {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("SourceObservationEntry")
            .field("binding", &self.binding)
            .field(
                "ingress_sequence",
                &self.ingress_sequence.load(Ordering::Acquire),
            )
            .field(
                "continuity_lost",
                &self.continuity_lost.load(Ordering::Acquire),
            )
            .finish_non_exhaustive()
    }
}

impl SourceObservationEntry {
    /// Start a bounded native watcher for one source binding.
    fn start(binding: SourceBinding) -> Result<Self, CliError> {
        let (sender, receiver) = sync_channel(SOURCE_OBSERVATION_QUEUE_CAPACITY);
        let receiver = Arc::new(Mutex::new(receiver));
        #[cfg(test)]
        let test_sender = sender.clone();
        let ingress_sequence = Arc::new(AtomicU64::new(0));
        let continuity_lost = Arc::new(AtomicBool::new(false));
        let watcher = notify::recommended_watcher(watcher_callback(
            sender,
            Arc::clone(&receiver),
            Arc::clone(&ingress_sequence),
            Arc::clone(&continuity_lost),
        ))
        .map_err(|source| observer_error(&binding.root, &source))?;
        let mut entry = Self {
            binding,
            watcher,
            receiver,
            ingress_sequence,
            continuity_lost,
            reconcile: Mutex::new(()),
            state: Mutex::new(ObservationState::default()),
            #[cfg(test)]
            test_sender,
            #[cfg(test)]
            acceptance_event: Mutex::new(None),
            #[cfg(test)]
            drain_continuity_invalidations: AtomicU64::new(0),
        };
        entry
            .watcher
            .watch(&entry.binding.root, RecursiveMode::Recursive)
            .map_err(|source| observer_error(&entry.binding.root, &source))?;
        if let Some(parent) = entry.binding.external_config_parent() {
            entry
                .watcher
                .watch(parent, RecursiveMode::NonRecursive)
                .map_err(|source| observer_error(&entry.binding.root, &source))?;
        }
        Ok(entry)
    }

    /// Publish one deterministic test event through the production handshake.
    #[cfg(test)]
    fn publish_test_event(&self, event: Event) -> Result<(), CliError> {
        let _receiver = self
            .receiver
            .lock()
            .map_err(|_poisoned| lock_error(&self.binding.root, "source observation receiver"))?;
        self.test_sender.try_send(event).map_err(|source| {
            CliError::InvalidInput(format!(
                "deterministic source observer event injection failed: {source}"
            ))
        })?;
        self.ingress_sequence.fetch_add(1, Ordering::AcqRel);
        Ok(())
    }

    /// Discard the current verified epoch after any uncertainty.
    fn invalidate(&self) {
        if let Ok(mut state) = self.state.lock() {
            state.verified = None;
        }
    }

    /// Discard only the failed operation's epoch without clearing a successor.
    fn invalidate_epoch(&self, epoch: &VerifiedSourceEpoch) {
        if let Ok(mut state) = self.state.lock()
            && state
                .verified
                .as_ref()
                .is_some_and(|current| current.stamp == epoch.stamp)
        {
            state.verified = None;
        }
    }

    /// Reject all cached and in-flight evidence after exact source verification fails.
    fn invalidate_after_proven_source_mismatch(&self) -> Result<(), CliError> {
        let _receiver = self
            .receiver
            .lock()
            .map_err(|_poisoned| lock_error(&self.binding.root, "source observation receiver"))?;
        self.continuity_lost.store(true, Ordering::Release);
        self.invalidate();
        Ok(())
    }

    /// Reset continuity and drain events before sampling exact source truth.
    fn clear_before_exact_verification(&self) -> Result<(), CliError> {
        let receiver = self
            .receiver
            .lock()
            .map_err(|_poisoned| lock_error(&self.binding.root, "source observation receiver"))?;
        self.invalidate();
        self.continuity_lost.store(false, Ordering::Release);
        loop {
            match receiver.try_recv() {
                Ok(_event) => {}
                Err(TryRecvError::Empty) => return Ok(()),
                Err(TryRecvError::Disconnected) => {
                    self.continuity_lost.store(true, Ordering::Release);
                    return Ok(());
                }
            }
        }
    }

    /// Return whether any potentially relevant event arrived since exact truth was sampled.
    fn changed_since_exact_verification(
        &self,
        scan_options: &projectatlas_fs::ScanOptions,
    ) -> Result<bool, CliError> {
        let receiver = self
            .receiver
            .lock()
            .map_err(|_poisoned| lock_error(&self.binding.root, "source observation receiver"))?;
        self.drain_source_events(&receiver, scan_options)
    }

    /// Drain and invalidate atomically with callback publication and epoch installation.
    fn drain_source_events(
        &self,
        receiver: &Receiver<Event>,
        scan_options: &projectatlas_fs::ScanOptions,
    ) -> Result<bool, CliError> {
        #[cfg(test)]
        if self
            .drain_continuity_invalidations
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
                remaining.checked_sub(1)
            })
            .is_ok()
        {
            self.continuity_lost.store(true, Ordering::Release);
        }
        if self.continuity_lost.load(Ordering::Acquire) {
            self.invalidate();
            return Ok(true);
        }
        let mut changes = WatchChangeSet::default();
        loop {
            match receiver.try_recv() {
                Ok(event) => {
                    let event_changes = observer_event_changes(&self.binding, scan_options, &event);
                    changes.requires_full_scan |= event_changes.requires_full_scan;
                    changes.paths.extend(event_changes.paths);
                    if self
                        .binding
                        .config
                        .as_ref()
                        .is_some_and(|config| event.paths.iter().any(|path| path == config))
                    {
                        changes.requires_full_scan = true;
                    }
                }
                Err(TryRecvError::Empty) => break,
                Err(TryRecvError::Disconnected) => {
                    self.continuity_lost.store(true, Ordering::Release);
                    self.invalidate();
                    return Ok(true);
                }
            }
        }
        let changed = changes.requires_full_scan || !changes.paths.is_empty();
        if changed {
            // Preserve consumed changes for an exact verification already in flight.
            self.continuity_lost.store(true, Ordering::Release);
            self.invalidate();
        } else {
            let acknowledged = self.ingress_sequence.load(Ordering::Acquire);
            let mut state = self
                .state
                .lock()
                .map_err(|_poisoned| lock_error(&self.binding.root, "source observation state"))?;
            if let Some(epoch) = state.verified.as_mut() {
                epoch.ingress_sequence = acknowledged;
            }
        }
        Ok(changed)
    }

    /// Consume events only while the sampled epoch still owns their selection policy.
    fn drain_epoch_events(
        &self,
        receiver: &Receiver<Event>,
        epoch: &VerifiedSourceEpoch,
        scan_options: &projectatlas_fs::ScanOptions,
    ) -> Result<ObservedAcceptance, CliError> {
        let Some(current) = self.current_epoch()? else {
            return Ok(ObservedAcceptance::Invalidated);
        };
        if current.stamp != epoch.stamp {
            return Ok(ObservedAcceptance::Superseded);
        }
        if self.drain_source_events(receiver, scan_options)? {
            return Ok(ObservedAcceptance::Invalidated);
        }
        Ok(ObservedAcceptance::Accepted)
    }

    /// Install a new verified epoch after exact source and policy reconciliation.
    fn install_epoch(
        &self,
        process_nonce: [u8; 16],
        binding: &CapturedProjectBinding,
        generation: IndexGeneration,
        ingress_sequence: u64,
        contract_fingerprint: String,
        policy_witness: String,
    ) -> Result<VerifiedSourceEpoch, CliError> {
        let mut state = self
            .state
            .lock()
            .map_err(|_poisoned| lock_error(&self.binding.root, "source observation state"))?;
        state.next_epoch = state.next_epoch.saturating_add(1);
        let epoch = VerifiedSourceEpoch {
            stamp: VerifiedReadStamp {
                process_nonce,
                epoch: state.next_epoch,
                generation,
                project_instance_id: binding.project_instance_id,
            },
            ingress_sequence,
            contract_fingerprint,
            policy_witness,
        };
        state.verified = Some(epoch.clone());
        Ok(epoch)
    }

    /// Clone the current verified epoch without holding the state lock.
    fn current_epoch(&self) -> Result<Option<VerifiedSourceEpoch>, CliError> {
        self.state
            .lock()
            .map(|state| state.verified.clone())
            .map_err(|_poisoned| lock_error(&self.binding.root, "source observation state"))
    }
}

/// Classify source events without letting this binding's own `SQLite` files self-invalidate it.
fn observer_event_changes(
    binding: &SourceBinding,
    scan_options: &projectatlas_fs::ScanOptions,
    event: &Event,
) -> WatchChangeSet {
    let mut filtered = event.clone();
    let metadata_directory = binding.root.join(".projectatlas");
    filtered.paths.retain(|path| {
        let candidate = super::absolute_watch_path(&binding.root, path);
        !same_native_path(&candidate, &metadata_directory)
            && !is_database_runtime_path(&candidate, &binding.database)
    });
    if filtered.paths.is_empty() && !event.need_rescan() {
        return WatchChangeSet::default();
    }
    super::notify_event_changes(&binding.root, scan_options, &filtered)
}

/// Return whether a path is the database or one of its runtime sidecar files.
fn is_database_runtime_path(candidate: &Path, database: &Path) -> bool {
    if same_native_path(candidate, database) {
        return true;
    }
    ["-wal", "-shm", "-journal"].into_iter().any(|suffix| {
        let mut sidecar = database.as_os_str().to_os_string();
        sidecar.push(suffix);
        let sidecar = Path::new(&sidecar);
        same_native_path(candidate, sidecar) || {
            #[cfg(windows)]
            {
                missing_windows_sidecar_matches(candidate, sidecar, database, suffix)
            }
            #[cfg(not(windows))]
            {
                false
            }
        }
    })
}

/// Match a removed sidecar through its existing database leaf on ordinary
/// case-insensitive Windows directories. The base-file proof keeps a
/// case-sensitive directory's distinct spelling fail-closed.
#[cfg(windows)]
fn missing_windows_sidecar_matches(
    candidate: &Path,
    sidecar: &Path,
    database: &Path,
    suffix: &str,
) -> bool {
    if candidate.exists() || sidecar.exists() {
        return false;
    }
    let (Some(parent), Some(leaf)) = (candidate.parent(), candidate.file_name()) else {
        return false;
    };
    let suffix_units: Vec<u16> = OsStr::new(suffix).encode_wide().collect();
    let leaf_units: Vec<u16> = leaf.encode_wide().collect();
    if leaf_units.len() <= suffix_units.len()
        || !leaf_units[leaf_units.len() - suffix_units.len()..]
            .iter()
            .zip(&suffix_units)
            .all(|(actual, expected)| {
                let actual = *actual;
                let expected = *expected;
                let fold_ascii = |unit: u16| {
                    if (u16::from(b'A')..=u16::from(b'Z')).contains(&unit) {
                        unit + (u16::from(b'a') - u16::from(b'A'))
                    } else {
                        unit
                    }
                };
                actual == expected || fold_ascii(actual) == fold_ascii(expected)
            })
    {
        return false;
    }
    let base = parent.join(OsString::from_wide(
        &leaf_units[..leaf_units.len() - suffix_units.len()],
    ));
    // An unresolved candidate whose base spelling is already exact has no
    // native evidence that the missing suffix belongs to a case-insensitive
    // namespace; leave that event observable instead of guessing.
    base != database && same_native_path(&base, database)
}

/// Compare native paths without converting identity through display text.
fn same_native_path(left: &Path, right: &Path) -> bool {
    if left == right {
        return true;
    }
    let canonical = |path: &Path| -> Option<PathBuf> {
        if let Ok(path) = fs::canonicalize(path) {
            return Some(path);
        }
        let parent = path.parent()?;
        let identity = CanonicalProjectRoot::from_path(parent).ok()?;
        Some(identity.as_path().join(path.file_name()?))
    };
    match (canonical(left), canonical(right)) {
        (Some(left), Some(right)) => left == right,
        _ => false,
    }
}

/// Bounded process-local registry shared by every clone of one MCP server.
pub(crate) struct SourceObservationRegistry {
    /// Random identity distinguishing epochs created by this process.
    process_nonce: [u8; 16],
    /// Bounded observers keyed by exact source binding.
    entries: Mutex<HashMap<SourceBinding, Arc<SourceObservationEntry>>>,
    /// Acceptance invalidations forced by owning mutation-admission tests.
    #[cfg(test)]
    mutation_acceptance_invalidations: AtomicU64,
    /// Preparation invalidations forced by owning mutation-admission tests.
    #[cfg(test)]
    preparation_invalidations: AtomicU64,
}

impl fmt::Debug for SourceObservationRegistry {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        let entries = self.entries.lock().map_or(0, |entries| entries.len());
        formatter
            .debug_struct("SourceObservationRegistry")
            .field("entries", &entries)
            .field("capacity", &SOURCE_OBSERVATION_CAPACITY)
            .finish_non_exhaustive()
    }
}

impl Default for SourceObservationRegistry {
    fn default() -> Self {
        Self {
            process_nonce: process_nonce(),
            entries: Mutex::new(HashMap::new()),
            #[cfg(test)]
            mutation_acceptance_invalidations: AtomicU64::new(0),
            #[cfg(test)]
            preparation_invalidations: AtomicU64::new(0),
        }
    }
}

impl SourceObservationRegistry {
    /// Run one complete query inside a source-epoch and `SQLite`-generation boundary.
    pub(crate) fn with_verified_read<T, F>(
        &self,
        database: &Path,
        root: &Path,
        config: Option<&Path>,
        control: &IndexWorkControl,
        query: F,
    ) -> Result<VerifiedReadOutcome<T>, CliError>
    where
        F: FnMut(&AtlasStore, VerifiedReadStamp) -> Result<T, CliError>,
    {
        let binding = SourceBinding::new(database, root, config)?;
        let Some(entry) = self.entry(binding.clone())? else {
            return self.with_exact_fallback(&binding, control, query);
        };
        self.with_observed_read(&entry, control, query)
    }

    /// Admit a purpose mutation and retain its source witness through commit.
    pub(crate) fn admit_mutation(
        &self,
        database: &Path,
        root: &Path,
        config: Option<&Path>,
        control: &IndexWorkControl,
    ) -> Result<VerifiedMutationAdmission, CliError> {
        let binding = SourceBinding::new(database, root, config)?;
        let Some(entry) = self.entry(binding.clone())? else {
            return self.admit_exact_mutation(binding, control);
        };
        let mut work = VerifiedReadWork::default();
        for attempt in 0..VERIFIED_READ_ATTEMPTS {
            if let Err(error) = control.check(IndexWorkStage::Publication) {
                entry.invalidate();
                return Err(error.into());
            }
            let (store, epoch) =
                match self.prepare_observed_store(&entry, control, &mut work, attempt == 0) {
                    Ok(Some(prepared)) => prepared,
                    Ok(None) => {
                        entry.invalidate();
                        return self.admit_exact_mutation(binding, control);
                    }
                    Err(error) => {
                        entry.invalidate();
                        return Err(error);
                    }
                };
            #[cfg(test)]
            if self
                .mutation_acceptance_invalidations
                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
                    remaining.checked_sub(1)
                })
                .is_ok()
            {
                entry.continuity_lost.store(true, Ordering::Release);
            }
            match Self::accepts_observed_result(&entry, &epoch, control) {
                Ok(ObservedAcceptance::Accepted) => {
                    if let Err(error) = store.finish_index_read_snapshot() {
                        entry.invalidate_epoch(&epoch);
                        return Err(error.into());
                    }
                    return Ok(VerifiedMutationAdmission {
                        witness: MutationSourceWitness::Observed { entry, epoch },
                        control: control.clone(),
                    });
                }
                Ok(ObservedAcceptance::Superseded | ObservedAcceptance::Invalidated) => {}
                Err(error) => {
                    entry.invalidate_epoch(&epoch);
                    drop(store.finish_index_read_snapshot());
                    return Err(error);
                }
            }
            drop(store.finish_index_read_snapshot());
        }
        self.admit_exact_mutation(binding, control)
    }

    /// Admit a mutation through exact source when no native observer is available.
    fn admit_exact_mutation(
        &self,
        binding: SourceBinding,
        control: &IndexWorkControl,
    ) -> Result<VerifiedMutationAdmission, CliError> {
        for _attempt in 0..VERIFIED_READ_ATTEMPTS {
            control.check(IndexWorkStage::Publication)?;
            let before = exact_source_policy(&binding, control)?;
            let exact = open_exact_fresh_atlas_store_for_project_controlled(
                &binding.database,
                &binding.root,
                binding.config.as_deref(),
                control,
            )?;
            let after = match exact_source_policy(&binding, control) {
                Ok(after) => after,
                Err(error) => {
                    drop(exact.store.finish_index_read_snapshot());
                    return Err(error);
                }
            };
            if before != after {
                drop(exact.store.finish_index_read_snapshot());
                continue;
            }
            let stamp = match self.exact_stamp(&exact.store, &binding.root) {
                Ok(stamp) => stamp,
                Err(error) => {
                    drop(exact.store.finish_index_read_snapshot());
                    return Err(error);
                }
            };
            exact.store.finish_index_read_snapshot()?;
            return Ok(VerifiedMutationAdmission {
                witness: MutationSourceWitness::Exact {
                    binding,
                    stamp,
                    contract_fingerprint: after.0,
                    policy_witness: after.1,
                },
                control: control.clone(),
            });
        }
        Err(source_changed_during_derivation(&binding.root, "."))
    }

    /// Inject one deterministic observer event for owning integration tests.
    #[cfg(test)]
    pub(crate) fn inject_test_event(
        &self,
        database: &Path,
        root: &Path,
        config: Option<&Path>,
        event: Event,
    ) -> Result<(), CliError> {
        let binding = SourceBinding::new(database, root, config)?;
        let entry = self
            .entries
            .lock()
            .map_err(|_poisoned| lock_error(&binding.root, "source observation registry"))?
            .get(&binding)
            .cloned()
            .ok_or_else(|| {
                CliError::InvalidInput(format!(
                    "source observer test entry is missing for '{}'",
                    binding.root.display()
                ))
            })?;
        entry.publish_test_event(event)
    }

    /// Reuse or admit the bounded observer for an exact source binding.
    fn entry(
        &self,
        binding: SourceBinding,
    ) -> Result<Option<Arc<SourceObservationEntry>>, CliError> {
        let mut entries = self
            .entries
            .lock()
            .map_err(|_poisoned| lock_error(&binding.root, "source observation registry"))?;
        if let Some(entry) = entries.get(&binding) {
            return Ok(Some(Arc::clone(entry)));
        }
        if entries.len() >= SOURCE_OBSERVATION_CAPACITY {
            return Ok(None);
        }
        let entry = match SourceObservationEntry::start(binding.clone()) {
            Ok(entry) => Arc::new(entry),
            Err(_observer_unavailable) => return Ok(None),
        };
        entries.insert(binding, Arc::clone(&entry));
        Ok(Some(entry))
    }

    /// Retry a query until one observed epoch remains valid through acceptance.
    fn with_observed_read<T, F>(
        &self,
        entry: &SourceObservationEntry,
        control: &IndexWorkControl,
        mut query: F,
    ) -> Result<VerifiedReadOutcome<T>, CliError>
    where
        F: FnMut(&AtlasStore, VerifiedReadStamp) -> Result<T, CliError>,
    {
        let started = Instant::now();
        let mut work = VerifiedReadWork::default();
        for attempt in 0..VERIFIED_READ_ATTEMPTS {
            if let Err(error) = control.check(IndexWorkStage::Publication) {
                entry.invalidate();
                return Err(error.into());
            }
            let (store, epoch) = match self.prepare_observed_store(entry, control, &mut work, false)
            {
                Ok(Some(prepared)) => prepared,
                Ok(None) => {
                    entry.invalidate();
                    return Err(source_changed_during_derivation(&entry.binding.root, "."));
                }
                Err(error) => {
                    entry.invalidate();
                    return Err(error);
                }
            };
            let value = match query(&store, epoch.stamp.clone()) {
                Ok(value) => value,
                Err(error) => {
                    if matches!(error, CliError::IndexWork(_)) {
                        entry.invalidate_epoch(&epoch);
                    }
                    drop(store.finish_index_read_snapshot());
                    return Err(error);
                }
            };
            match Self::accepts_observed_result(entry, &epoch, control) {
                Ok(ObservedAcceptance::Accepted) => {
                    store.finish_index_read_snapshot()?;
                    work.elapsed = started.elapsed();
                    return Ok(VerifiedReadOutcome {
                        value,
                        stamp: epoch.stamp,
                        work,
                    });
                }
                Ok(ObservedAcceptance::Superseded | ObservedAcceptance::Invalidated) => {}
                Err(error) => {
                    entry.invalidate_epoch(&epoch);
                    drop(store.finish_index_read_snapshot());
                    return Err(error);
                }
            }
            drop(store.finish_index_read_snapshot());
            if attempt + 1 < VERIFIED_READ_ATTEMPTS {
                work.retries = work.retries.saturating_add(1);
            }
        }
        Err(source_changed_during_derivation(&entry.binding.root, "."))
    }

    /// Open a snapshot backed by a current epoch or establish a new exact epoch.
    fn prepare_observed_store(
        &self,
        entry: &SourceObservationEntry,
        control: &IndexWorkControl,
        work: &mut VerifiedReadWork,
        force_exact: bool,
    ) -> Result<Option<(AtlasStore, VerifiedSourceEpoch)>, CliError> {
        let _reconcile = entry
            .reconcile
            .lock()
            .map_err(|_poisoned| lock_error(&entry.binding.root, "source reconciliation"))?;
        // Mutation authority starts from exact saved source under the reconciliation lock.
        if force_exact {
            entry.invalidate();
        }
        let plan = ScanRuntimePlan::for_path_controlled(
            entry.binding.config.as_deref(),
            &entry.binding.root,
            None,
            control,
        )
        .map_err(|source| publication_input_error(&entry.binding.root, source))?;
        entry.changed_since_exact_verification(&plan.scan_options)?;
        let contract_fingerprint = plan.publication_contract_fingerprint();
        let policy_witness = source_policy_witness(&plan, control)?;
        if let Some(epoch) = entry.current_epoch()?
            && epoch.contract_fingerprint == contract_fingerprint
            && epoch.policy_witness == policy_witness
            && !entry.continuity_lost.load(Ordering::Acquire)
        {
            let store = open_atlas_store_read_only_for_project(
                &entry.binding.database,
                &entry.binding.root,
            )?;
            let publication = store.index_publication()?.filter(|publication| {
                publication.state == IndexPublicationState::Complete
                    && publication.contract_fingerprint.as_deref()
                        == Some(contract_fingerprint.as_str())
            });
            let captured = store.captured_project_binding()?;
            work.sqlite_read_statements = work.sqlite_read_statements.saturating_add(1);
            if publication
                .is_some_and(|publication| publication.generation == epoch.stamp.generation)
                && captured.project_instance_id == epoch.stamp.project_instance_id
            {
                return Ok(Some((store, epoch)));
            }
            entry.invalidate();
            drop(store.finish_index_read_snapshot());
        }

        for _attempt in 0..VERIFIED_READ_ATTEMPTS {
            entry.clear_before_exact_verification()?;
            let before_plan = ScanRuntimePlan::for_path_controlled(
                entry.binding.config.as_deref(),
                &entry.binding.root,
                None,
                control,
            )
            .map_err(|source| publication_input_error(&entry.binding.root, source))?;
            let before_contract = before_plan.publication_contract_fingerprint();
            let before_policy = source_policy_witness(&before_plan, control)?;
            let exact = open_exact_fresh_atlas_store_for_project_controlled(
                &entry.binding.database,
                &entry.binding.root,
                entry.binding.config.as_deref(),
                control,
            )?;
            work.add_exact(exact.work);
            let after_plan = ScanRuntimePlan::for_path_controlled(
                entry.binding.config.as_deref(),
                &entry.binding.root,
                None,
                control,
            )
            .map_err(|source| publication_input_error(&entry.binding.root, source))?;
            let after_contract = after_plan.publication_contract_fingerprint();
            let after_policy = source_policy_witness(&after_plan, control)?;
            #[cfg(test)]
            if self
                .preparation_invalidations
                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
                    remaining.checked_sub(1)
                })
                .is_ok()
            {
                entry.invalidate_after_proven_source_mismatch()?;
            }
            if before_contract != after_contract || before_policy != after_policy {
                drop(exact.store.finish_index_read_snapshot());
                continue;
            }
            let publication = exact
                .store
                .index_publication()?
                .ok_or_else(|| source_changed_during_derivation(&entry.binding.root, "."))?;
            let captured = exact.store.captured_project_binding()?;
            work.sqlite_read_statements = work.sqlite_read_statements.saturating_add(1);
            let receiver = entry.receiver.lock().map_err(|_poisoned| {
                lock_error(&entry.binding.root, "source observation receiver")
            })?;
            if entry.drain_source_events(&receiver, &after_plan.scan_options)? {
                drop(receiver);
                drop(exact.store.finish_index_read_snapshot());
                continue;
            }
            let ingress_sequence = entry.ingress_sequence.load(Ordering::Acquire);
            let epoch = entry.install_epoch(
                self.process_nonce,
                &captured,
                publication.generation,
                ingress_sequence,
                after_contract,
                after_policy,
            )?;
            return Ok(Some((exact.store, epoch)));
        }
        Ok(None)
    }

    /// Confirm that no source, policy, or observer event invalidated a query result.
    fn accepts_observed_result(
        entry: &SourceObservationEntry,
        epoch: &VerifiedSourceEpoch,
        control: &IndexWorkControl,
    ) -> Result<ObservedAcceptance, CliError> {
        for _attempt in 0..VERIFIED_READ_ATTEMPTS {
            control.check(IndexWorkStage::Publication)?;
            let Some(sampled) = entry.current_epoch()? else {
                return Ok(ObservedAcceptance::Invalidated);
            };
            let plan = ScanRuntimePlan::for_path_controlled(
                entry.binding.config.as_deref(),
                &entry.binding.root,
                None,
                control,
            )
            .map_err(|source| publication_input_error(&entry.binding.root, source))?;
            let contract = plan.publication_contract_fingerprint();
            let policy = source_policy_witness(&plan, control)?;
            {
                let receiver = entry.receiver.lock().map_err(|_poisoned| {
                    lock_error(&entry.binding.root, "source observation receiver")
                })?;
                match entry.drain_epoch_events(&receiver, &sampled, &plan.scan_options)? {
                    ObservedAcceptance::Accepted => {}
                    ObservedAcceptance::Superseded => continue,
                    ObservedAcceptance::Invalidated => return Ok(ObservedAcceptance::Invalidated),
                }
            }
            #[cfg(test)]
            if let Some(event) = entry
                .acceptance_event
                .lock()
                .map_err(|_poisoned| lock_error(&entry.binding.root, "acceptance test event"))?
                .take()
            {
                entry.publish_test_event(event)?;
            }
            let receiver = entry.receiver.lock().map_err(|_poisoned| {
                lock_error(&entry.binding.root, "source observation receiver")
            })?;
            match entry.drain_epoch_events(&receiver, &sampled, &plan.scan_options)? {
                ObservedAcceptance::Accepted => {}
                ObservedAcceptance::Superseded => continue,
                ObservedAcceptance::Invalidated => return Ok(ObservedAcceptance::Invalidated),
            }
            let mut state = entry
                .state
                .lock()
                .map_err(|_poisoned| lock_error(&entry.binding.root, "source observation state"))?;
            let Some(current) = state.verified.as_ref() else {
                return Ok(ObservedAcceptance::Invalidated);
            };
            // A reconciliation during policy sampling requires fresh policy evidence.
            if current.stamp != sampled.stamp {
                continue;
            }
            if current.contract_fingerprint != contract || current.policy_witness != policy {
                state.verified = None;
                return Ok(ObservedAcceptance::Invalidated);
            }
            if current.stamp != epoch.stamp {
                return Ok(ObservedAcceptance::Superseded);
            }
            if current.ingress_sequence == entry.ingress_sequence.load(Ordering::Acquire) {
                return Ok(ObservedAcceptance::Accepted);
            }
        }
        Ok(ObservedAcceptance::Invalidated)
    }

    /// Exact-per-call compatibility path when a native observer cannot be admitted.
    fn with_exact_fallback<T, F>(
        &self,
        binding: &SourceBinding,
        control: &IndexWorkControl,
        mut query: F,
    ) -> Result<VerifiedReadOutcome<T>, CliError>
    where
        F: FnMut(&AtlasStore, VerifiedReadStamp) -> Result<T, CliError>,
    {
        let started = Instant::now();
        let mut work = VerifiedReadWork::default();
        for attempt in 0..VERIFIED_READ_ATTEMPTS {
            let exact = open_exact_fresh_atlas_store_for_project_controlled(
                &binding.database,
                &binding.root,
                binding.config.as_deref(),
                control,
            )?;
            work.add_exact(exact.work);
            let stamp = self.exact_stamp(&exact.store, &binding.root)?;
            let value = query(&exact.store, stamp.clone())?;
            exact.store.finish_index_read_snapshot()?;

            let post = open_exact_fresh_atlas_store_for_project_controlled(
                &binding.database,
                &binding.root,
                binding.config.as_deref(),
                control,
            )?;
            work.add_exact(post.work);
            let post_publication = post.store.index_publication()?;
            let post_binding = post.store.captured_project_binding()?;
            post.store.finish_index_read_snapshot()?;
            if post_publication.is_some_and(|candidate| {
                candidate.generation == stamp.generation
                    && post_binding.project_instance_id == stamp.project_instance_id
            }) {
                work.elapsed = started.elapsed();
                return Ok(VerifiedReadOutcome { value, stamp, work });
            }
            if attempt + 1 < VERIFIED_READ_ATTEMPTS {
                work.retries = work.retries.saturating_add(1);
            }
        }
        Err(source_changed_during_derivation(&binding.root, "."))
    }

    /// Capture durable identity from one exact current read snapshot.
    fn exact_stamp(&self, store: &AtlasStore, root: &Path) -> Result<VerifiedReadStamp, CliError> {
        let publication = store
            .index_publication()?
            .ok_or_else(|| source_changed_during_derivation(root, "."))?;
        let captured = store.captured_project_binding()?;
        Ok(VerifiedReadStamp {
            process_nonce: self.process_nonce,
            epoch: 0,
            generation: publication.generation,
            project_instance_id: captured.project_instance_id,
        })
    }
}

/// Convert native watcher callbacks into bounded ingress and continuity state.
fn watcher_callback(
    sender: SyncSender<Event>,
    receiver: Arc<Mutex<Receiver<Event>>>,
    ingress_sequence: Arc<AtomicU64>,
    continuity_lost: Arc<AtomicBool>,
) -> impl FnMut(notify::Result<Event>) + Send + 'static {
    move |result| match result {
        Ok(event) if matches!(event.kind, EventKind::Access(_)) => {}
        result => {
            let _receiver = match receiver.lock() {
                Ok(receiver) => receiver,
                Err(_poisoned) => {
                    continuity_lost.store(true, Ordering::Release);
                    ingress_sequence.fetch_add(1, Ordering::AcqRel);
                    return;
                }
            };
            match result {
                Ok(event) => {
                    let needs_rescan = event.need_rescan();
                    match sender.try_send(event) {
                        Ok(()) if !needs_rescan => {}
                        Ok(()) | Err(TrySendError::Full(_) | TrySendError::Disconnected(_)) => {
                            continuity_lost.store(true, Ordering::Release);
                        }
                    }
                }
                Err(_source) => {
                    continuity_lost.store(true, Ordering::Release);
                }
            }
            ingress_sequence.fetch_add(1, Ordering::AcqRel);
        }
    }
}

/// Hash the bounded external inputs that may change the scanner's source selection.
fn source_policy_witness(
    plan: &ScanRuntimePlan,
    control: &IndexWorkControl,
) -> Result<String, CliError> {
    let mut hasher = Hasher::new();
    hash_field(
        &mut hasher,
        "contract",
        plan.publication_contract_fingerprint().as_bytes(),
    );
    let root_identity = CanonicalProjectRoot::from_path(&plan.root)
        .map_err(|error| CliError::InvalidInput(error.to_string()))?;
    let root_identity_bytes = root_identity
        .encode()
        .map_err(|error| CliError::InvalidInput(error.to_string()))?;
    hash_field(&mut hasher, "root", &root_identity_bytes);
    for path in source_policy_paths(plan, control)? {
        control.check(IndexWorkStage::Publication)?;
        hash_field(&mut hasher, "path", path.as_os_str().as_encoded_bytes());
        match path.metadata() {
            Ok(metadata) if metadata.is_file() => {
                let file = File::open(&path).map_err(|source| CliError::Io {
                    path: path.clone(),
                    source,
                })?;
                hash_policy_file(&mut hasher, path.as_path(), file, control)?;
            }
            Ok(metadata) if metadata.is_dir() => {
                hash_field(&mut hasher, "state", b"directory");
            }
            Ok(_metadata) => {
                hash_field(&mut hasher, "state", b"other");
            }
            Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
                hash_field(&mut hasher, "state", b"absent");
            }
            Err(source) => {
                return Err(CliError::Io { path, source });
            }
        }
    }
    Ok(hasher.finalize().to_hex().to_string())
}

/// Capture the scanner contract and dynamic policy for one exact source binding.
fn exact_source_policy(
    binding: &SourceBinding,
    control: &IndexWorkControl,
) -> Result<(String, String), CliError> {
    let plan = ScanRuntimePlan::for_path_controlled(
        binding.config.as_deref(),
        &binding.root,
        None,
        control,
    )
    .map_err(|source| publication_input_error(&binding.root, source))?;
    let contract_fingerprint = plan.publication_contract_fingerprint();
    let policy_witness = source_policy_witness(&plan, control)?;
    Ok((contract_fingerprint, policy_witness))
}

/// Collect every scanner policy input whose state contributes to source selection.
fn source_policy_paths(
    plan: &ScanRuntimePlan,
    control: &IndexWorkControl,
) -> Result<Vec<PathBuf>, CliError> {
    let mut paths = projectatlas_fs::source_selection_policy_paths_controlled(&plan.root, control)
        .map_err(|source| source_inspection_error(&plan.root, source))?;
    if let Some(config) = plan.selected_config_path.as_ref() {
        paths.push(config.clone());
    } else {
        paths.push(plan.root.join(".projectatlas").join("config.toml"));
        paths.push(plan.root.join("projectatlas.toml"));
    }
    paths.sort();
    paths.dedup();
    Ok(paths)
}

/// Hash one bounded policy file with cancellation checks between chunks.
fn hash_policy_file(
    hasher: &mut Hasher,
    path: &Path,
    file: File,
    control: &IndexWorkControl,
) -> Result<(), CliError> {
    let mut reader = file.take(MAX_POLICY_INPUT_BYTES.saturating_add(1));
    let mut buffer = [0_u8; 8_192];
    let mut observed = 0_u64;
    loop {
        control.check(IndexWorkStage::Publication)?;
        let read = read_policy_chunk(&mut reader, &mut buffer, path)?;
        if read == 0 {
            break;
        }
        observed = observed.saturating_add(u64::try_from(read).unwrap_or(u64::MAX));
        if observed > MAX_POLICY_INPUT_BYTES {
            return Err(CliError::InvalidInput(format!(
                "source-selection policy input '{}' exceeds the {} byte limit",
                path.display(),
                MAX_POLICY_INPUT_BYTES
            )));
        }
        hasher.update(&buffer[..read]);
    }
    hash_field(hasher, "state", b"present");
    Ok(())
}

/// Read one policy chunk while retaining the file path in any I/O error.
fn read_policy_chunk(
    reader: &mut Take<File>,
    buffer: &mut [u8],
    path: &Path,
) -> Result<usize, CliError> {
    reader.read(buffer).map_err(|source| CliError::Io {
        path: path.to_path_buf(),
        source,
    })
}

/// Append one domain-separated field to a policy witness.
fn hash_field(hasher: &mut Hasher, name: &str, value: &[u8]) {
    hasher.update(name.as_bytes());
    hasher.update(&[0]);
    hasher.update(value);
    hasher.update(&[0xff]);
}

/// Resolve a possibly relative path against the canonical source root.
fn absolute_path_from(root: &Path, path: &Path) -> PathBuf {
    if path.is_absolute() {
        path.to_path_buf()
    } else {
        root.join(path)
    }
}

/// Generate one process identity, with a deterministic local entropy fallback.
fn process_nonce() -> [u8; 16] {
    let mut nonce = [0_u8; 16];
    if getrandom::fill(&mut nonce).is_ok() {
        return nonce;
    }
    let mut hasher = Hasher::new();
    hasher.update(&std::process::id().to_le_bytes());
    let time = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| duration.as_nanos());
    hasher.update(&time.to_le_bytes());
    nonce.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
    nonce
}

/// Convert watcher startup failure into a root-scoped observer diagnostic.
fn observer_error(root: &Path, source: &notify::Error) -> CliError {
    CliError::InvalidInput(format!(
        "source observation is unavailable for '{}': {source}",
        root.display()
    ))
}

/// Build a root-scoped diagnostic for poisoned observer state.
fn lock_error(root: &Path, owner: &str) -> CliError {
    CliError::InvalidInput(format!(
        "{owner} lock is unavailable for project root '{}'",
        root.display()
    ))
}

#[cfg(test)]
mod tests {
    use super::*;
    use notify::event::{ModifyKind, RenameMode};
    use notify::{EventKind, event::AccessKind};
    use projectatlas_core::{IndexCancellation, PurposeSource};
    use std::error::Error;
    #[cfg(windows)]
    use std::process::Command;

    /// Create and publish a minimal indexed repository for observer tests.
    fn indexed_project(root: &Path) -> Result<(PathBuf, PathBuf), Box<dyn Error>> {
        fs::create_dir_all(root.join(".projectatlas"))?;
        let source = root.join("source.rs");
        fs::write(&source, "fn original() {}\n")?;
        let database = root.join(".projectatlas").join("projectatlas.db");
        let mut store = super::super::open_atlas_store_for_project(&database, root)?;
        let plan = ScanRuntimePlan::for_path(None, root, None)?;
        super::super::run_scan_pipeline(
            &mut store,
            &plan,
            &super::super::SymbolBuildOptions::new(
                super::super::MAX_SYMBOL_FILE_BYTES,
                Some(1),
                None,
            ),
        )?;
        drop(store);
        Ok((database, source))
    }

    /// Build a bounded non-cancelled control for observer tests.
    fn test_control() -> IndexWorkControl {
        IndexWorkControl::new(IndexCancellation::new(), Some(Duration::from_secs(30)))
    }

    /// Return a fallible test failure without panicking inside result-returning tests.
    fn require(condition: bool, message: &str) -> Result<(), Box<dyn Error>> {
        if condition {
            Ok(())
        } else {
            Err(std::io::Error::other(message).into())
        }
    }

    #[test]
    fn native_observer_teardown_releases_root_and_external_config() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let root = temp.path().join("repo");
        let external = temp.path().join("settings");
        fs::create_dir_all(&root)?;
        fs::create_dir_all(&external)?;
        let config = external.join("config.toml");
        fs::write(&config, "")?;
        let database = root.join("atlas.db");
        let entry =
            SourceObservationEntry::start(SourceBinding::new(&database, &root, Some(&config))?)?;
        fs::write(root.join("source.rs"), "fn observed() {}")?;
        drop(entry);
        require(!database.exists(), "observer cleanup created an index")?;
        fs::remove_dir_all(&root)?;
        fs::remove_dir_all(&external)?;
        require(
            !root.exists() && !external.exists(),
            "watch directories remain",
        )
    }

    #[test]
    fn native_observer_failed_registration_releases_admitted_root() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let root = temp.path().join("repo");
        fs::create_dir_all(&root)?;
        let database = root.join("atlas.db");
        let config = temp.path().join("missing-parent").join("config.toml");
        let binding = SourceBinding::new(&database, &root, Some(&config))?;
        let result = SourceObservationEntry::start(binding.clone());
        require(result.is_err(), "missing configuration parent was accepted")?;
        require(
            !database.exists(),
            "failed observer startup created an index",
        )?;
        fs::remove_dir_all(&root)?;
        require(!root.exists(), "failed startup retained its root")?;
        require(
            SourceObservationEntry::start(binding).is_err(),
            "missing root was accepted",
        )
    }

    #[test]
    fn watcher_callback_is_bounded_and_marks_overflow_and_rescan() {
        let (sender, receiver) = sync_channel(1);
        let receiver = Arc::new(Mutex::new(receiver));
        let sequence = Arc::new(AtomicU64::new(0));
        let lost = Arc::new(AtomicBool::new(false));
        let mut callback = watcher_callback(
            sender,
            Arc::clone(&receiver),
            Arc::clone(&sequence),
            Arc::clone(&lost),
        );
        callback(Ok(Event::new(EventKind::Modify(ModifyKind::Name(
            RenameMode::Any,
        )))));
        callback(Ok(Event::new(EventKind::Modify(ModifyKind::Any))));

        assert_eq!(sequence.load(Ordering::Acquire), 2);
        assert!(lost.load(Ordering::Acquire));
        assert!(
            receiver
                .lock()
                .is_ok_and(|receiver| receiver.try_recv().is_ok())
        );
    }

    #[test]
    fn watcher_callback_ignores_access_events_without_advancing_epoch() {
        let (sender, receiver) = sync_channel(1);
        let receiver = Arc::new(Mutex::new(receiver));
        let sequence = Arc::new(AtomicU64::new(0));
        let lost = Arc::new(AtomicBool::new(false));
        let mut callback =
            watcher_callback(sender, receiver, Arc::clone(&sequence), Arc::clone(&lost));
        callback(Ok(Event::new(EventKind::Access(AccessKind::Any))));

        assert_eq!(sequence.load(Ordering::Acquire), 0);
        assert!(!lost.load(Ordering::Acquire));
    }

    #[test]
    fn verified_epoch_avoids_repeat_tree_and_node_table_work() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();

        let first = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, stamp| Ok((store.overview()?, stamp)),
        )?;
        require(
            (1..=u64::try_from(VERIFIED_READ_ATTEMPTS)?).contains(&first.work.exact_verifications),
            "initial read did not perform a bounded exact verification",
        )?;
        require(
            first.work.filesystem_entries > 0,
            "initial read did not inspect filesystem entries",
        )?;
        require(
            first.work.decoded_nodes > 0,
            "initial read did not decode indexed nodes",
        )?;

        let second = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, stamp| Ok((store.overview()?, stamp)),
        )?;
        require(
            second.work.exact_verifications == 0,
            "warm read unexpectedly repeated exact verification",
        )?;
        require(
            second.work.filesystem_entries == 0,
            "warm read unexpectedly inspected filesystem entries",
        )?;
        require(
            second.work.filesystem_bytes == 0,
            "warm read unexpectedly hashed source bytes",
        )?;
        require(
            second.work.decoded_nodes == 0,
            "warm read unexpectedly decoded the full node table",
        )?;
        require(
            second.work.sqlite_read_statements == 1,
            "warm freshness check used an unexpected SQLite statement count",
        )?;
        require(
            first.stamp == second.stamp,
            "warm read did not reuse the verified epoch",
        )?;
        Ok(())
    }

    #[test]
    fn ignore_policy_changes_invalidate_the_epoch_and_refresh_source_truth()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let first = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.load_node_by_path("source.rs")?.is_some()),
        )?;
        require(first.value, "initial indexed source was missing")?;

        fs::write(temp.path().join(".ignore"), "source.rs\n")?;
        let ignored = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.load_node_by_path("source.rs")?.is_some()),
        );

        require(
            matches!(ignored, Err(CliError::RefreshRequired(_))),
            "ignore policy change did not require refresh",
        )?;
        let binding = SourceBinding::new(&database, temp.path(), None)?;
        let entry = registry
            .entries
            .lock()
            .map_err(|_poisoned| std::io::Error::other("registry lock poisoned"))?
            .get(&binding)
            .cloned()
            .ok_or_else(|| std::io::Error::other("observer entry missing"))?;
        require(
            entry.current_epoch()?.is_none(),
            "ignore policy change left a verified epoch installed",
        )?;
        Ok(())
    }

    #[test]
    fn git_exclude_changes_are_witnessed_with_a_git_directory() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let git = temp.path().join(".git");
        let git_info = git.join("info");
        fs::create_dir_all(&git_info)?;
        fs::create_dir_all(git.join("objects"))?;
        fs::create_dir_all(git.join("refs"))?;
        fs::write(git.join("HEAD"), "ref: refs/heads/main\n")?;
        fs::write(git.join("config"), "[core]\n")?;
        fs::write(git_info.join("exclude"), "")?;
        let registry = SourceObservationRegistry::default();
        let first = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.load_node_by_path("source.rs")?.is_some()),
        )?;
        require(first.value, "initial indexed source was missing")?;

        fs::write(git_info.join("exclude"), "source.rs\n")?;
        let ignored = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.load_node_by_path("source.rs")?.is_some()),
        )?;

        require(
            !ignored.value,
            "Git exclude change left the excluded source indexed",
        )?;
        require(
            ignored.work.exact_verifications >= 1,
            "Git exclude change reused the stale source epoch",
        )?;
        require(
            ignored.stamp.epoch > first.stamp.epoch,
            "Git exclude change did not advance the verified source epoch",
        )?;
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn symlinked_git_exclude_target_changes_invalidate_the_epoch() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let policy = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let git = temp.path().join(".git");
        let git_info = git.join("info");
        let exclude_target = policy.path().join("exclude");
        fs::create_dir_all(&git_info)?;
        fs::create_dir_all(git.join("objects"))?;
        fs::create_dir_all(git.join("refs"))?;
        fs::write(git.join("HEAD"), "ref: refs/heads/main\n")?;
        fs::write(git.join("config"), "[core]\n")?;
        fs::write(&exclude_target, "")?;
        std::os::unix::fs::symlink(&exclude_target, git_info.join("exclude"))?;
        let registry = SourceObservationRegistry::default();
        let first = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.load_node_by_path("source.rs")?.is_some()),
        )?;
        require(first.value, "initial indexed source was missing")?;

        fs::write(&exclude_target, "source.rs\n")?;
        let ignored = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.load_node_by_path("source.rs")?.is_some()),
        )?;

        require(
            !ignored.value,
            "symlinked Git exclude target change left the source indexed",
        )?;
        require(
            ignored.work.exact_verifications >= 1,
            "symlinked Git exclude target change reused the stale source epoch",
        )?;
        require(
            ignored.stamp.epoch > first.stamp.epoch,
            "symlinked Git exclude target change did not advance the verified source epoch",
        )?;
        Ok(())
    }

    #[test]
    fn mid_query_edit_discards_provisional_result_and_reconciles() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let _initial = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.overview()?),
        )?;
        let binding = SourceBinding::new(&database, temp.path(), None)?;
        let entry = registry
            .entries
            .lock()
            .map_err(|_poisoned| std::io::Error::other("registry lock poisoned"))?
            .get(&binding)
            .cloned()
            .ok_or_else(|| std::io::Error::other("observer entry missing"))?;
        let mut calls = 0_u64;
        let revised = "fn revised() {}\n";

        let outcome = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| {
                calls = calls.saturating_add(1);
                let hash = store
                    .load_node_by_path("source.rs")?
                    .and_then(|node| node.node.content_hash)
                    .ok_or_else(|| CliError::InvalidInput("source hash missing".to_string()))?;
                if calls == 1 {
                    fs::write(&source, revised).map_err(|source_error| CliError::Io {
                        path: source.clone(),
                        source: source_error,
                    })?;
                    entry.publish_test_event(
                        Event::new(EventKind::Modify(ModifyKind::Any)).add_path(source.clone()),
                    )?;
                }
                Ok(hash)
            },
        )?;

        require(calls >= 2, "mid-query edit did not retry the query")?;
        require(
            outcome.work.retries >= 1,
            "mid-query edit was not reported as a retry",
        )?;
        require(
            outcome.work.exact_verifications >= 1,
            "mid-query edit did not trigger exact verification",
        )?;
        require(
            outcome.value == blake3::hash(revised.as_bytes()).to_hex().to_string(),
            "accepted result did not reflect the revised source",
        )?;
        Ok(())
    }

    #[test]
    fn observed_acceptance_does_not_wait_for_exact_reconciliation() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let control = test_control();
        let entry = registry
            .entry(SourceBinding::new(&database, temp.path(), None)?)?
            .ok_or_else(|| std::io::Error::other("source observer unavailable"))?;
        let (store, epoch) = registry
            .prepare_observed_store(&entry, &control, &mut VerifiedReadWork::default(), false)?
            .ok_or_else(|| std::io::Error::other("read epoch unavailable"))?;
        let reconcile = entry
            .reconcile
            .lock()
            .map_err(|_poisoned| std::io::Error::other("source reconciliation lock poisoned"))?;
        let (sender, receiver) = std::sync::mpsc::channel();
        let accepted = std::thread::scope(|scope| {
            scope.spawn(|| {
                let accepted = matches!(
                    SourceObservationRegistry::accepts_observed_result(&entry, &epoch, &control),
                    Ok(ObservedAcceptance::Accepted)
                );
                let _sent = sender.send(accepted);
            });
            let result = receiver.recv_timeout(Duration::from_secs(5));
            // Release even on failure so the regression cannot strand the test worker.
            drop(reconcile);
            result
        });
        require(
            accepted == Ok(true),
            "observed acceptance waited for exact reconciliation or rejected valid evidence",
        )?;
        store.finish_index_read_snapshot()?;
        Ok(())
    }

    #[test]
    fn older_read_acceptance_preserves_newer_mutation_witness() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let before = fs::read(&source)?;
        let registry = SourceObservationRegistry::default();
        let control = test_control();
        let mut admission = None;
        registry.with_verified_read(&database, temp.path(), None, &control, |store, _stamp| {
            if admission.is_none() {
                admission =
                    Some(registry.admit_mutation(&database, temp.path(), None, &control)?);
            }
            Ok(store.overview()?)
        })?;
        require(
            fs::read(&source)? == before,
            "source changed during read admission",
        )?;
        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let transaction = store.begin_purpose_mutation()?;
        store.set_purpose(
            "source.rs",
            "Accepted concurrent purpose",
            PurposeSource::Agent,
        )?;
        admission
            .ok_or_else(|| std::io::Error::other("mutation admission missing"))?
            .verify()?;
        transaction.commit()?;
        require(
            store.load_node_by_path("source.rs")?.is_some_and(|node| {
                node.purpose.purpose.as_deref() == Some("Accepted concurrent purpose")
            }),
            "valid concurrent mutation did not commit its purpose",
        )?;
        Ok(())
    }

    #[test]
    fn superseded_mutation_cleanup_preserves_successor_witness() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let control = test_control();
        let previous = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let current = registry.admit_mutation(&database, temp.path(), None, &control)?;
        require(
            matches!(previous.verify(), Err(CliError::RefreshRequired(_))),
            "superseded mutation witness was unexpectedly accepted",
        )?;
        current.verify()?;
        Ok(())
    }

    #[test]
    fn exact_source_mismatch_invalidates_successor_without_observer_delivery()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let binding = SourceBinding::new(&database, temp.path(), None)?;
        let mut observer = SourceObservationEntry::start(binding.clone())?;
        observer.watcher.unwatch(&binding.root)?;
        observer.watcher.configure(notify::Config::default())?;
        let entry = Arc::new(observer);
        registry
            .entries
            .lock()
            .map_err(|_poisoned| std::io::Error::other("observer registry lock poisoned"))?
            .insert(binding, Arc::clone(&entry));
        let control = test_control();
        let previous = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let MutationSourceWitness::Observed { epoch, .. } = &previous.witness else {
            return Err(std::io::Error::other("observed mutation missing").into());
        };
        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let before_revision = store.authored_purpose_revision()?;
        let transaction = store.begin_purpose_mutation()?;
        let mut successor = None;
        let result = VerifiedMutationAdmission::verify_observed(&entry, epoch, &control, || {
            successor = Some(registry.admit_mutation(&database, temp.path(), None, &control)?);
            store.set_purpose("source.rs", "Uncommitted purpose", PurposeSource::Agent)?;
            fs::write(&source, "fn changed_after_successor() {}\n").map_err(|error| {
                CliError::Io {
                    path: source.clone(),
                    source: error,
                }
            })?;
            verify_saved_source_matches_index_controlled(&database, temp.path(), None, &control)
        });
        drop(transaction);
        require(
            matches!(result, Err(CliError::RefreshRequired(_))),
            "exact source mismatch did not reject the previous mutation",
        )?;
        require(
            store.authored_purpose_revision()? == before_revision,
            "rejected mutation changed the authored-purpose revision",
        )?;
        require(
            entry.current_epoch()?.is_none(),
            "proven saved-source mismatch left successor evidence reusable",
        )?;
        let successor =
            successor.ok_or_else(|| std::io::Error::other("successor admission missing"))?;
        require(
            matches!(successor.verify(), Err(CliError::RefreshRequired(_))),
            "successor survived a proven saved-source mismatch",
        )?;
        Ok(())
    }

    #[test]
    fn cancelled_exact_verification_preserves_successor_witness() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let cancellation = IndexCancellation::new();
        let control = IndexWorkControl::new(cancellation.clone(), Some(Duration::from_secs(30)));
        let previous = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let MutationSourceWitness::Observed { entry, epoch } = &previous.witness else {
            return Err(std::io::Error::other("observed mutation missing").into());
        };
        let mut successor = None;
        let result = VerifiedMutationAdmission::verify_observed(entry, epoch, &control, || {
            successor =
                Some(registry.admit_mutation(&database, temp.path(), None, &test_control())?);
            cancellation.cancel();
            verify_saved_source_matches_index_controlled(&database, temp.path(), None, &control)
        });
        require(
            matches!(result, Err(CliError::IndexWork(_))),
            "exact verification did not return cancellation",
        )?;
        successor
            .ok_or_else(|| std::io::Error::other("successor admission missing"))?
            .verify()?;
        Ok(())
    }

    #[test]
    fn stale_read_source_event_invalidates_newer_mutation_and_rolls_back()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let control = test_control();
        let entry = registry
            .entry(SourceBinding::new(&database, temp.path(), None)?)?
            .ok_or_else(|| std::io::Error::other("source observer unavailable"))?;
        let (read_store, old_epoch) = registry
            .prepare_observed_store(&entry, &control, &mut VerifiedReadWork::default(), false)?
            .ok_or_else(|| std::io::Error::other("read epoch unavailable"))?;
        let admission = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let before_revision = store.authored_purpose_revision()?;
        let before_purpose = store
            .load_node_by_path("source.rs")?
            .ok_or_else(|| std::io::Error::other("indexed source missing"))?
            .purpose;
        let transaction = store.begin_purpose_mutation()?;
        store.set_purpose(
            "source.rs",
            "Rejected concurrent purpose",
            PurposeSource::Agent,
        )?;
        let event = Event::new(EventKind::Modify(ModifyKind::Any)).add_path(source);
        let previous_options = projectatlas_fs::ScanOptions {
            exclude_path_prefixes: vec!["source.rs".to_owned()],
            ..projectatlas_fs::ScanOptions::default()
        };
        let ignored = observer_event_changes(&entry.binding, &previous_options, &event);
        require(
            ignored.paths.is_empty() && !ignored.requires_full_scan,
            "previous policy did not exclude the event in the regression fixture",
        )?;
        entry.publish_test_event(event)?;
        {
            let receiver = entry.receiver.lock().map_err(|_poisoned| {
                std::io::Error::other("source observation receiver lock poisoned")
            })?;
            require(
                entry.drain_epoch_events(&receiver, &old_epoch, &previous_options)?
                    == ObservedAcceptance::Superseded,
                "older policy consumed an event belonging to a successor epoch",
            )?;
        }
        require(
            SourceObservationRegistry::accepts_observed_result(&entry, &old_epoch, &control)?
                == ObservedAcceptance::Invalidated,
            "stale reader treated a source event as mere supersession",
        )?;
        require(
            entry.current_epoch()?.is_none(),
            "source event left newer evidence installed",
        )?;
        let plan = ScanRuntimePlan::for_path_controlled(None, temp.path(), None, &control)?;
        require(
            entry.changed_since_exact_verification(&plan.scan_options)?,
            "consumed source event was lost to concurrent exact verification",
        )?;
        read_store.finish_index_read_snapshot()?;
        let verification = admission.verify();
        transaction.rollback()?;
        require(
            matches!(verification, Err(CliError::RefreshRequired(_))),
            "consumed source event did not reject the mutation witness",
        )?;
        require(
            store.authored_purpose_revision()? == before_revision
                && store
                    .load_node_by_path("source.rs")?
                    .is_some_and(|node| node.purpose == before_purpose),
            "rejected mutation changed authored purpose state",
        )?;
        Ok(())
    }

    #[test]
    fn mutation_admission_reconciles_saved_source_without_waiting_for_observer_delivery()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let control = test_control();
        let _warm = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &control,
            |store, _stamp| Ok(store.overview()?),
        )?;
        let revised = "fn revised_before_admission() {}\n";
        fs::write(&source, revised)?;

        let admission = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let content_hash = store
            .load_node_by_path("source.rs")?
            .and_then(|node| node.node.content_hash)
            .ok_or_else(|| std::io::Error::other("reconciled source hash missing"))?;
        require(
            content_hash == blake3::hash(revised.as_bytes()).to_hex().to_string(),
            "mutation admission reused the warm source epoch",
        )?;
        admission.verify()?;
        Ok(())
    }

    #[test]
    fn mutation_admission_retries_transient_invalidation_and_falls_back_to_exact_source()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let control = test_control();
        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let before_source = fs::read(&source)?;
        let before_revision = store.authored_purpose_revision()?;
        let before_purpose = store
            .load_node_by_path("source.rs")?
            .ok_or_else(|| std::io::Error::other("indexed source missing"))?
            .purpose;

        registry
            .mutation_acceptance_invalidations
            .store(1, Ordering::Release);
        let admission = registry.admit_mutation(&database, temp.path(), None, &control)?;
        require(
            registry
                .mutation_acceptance_invalidations
                .load(Ordering::Acquire)
                == 0,
            "transient mutation invalidation was not consumed",
        )?;
        admission.verify()?;

        registry
            .mutation_acceptance_invalidations
            .store(u64::try_from(VERIFIED_READ_ATTEMPTS)?, Ordering::Release);
        let exact = registry.admit_mutation(&database, temp.path(), None, &control)?;
        require(
            matches!(&exact.witness, MutationSourceWitness::Exact { .. }),
            "persistent mutation invalidation did not fall back to exact source",
        )?;
        require(
            registry
                .mutation_acceptance_invalidations
                .load(Ordering::Acquire)
                == 0,
            "mutation invalidation attempts were not bounded",
        )?;
        exact.verify()?;
        require(
            fs::read(&source)? == before_source,
            "mutation admission changed saved source",
        )?;
        require(
            store.authored_purpose_revision()? == before_revision,
            "mutation admission changed authored-purpose revision",
        )?;
        require(
            store
                .load_node_by_path("source.rs")?
                .is_some_and(|node| node.purpose == before_purpose),
            "mutation admission changed the purpose row",
        )?;
        Ok(())
    }

    #[test]
    fn exact_mismatch_during_preparation_prevents_observed_installation()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        registry
            .preparation_invalidations
            .store(u64::try_from(VERIFIED_READ_ATTEMPTS)?, Ordering::Release);

        let admission = registry.admit_mutation(&database, temp.path(), None, &test_control())?;

        require(
            matches!(&admission.witness, MutationSourceWitness::Exact { .. }),
            "preparation continuity loss did not fall back to exact source",
        )?;
        require(
            registry.preparation_invalidations.load(Ordering::Acquire) == 0,
            "preparation invalidation attempts were not bounded",
        )?;
        admission.verify()?;

        registry
            .preparation_invalidations
            .store(u64::try_from(VERIFIED_READ_ATTEMPTS)?, Ordering::Release);
        let read = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.overview()?),
        );
        require(
            matches!(read, Err(CliError::RefreshRequired(_))),
            "observer read did not retain continuity-loss recovery",
        )?;
        Ok(())
    }

    #[test]
    fn mutation_verification_redrains_delayed_observer_events() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let admission = registry.admit_mutation(&database, temp.path(), None, &test_control())?;
        let MutationSourceWitness::Observed { entry, .. } = &admission.witness else {
            return Err(
                std::io::Error::other("mutation did not retain an observed witness").into(),
            );
        };
        *entry
            .acceptance_event
            .lock()
            .map_err(|_poisoned| std::io::Error::other("acceptance event lock poisoned"))? =
            Some(Event::new(EventKind::Modify(ModifyKind::Any)).add_path(database));

        admission.verify()?;

        require(
            entry
                .acceptance_event
                .lock()
                .map_err(|_poisoned| std::io::Error::other("acceptance event lock poisoned"))?
                .is_none(),
            "delayed irrelevant event was not consumed",
        )?;
        *entry
            .acceptance_event
            .lock()
            .map_err(|_poisoned| std::io::Error::other("acceptance event lock poisoned"))? =
            Some(Event::new(EventKind::Modify(ModifyKind::Any)).add_path(source));
        require(
            matches!(admission.verify(), Err(CliError::RefreshRequired(_))),
            "delayed relevant event was accepted",
        )?;
        Ok(())
    }

    #[test]
    fn mutation_verification_rechecks_continuity_under_drain_lock() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let admission = registry.admit_mutation(&database, temp.path(), None, &test_control())?;
        let MutationSourceWitness::Observed { entry, .. } = &admission.witness else {
            return Err(
                std::io::Error::other("mutation did not retain an observed witness").into(),
            );
        };
        entry
            .drain_continuity_invalidations
            .store(1, Ordering::Release);

        require(
            matches!(admission.verify(), Err(CliError::RefreshRequired(_))),
            "continuity loss during drain admission was accepted",
        )?;
        require(
            entry.drain_continuity_invalidations.load(Ordering::Acquire) == 0,
            "drain continuity invalidation was not consumed",
        )?;
        Ok(())
    }

    #[test]
    fn mutation_admission_uses_exact_fallback_when_observer_capacity_is_full()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let git = temp.path().join(".git");
        let git_info = git.join("info");
        let git_exclude = git_info.join("exclude");
        fs::create_dir_all(&git_info)?;
        fs::create_dir_all(git.join("objects"))?;
        fs::create_dir_all(git.join("refs"))?;
        fs::write(git.join("HEAD"), "ref: refs/heads/main\n")?;
        fs::write(git.join("config"), "[core]\n")?;
        fs::write(&git_exclude, "")?;
        let registry = SourceObservationRegistry::default();
        let binding = SourceBinding::new(&database, temp.path(), None)?;
        let filler = registry
            .entry(binding.clone())?
            .ok_or_else(|| std::io::Error::other("source observer was unavailable"))?;
        {
            let mut entries = registry
                .entries
                .lock()
                .map_err(|_poisoned| std::io::Error::other("observer registry was poisoned"))?;
            entries.clear();
            for index in 0..SOURCE_OBSERVATION_CAPACITY {
                let mut filler_binding = binding.clone();
                filler_binding.config = Some(temp.path().join(format!("observer-{index}.toml")));
                entries.insert(filler_binding, Arc::clone(&filler));
            }
        }

        let control = test_control();
        let admission = registry.admit_mutation(&database, temp.path(), None, &control)?;
        require(
            matches!(&admission.witness, MutationSourceWitness::Exact { .. }),
            "full observer registry did not use exact mutation admission",
        )?;
        let entries = registry
            .entries
            .lock()
            .map_err(|_poisoned| std::io::Error::other("observer registry was poisoned"))?;
        require(
            entries.len() == SOURCE_OBSERVATION_CAPACITY && !entries.contains_key(&binding),
            "exact mutation admission changed the bounded observer registry",
        )?;
        drop(entries);

        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let transaction = store.begin_purpose_mutation()?;
        store.set_purpose("source.rs", "Exact fallback purpose", PurposeSource::Agent)?;
        admission.verify()?;
        transaction.commit()?;
        require(
            store.load_node_by_path("source.rs")?.is_some_and(|node| {
                node.purpose.purpose.as_deref() == Some("Exact fallback purpose")
            }),
            "exact fallback did not commit an unchanged-source mutation",
        )?;

        let policy_admission = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let before_revision = store.authored_purpose_revision()?;
        let before_purpose = store
            .load_node_by_path("source.rs")?
            .ok_or_else(|| std::io::Error::other("indexed source missing"))?
            .purpose;
        let transaction = store.begin_purpose_mutation()?;
        store.set_purpose("source.rs", "Rejected policy purpose", PurposeSource::Agent)?;
        fs::write(&git_exclude, "never-present-policy-only.tmp\n")?;
        let verification = policy_admission.verify();
        transaction.rollback()?;
        require(
            matches!(verification, Err(CliError::RefreshRequired(_))),
            "exact fallback accepted changed source-selection policy",
        )?;
        require(
            store.authored_purpose_revision()? == before_revision
                && store
                    .load_node_by_path("source.rs")?
                    .is_some_and(|node| node.purpose == before_purpose),
            "policy-drift rejection changed authored purpose",
        )?;

        let replaced = registry.admit_mutation(&database, temp.path(), None, &control)?;
        drop(store);
        fs::write(&source, "fn concurrently_published() {}\n")?;
        let mut publisher = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let plan = ScanRuntimePlan::for_path(None, temp.path(), None)?;
        super::super::run_scan_pipeline(
            &mut publisher,
            &plan,
            &super::super::SymbolBuildOptions::new(
                super::super::MAX_SYMBOL_FILE_BYTES,
                Some(1),
                None,
            ),
        )?;
        drop(publisher);
        verify_saved_source_matches_index_controlled(&database, temp.path(), None, &control)?;
        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let before_revision = store.authored_purpose_revision()?;
        let before_purpose = store
            .load_node_by_path("source.rs")?
            .ok_or_else(|| std::io::Error::other("indexed source missing"))?
            .purpose;
        let transaction = store.begin_purpose_mutation()?;
        store.set_purpose(
            "source.rs",
            "Replaced generation purpose",
            PurposeSource::Agent,
        )?;
        let verification = replaced.verify();
        transaction.rollback()?;
        require(
            matches!(verification, Err(CliError::RefreshRequired(_))),
            "exact fallback accepted a replacement publication generation",
        )?;
        require(
            store.authored_purpose_revision()? == before_revision
                && store
                    .load_node_by_path("source.rs")?
                    .is_some_and(|node| node.purpose == before_purpose),
            "replacement-generation rejection changed authored purpose",
        )?;

        let cancellation = IndexCancellation::new();
        let canceled_control =
            IndexWorkControl::new(cancellation.clone(), Some(Duration::from_secs(30)));
        let canceled = registry.admit_mutation(&database, temp.path(), None, &canceled_control)?;
        cancellation.cancel();
        require(
            matches!(canceled.verify(), Err(CliError::IndexWork(_))),
            "exact fallback did not retain cancellation through verification",
        )?;

        let before_revision = store.authored_purpose_revision()?;
        let before_purpose = store
            .load_node_by_path("source.rs")?
            .ok_or_else(|| std::io::Error::other("indexed source missing"))?
            .purpose;
        let stale = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let transaction = store.begin_purpose_mutation()?;
        store.set_purpose(
            "source.rs",
            "Rejected fallback purpose",
            PurposeSource::Agent,
        )?;
        fs::write(&source, "fn changed_after_exact_admission() {}\n")?;
        let verification = stale.verify();
        transaction.rollback()?;
        require(
            matches!(verification, Err(CliError::RefreshRequired(_))),
            "exact fallback admitted a mutation after saved source changed",
        )?;
        require(
            store.authored_purpose_revision()? == before_revision,
            "rejected exact fallback advanced authored-purpose revision",
        )?;
        require(
            store
                .load_node_by_path("source.rs")?
                .is_some_and(|node| node.purpose == before_purpose),
            "rejected exact fallback changed the purpose row",
        )?;
        Ok(())
    }

    #[test]
    fn post_admission_edit_rolls_back_without_waiting_for_observer_delivery()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let control = test_control();
        let admission = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let before = store
            .load_node_by_path("source.rs")?
            .ok_or_else(|| std::io::Error::other("indexed source missing"))?
            .purpose;
        let before_revision = store.authored_purpose_revision()?;
        let transaction = store.begin_purpose_mutation()?;
        store.set_purpose("source.rs", "Stale purpose", PurposeSource::Agent)?;

        fs::write(&source, "fn changed_after_admission() {}\n")?;
        let verification = admission.verify();
        drop(transaction);
        require(
            matches!(verification, Err(CliError::RefreshRequired(_))),
            "post-admission edit did not invalidate the purpose commit",
        )?;
        require(
            store.authored_purpose_revision()? == before_revision,
            "rolled-back mutation advanced authored-purpose revision",
        )?;
        require(
            store
                .load_node_by_path("source.rs")?
                .is_some_and(|node| node.purpose == before),
            "rolled-back mutation changed the purpose row",
        )?;
        Ok(())
    }

    #[test]
    fn post_admission_cancellation_rolls_back_purpose_mutation() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let cancellation = IndexCancellation::new();
        let control = IndexWorkControl::new(cancellation.clone(), Some(Duration::from_secs(30)));
        let admission = registry.admit_mutation(&database, temp.path(), None, &control)?;
        let store = super::super::open_atlas_store_for_project(&database, temp.path())?;
        let before_revision = store.authored_purpose_revision()?;
        let transaction = store.begin_purpose_mutation()?;
        store.set_purpose("source.rs", "Canceled purpose", PurposeSource::Agent)?;

        cancellation.cancel();
        let verification = admission.verify();
        drop(transaction);
        require(
            matches!(verification, Err(CliError::IndexWork(_))),
            "post-admission cancellation did not invalidate the purpose commit",
        )?;
        require(
            store.authored_purpose_revision()? == before_revision,
            "canceled mutation advanced authored-purpose revision",
        )?;
        require(
            store
                .load_node_by_path("source.rs")?
                .is_none_or(|node| node.purpose.purpose.as_deref() != Some("Canceled purpose")),
            "canceled mutation changed the purpose row",
        )?;
        Ok(())
    }

    #[test]
    fn same_native_path_rejects_distinct_unresolved_paths() -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let left = temp.path().join("missing-left").join("leaf");
        let right = temp.path().join("missing-right").join("leaf");

        require(
            !super::same_native_path(&left, &right),
            "unresolved distinct paths were treated as equal",
        )?;
        Ok(())
    }

    #[test]
    fn same_native_path_matches_paths_below_a_canonicalizable_parent() -> Result<(), Box<dyn Error>>
    {
        let temp = tempfile::tempdir()?;
        let parent = temp.path().join("canonical-parent");
        let nested = parent.join("nested");
        fs::create_dir(&parent)?;
        fs::create_dir(&nested)?;
        let left = parent.join("missing");
        let right = nested.join("..").join("missing");

        require(
            super::same_native_path(&left, &right),
            "equivalent paths below a canonicalizable parent diverged",
        )?;
        Ok(())
    }

    #[cfg(windows)]
    #[test]
    fn observer_filters_case_variant_database_sidecars_without_rescan() -> Result<(), Box<dyn Error>>
    {
        let temp = tempfile::tempdir()?;
        let root = temp.path().join("MiXeDAtlasRoot");
        let metadata = root.join(".projectatlas");
        fs::create_dir_all(&metadata)?;
        let database = metadata.join("ProjectAtlas.db");
        fs::write(&database, b"SQLite format 3\0")?;
        let sidecars = ["-wal", "-shm", "-journal"].map(|suffix| {
            let mut sidecar = database.as_os_str().to_os_string();
            sidecar.push(suffix);
            PathBuf::from(sidecar)
        });
        for sidecar in &sidecars {
            fs::write(sidecar, b"SQLite sidecar")?;
        }
        let source = root.join("source.rs");
        fs::write(&source, "fn source() {}\n")?;
        let binding = SourceBinding::new(&database, &root, None)?;
        let scan_options = projectatlas_fs::ScanOptions::default();
        let differently_cased = |path: &Path| -> Result<PathBuf, Box<dyn Error>> {
            let path = path
                .to_str()
                .ok_or_else(|| std::io::Error::other("test path was not UTF-8"))?;
            Ok(PathBuf::from(path.to_ascii_uppercase()))
        };

        for path in std::iter::once(&database).chain(sidecars.iter()) {
            let event =
                Event::new(EventKind::Modify(ModifyKind::Any)).add_path(differently_cased(path)?);
            let changes = super::observer_event_changes(&binding, &scan_options, &event);
            require(
                !changes.has_changes(),
                "case-variant SQLite runtime event triggered a rescan",
            )?;
        }

        for sidecar in &sidecars {
            fs::remove_file(sidecar)?;
            let event = Event::new(EventKind::Modify(ModifyKind::Any))
                .add_path(differently_cased(sidecar)?);
            let changes = super::observer_event_changes(&binding, &scan_options, &event);
            require(
                !changes.has_changes(),
                "case-variant removed SQLite sidecar event triggered a rescan",
            )?;
        }

        let event = Event::new(EventKind::Modify(ModifyKind::Any)).add_path(source);
        let changes = super::observer_event_changes(&binding, &scan_options, &event);
        require(
            changes.has_changes(),
            "non-runtime source event was incorrectly filtered",
        )?;
        Ok(())
    }

    #[cfg(windows)]
    #[test]
    fn observer_keeps_case_variant_sidecar_event_in_case_sensitive_directory()
    -> Result<(), Box<dyn Error>> {
        let temp = tempfile::tempdir()?;
        let parent = temp.path().join("case-sensitive-observer");
        fs::create_dir(&parent)?;
        let enabled = Command::new("fsutil")
            .args(["file", "SetCaseSensitiveInfo"])
            .arg(&parent)
            .arg("enable")
            .status()
            .is_ok_and(|status| status.success());
        if !enabled {
            return Ok(());
        }
        let root = parent.join("Root");
        let metadata = root.join(".projectatlas");
        fs::create_dir_all(&metadata)?;
        let database = metadata.join("ProjectAtlas.db");
        fs::write(&database, b"SQLite format 3\0")?;
        let binding = SourceBinding::new(&database, &root, None)?;
        let event_path = metadata.join("PROJECTATLAS.DB-WAL");
        let event = Event::new(EventKind::Modify(ModifyKind::Any)).add_path(event_path);
        let changes = super::observer_event_changes(
            &binding,
            &projectatlas_fs::ScanOptions::default(),
            &event,
        );
        require(
            changes.has_changes(),
            "case-variant sidecar event was suppressed in a case-sensitive directory",
        )
    }

    #[test]
    fn cancellation_and_continuity_loss_never_certify_partial_truth() -> Result<(), Box<dyn Error>>
    {
        let temp = tempfile::tempdir()?;
        let (database, _source) = indexed_project(temp.path())?;
        let registry = SourceObservationRegistry::default();
        let initial = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.overview()?),
        )?;
        require(
            initial.work.exact_verifications >= 1,
            "initial read did not establish exact truth",
        )?;
        let binding = SourceBinding::new(&database, temp.path(), None)?;
        let entry = registry
            .entries
            .lock()
            .map_err(|_poisoned| std::io::Error::other("registry lock poisoned"))?
            .get(&binding)
            .cloned()
            .ok_or_else(|| std::io::Error::other("observer entry missing"))?;

        let cancellation = IndexCancellation::new();
        cancellation.cancel();
        let canceled_control = IndexWorkControl::new(cancellation, None);
        let canceled = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &canceled_control,
            |store, _stamp| Ok(store.overview()?),
        );
        require(
            matches!(canceled, Err(CliError::IndexWork(_))),
            "cancelled observer read did not return index-work cancellation",
        )?;
        require(
            entry.current_epoch()?.is_none(),
            "cancellation left the previous verified epoch installed",
        )?;

        let recovered = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.overview()?),
        )?;
        require(
            recovered.work.exact_verifications >= 1,
            "read after cancellation did not establish exact truth",
        )?;
        entry.continuity_lost.store(true, Ordering::Release);
        let reverified = registry.with_verified_read(
            &database,
            temp.path(),
            None,
            &test_control(),
            |store, _stamp| Ok(store.overview()?),
        )?;
        require(
            reverified.work.exact_verifications >= 1,
            "continuity loss did not trigger exact verification",
        )?;
        Ok(())
    }
}