lean_ctx/core/

events.rs

// SPDX-License-Identifier: Apache-2.0

use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::io::Write;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock};

const RING_CAPACITY: usize = 1000;
#[cfg(test)]
const JSONL_MAX_LINES: usize = 10_000;
const EVENT_ID_BLOCK_SIZE: u64 = 1_024;

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct LeanCtxEvent {
    pub id: u64,
    pub timestamp: String,
    pub kind: EventKind,
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(tag = "type")]
pub enum EventKind {
    ToolCall {
        tool: String,
        tokens_original: u64,
        tokens_saved: u64,
        mode: Option<String>,
        duration_ms: u64,
        path: Option<String>,
    },
    CacheHit {
        path: String,
        saved_tokens: u64,
    },
    Compression {
        path: String,
        before_lines: u32,
        after_lines: u32,
        strategy: String,
        kept_line_count: u32,
        removed_line_count: u32,
    },
    AgentAction {
        agent_id: String,
        action: String,
        tool: Option<String>,
    },
    KnowledgeUpdate {
        category: String,
        key: String,
        action: String,
    },
    ThresholdShift {
        language: String,
        old_entropy: f64,
        new_entropy: f64,
        old_jaccard: f64,
        new_jaccard: f64,
    },
    PolicyViolation {
        role: String,
        tool: String,
        reason: String,
    },
    RoleChanged {
        from: String,
        to: String,
    },
    ProfileChanged {
        from: String,
        to: String,
    },
    VerificationWarning {
        warning_kind: String,
        detail: String,
        severity: String,
    },
    ThresholdAdapted {
        language: String,
        arm: String,
        old_threshold: f64,
        new_threshold: f64,
    },
}

/// Event kinds written by older releases whose feature was removed (agent
/// budget limits, SLO evaluation, anomaly detection). Persisted records of
/// these kinds are skipped on read so existing journals and observation stores
/// stay loadable; every other record is still validated strictly.
const RETIRED_EVENT_KINDS: &[&str] = &[
    "BudgetWarning",
    "BudgetExhausted",
    "SloViolation",
    "Anomaly",
];

/// True when `json` is a serialized [`LeanCtxEvent`] of a retired kind.
pub(crate) fn is_retired_event_record(json: &str) -> bool {
    #[derive(Deserialize)]
    struct KindTag {
        #[serde(rename = "type")]
        tag: String,
    }
    #[derive(Deserialize)]
    struct Record {
        kind: KindTag,
    }
    serde_json::from_str::<Record>(json)
        .is_ok_and(|record| RETIRED_EVENT_KINDS.contains(&record.kind.tag.as_str()))
}

struct EventBus {
    ring: Mutex<VecDeque<LeanCtxEvent>>,
}

impl EventBus {
    fn new() -> Self {
        Self {
            ring: Mutex::new(VecDeque::with_capacity(RING_CAPACITY)),
        }
    }

    fn record(&self, event: LeanCtxEvent) {
        let Ok(mut ring) = crate::core::context_os::observation::lock_observation(&self.ring)
        else {
            tracing::warn!(
                "legacy projection unavailable; committed observation retained in ContextBus"
            );
            return;
        };
        if ring.len() >= RING_CAPACITY {
            ring.pop_front();
        }
        ring.push_back(event);
    }

    fn events_since(&self, after_id: u64) -> Vec<LeanCtxEvent> {
        let ring = self
            .ring
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        ring.iter().filter(|e| e.id > after_id).cloned().collect()
    }

    fn latest_events(&self, n: usize) -> Vec<LeanCtxEvent> {
        let ring = self
            .ring
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let len = ring.len();
        let start = len.saturating_sub(n);
        ring.iter().skip(start).cloned().collect()
    }
}

fn bus() -> &'static EventBus {
    static INSTANCE: OnceLock<EventBus> = OnceLock::new();
    INSTANCE.get_or_init(EventBus::new)
}

fn jsonl_path() -> Option<std::path::PathBuf> {
    crate::core::paths::state_dir()
        .ok()
        .map(|d| d.join("events.jsonl"))
}

fn event_sequence_path() -> Option<std::path::PathBuf> {
    crate::core::paths::state_dir()
        .ok()
        .map(|d| d.join("events.seq"))
}

/// Identity-only view of a journal line: retired event kinds still reserve
/// their IDs, so the next allocated ID never collides with old history.
#[derive(Deserialize)]
struct EventIdOnly {
    id: u64,
}

fn max_event_id_in_journal(path: &std::path::Path) -> u64 {
    let Ok(content) = std::fs::read_to_string(path) else {
        return 0;
    };
    content
        .lines()
        .filter_map(|line| serde_json::from_str::<EventIdOnly>(line).ok())
        .map(|event| event.id)
        .max()
        .unwrap_or(0)
}

fn parse_sequence_record(record: &str) -> Option<u64> {
    let (value, checksum) = record.trim().split_once(':')?;
    let value = value.parse::<u64>().ok()?;
    let checksum = checksum.parse::<u64>().ok()?;
    (checksum == !value).then_some(value)
}

fn read_persisted_sequence(path: &std::path::Path) -> Option<u64> {
    let content = std::fs::read_to_string(path).ok()?;
    content.lines().rev().find_map(parse_sequence_record)
}

fn sequence_record(value: u64) -> String {
    format!("{value}:{}", !value)
}

fn reserve_event_id_block_at(
    sequence_path: &std::path::Path,
    journal_path: &std::path::Path,
    block_size: u64,
) -> std::io::Result<(u64, u64)> {
    use fs2::FileExt;

    if block_size == 0 {
        return Err(std::io::Error::other("event id block must not be empty"));
    }
    if let Some(parent) = sequence_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    // Lock a stable companion inode. The append-only sequence journal can be
    // repaired without allowing another process to bypass the lock.
    let lock_path = sequence_path.with_extension("seq.lock");
    let lock = std::fs::OpenOptions::new()
        .create(true)
        .truncate(false)
        .read(true)
        .write(true)
        .open(lock_path)?;
    lock.lock_exclusive()?;

    let result = (|| {
        let persisted = read_persisted_sequence(sequence_path);
        let baseline = persisted.unwrap_or_else(|| {
            max_event_id_in_journal(journal_path).max(max_event_id_in_journal(
                &journal_path.with_extension("jsonl.old"),
            ))
        });
        let first = baseline
            .checked_add(1)
            .ok_or_else(|| std::io::Error::other("event id space exhausted"))?;
        let last = baseline
            .checked_add(block_size)
            .ok_or_else(|| std::io::Error::other("event id space exhausted"))?;

        // A prior writer's record (or corruption) may leave the file without
        // a trailing newline; appending directly would fuse onto that line
        // and make both records unparsable. Start a fresh line first.
        let needs_leading_newline = std::fs::read(sequence_path)
            .is_ok_and(|bytes| !bytes.is_empty() && bytes.last() != Some(&b'\n'));

        let mut sequence = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(sequence_path)?;
        let mut record = Vec::new();
        if needs_leading_newline {
            record.push(b'\n');
        }
        record.extend_from_slice(sequence_record(last).as_bytes());
        record.push(b'\n');
        sequence.write_all(&record)?;
        sequence.sync_data()?;
        Ok((first, last))
    })();

    let _ = FileExt::unlock(&lock);
    result
}

#[cfg(test)]
fn next_event_id_at(
    sequence_path: &std::path::Path,
    journal_path: &std::path::Path,
) -> std::io::Result<u64> {
    reserve_event_id_block_at(sequence_path, journal_path, 1).map(|(first, _)| first)
}

fn next_event_id() -> u64 {
    #[derive(Default)]
    struct LocalBlock {
        next: u64,
        last: u64,
    }

    static BLOCK: OnceLock<Mutex<LocalBlock>> = OnceLock::new();
    static FALLBACK: AtomicU64 = AtomicU64::new(0);
    if !is_test_environment()
        && let Some((sequence, journal)) = event_sequence_path().zip(jsonl_path())
    {
        let mut block = BLOCK
            .get_or_init(|| Mutex::new(LocalBlock::default()))
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if block.next <= block.last && block.next != 0 {
            let id = block.next;
            block.next = block.next.saturating_add(1);
            return id;
        }
        if let Ok((first, last)) =
            reserve_event_id_block_at(&sequence, &journal, EVENT_ID_BLOCK_SIZE)
        {
            block.next = first.saturating_add(1);
            block.last = last;
            return first;
        }
    }

    fallback_event_id(&FALLBACK, chrono::Utc::now().timestamp_millis())
}

/// Clock-derived id (milliseconds × 1000) that is strictly increasing within
/// the process even when the wall clock steps back (an NTP correction on a
/// CI runner): `max(last + 1, now)`. The former `now + counter` could repeat
/// an id after such a step — the canonical store then rejects the event as an
/// identity conflict — or order a later event before an earlier one.
fn fallback_event_id(last: &AtomicU64, now_millis: i64) -> u64 {
    let now = u64::try_from(now_millis)
        .unwrap_or_default()
        .saturating_mul(1_000);
    let mut current = last.load(Ordering::Relaxed);
    loop {
        let next = now.max(current.saturating_add(1));
        match last.compare_exchange_weak(current, next, Ordering::Relaxed, Ordering::Relaxed) {
            Ok(_) => return next,
            Err(observed) => current = observed,
        }
    }
}

fn is_test_environment() -> bool {
    use std::sync::OnceLock;
    static CACHED: OnceLock<bool> = OnceLock::new();
    *CACHED.get_or_init(|| {
        if cfg!(test) {
            return true;
        }
        if std::env::var_os("__LEAN_CTX_SKIP_EVENTS").is_some() {
            return true;
        }
        std::env::current_exe().is_ok_and(|p| {
            let s = p.to_string_lossy();
            s.contains("/deps/") || s.contains("\\deps\\")
        })
    })
}

#[cfg(test)]
fn append_jsonl_at(path: &std::path::Path, event: &LeanCtxEvent) -> std::io::Result<()> {
    use fs2::FileExt;
    use std::io::Write;

    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }

    // Lock a stable companion inode rather than events.jsonl itself: rotation
    // renames the data file, so locking that inode would let another process
    // open the replacement and bypass the lock.
    let lock_path = path.with_extension("jsonl.lock");
    let lock = std::fs::OpenOptions::new()
        .create(true)
        .truncate(false)
        .read(true)
        .write(true)
        .open(lock_path)?;
    lock.lock_exclusive()?;

    let result = (|| {
        if let Ok(content) = std::fs::read_to_string(path)
            && content.lines().count() >= JSONL_MAX_LINES
        {
            let old = path.with_extension("jsonl.old");
            let _ = std::fs::remove_file(&old);
            std::fs::rename(path, old)?;
        }

        let json = serde_json::to_string(event).map_err(std::io::Error::other)?;
        let mut line = json.into_bytes();
        line.push(b'\n');
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)?;
        // One logical write while holding the interprocess lock prevents JSON
        // objects and their trailing newlines from interleaving.
        file.write_all(&line)
    })();

    let _ = FileExt::unlock(&lock);
    result
}

// --- Public API ---

/// Persist a legacy event as a canonical local observation.
///
/// This compatibility adapter deliberately never appends to `events.jsonl`;
/// that journal remains an immutable historical read source.
///
/// In a serving process (after [`enable_detached_observations`]) the write is
/// handed to a dedicated writer thread: a tool handler never waits on the
/// observation lock or SQLite (#2196), and when the bounded queue is full the
/// observation is dropped and counted instead of blocking dispatch. Storage
/// admission still runs here, on the producer's thread and policy view.
pub fn emit_compatibility(kind: EventKind) -> u64 {
    let result = if DETACHED_ENABLED.load(Ordering::Acquire) {
        emit_detached(kind)
    } else {
        try_emit(kind)
    };
    match result {
        Ok(id) => id,
        Err(error) => {
            log_persistence_failure(&error);
            0
        }
    }
}

/// [`emit_compatibility`] that reports why persistence failed instead of
/// collapsing the failure into id `0`. Always synchronous.
pub(crate) fn try_emit(
    kind: EventKind,
) -> Result<u64, crate::core::context_os::ObservationPersistenceError> {
    persist_admitted(&admit_observation(kind)?)
}

fn admit_observation(
    kind: EventKind,
) -> Result<
    crate::core::context_os::AdmittedLegacyObservation,
    crate::core::context_os::ObservationPersistenceError,
> {
    let event = LeanCtxEvent {
        id: next_event_id(),
        timestamp: chrono::Local::now()
            .format("%Y-%m-%dT%H:%M:%S%.3f")
            .to_string(),
        kind,
    };
    crate::core::context_os::ContextBus::admit_legacy_observation(&event)
}

fn persist_admitted(
    admitted: &crate::core::context_os::AdmittedLegacyObservation,
) -> Result<u64, crate::core::context_os::ObservationPersistenceError> {
    crate::core::context_os::try_runtime().and_then(|runtime| {
        runtime.bus.append_admitted_legacy_observation(
            crate::core::context_os::observation::producer_instance(),
            admitted,
        )
    })?;
    bus().record(admitted.event.clone());
    Ok(admitted.event.id)
}

/// A withheld observation is the gateway working as configured, not a fault:
/// at warn level it flooded the worker log (765 lines in 5 minutes, #2196).
fn log_persistence_failure(error: &crate::core::context_os::ObservationPersistenceError) {
    if matches!(
        error,
        crate::core::context_os::ObservationPersistenceError::Withheld
    ) {
        tracing::debug!("canonical legacy observation persistence skipped: {error}");
    } else {
        tracing::warn!("canonical legacy observation persistence failed: {error}");
    }
}

// --- Detached persistence for serving processes (#2196) ---

/// Observations a serving process may have waiting for the writer thread.
const DETACHED_QUEUE_CAPACITY: usize = 4_096;

static DETACHED_ENABLED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
static DETACHED_QUEUED: AtomicU64 = AtomicU64::new(0);
static DETACHED_PERSISTED: AtomicU64 = AtomicU64::new(0);
static DETACHED_FAILED: AtomicU64 = AtomicU64::new(0);
static DETACHED_DROPPED: AtomicU64 = AtomicU64::new(0);

enum DetachedMessage {
    Persist(crate::core::context_os::AdmittedLegacyObservation),
    Flush(std::sync::mpsc::SyncSender<()>),
}

/// Counters of the detached observation writer, for doctor and health.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct DetachedObservationStats {
    pub enabled: bool,
    pub queued: u64,
    pub persisted: u64,
    pub failed: u64,
    pub dropped: u64,
}

pub fn detached_observation_stats() -> DetachedObservationStats {
    DetachedObservationStats {
        enabled: DETACHED_ENABLED.load(Ordering::Acquire),
        queued: DETACHED_QUEUED.load(Ordering::Relaxed),
        persisted: DETACHED_PERSISTED.load(Ordering::Relaxed),
        failed: DETACHED_FAILED.load(Ordering::Relaxed),
        dropped: DETACHED_DROPPED.load(Ordering::Relaxed),
    }
}

/// Route [`emit_compatibility`] through the detached writer for the rest of
/// this process. Called once by a serving process before it accepts requests;
/// one-shot CLI commands keep synchronous persistence.
pub fn enable_detached_observations() {
    if detached_sender().is_some() {
        DETACHED_ENABLED.store(true, Ordering::Release);
    }
}

fn detached_sender() -> Option<&'static std::sync::mpsc::SyncSender<DetachedMessage>> {
    static SENDER: OnceLock<Option<std::sync::mpsc::SyncSender<DetachedMessage>>> = OnceLock::new();
    SENDER
        .get_or_init(|| {
            let (sender, receiver) =
                std::sync::mpsc::sync_channel::<DetachedMessage>(DETACHED_QUEUE_CAPACITY);
            std::thread::Builder::new()
                .name("lean-ctx-observations".to_string())
                .spawn(move || run_detached_writer(&receiver))
                .map_err(|error| {
                    tracing::warn!("detached observation writer unavailable: {error}");
                })
                .ok()
                .map(|_| sender)
        })
        .as_ref()
}

fn run_detached_writer(receiver: &std::sync::mpsc::Receiver<DetachedMessage>) {
    while let Ok(message) = receiver.recv() {
        match message {
            DetachedMessage::Persist(admitted) => {
                // A panic in one write must not end the writer: the queue
                // behind it would be lost and every later flush would wait
                // for nothing.
                let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    persist_with_retry(&admitted)
                }));
                match outcome {
                    Ok(Ok(_)) => {
                        DETACHED_PERSISTED.fetch_add(1, Ordering::Relaxed);
                    }
                    Ok(Err(error)) => {
                        DETACHED_FAILED.fetch_add(1, Ordering::Relaxed);
                        log_persistence_failure(&error);
                    }
                    Err(_) => {
                        DETACHED_FAILED.fetch_add(1, Ordering::Relaxed);
                        tracing::warn!("canonical legacy observation persistence panicked");
                    }
                }
            }
            DetachedMessage::Flush(done) => {
                let _ = done.send(());
            }
        }
    }
}

/// The writer thread may wait where a request may not: one lock timeout
/// there (another process holding the store) is retried once after a short
/// pause before the observation is counted as failed. Each attempt already
/// waits the full lock bound, so more retries would only stall the queue.
fn persist_with_retry(
    admitted: &crate::core::context_os::AdmittedLegacyObservation,
) -> Result<u64, crate::core::context_os::ObservationPersistenceError> {
    match persist_admitted(admitted) {
        Err(crate::core::context_os::ObservationPersistenceError::LockTimeout) => {
            std::thread::sleep(std::time::Duration::from_millis(250));
            persist_admitted(admitted)
        }
        result => result,
    }
}

fn emit_detached(
    kind: EventKind,
) -> Result<u64, crate::core::context_os::ObservationPersistenceError> {
    let admitted = admit_observation(kind)?;
    let id = admitted.event.id;
    let Some(sender) = detached_sender() else {
        return persist_admitted(&admitted);
    };
    match sender.try_send(DetachedMessage::Persist(admitted)) {
        Ok(()) => {
            DETACHED_QUEUED.fetch_add(1, Ordering::Relaxed);
            Ok(id)
        }
        Err(std::sync::mpsc::TrySendError::Full(_)) => {
            // Dropping one observation is the price of never stalling a
            // request on a saturated store; the counter keeps it visible.
            let dropped = DETACHED_DROPPED.fetch_add(1, Ordering::Relaxed) + 1;
            if dropped.is_power_of_two() {
                tracing::warn!(
                    "observation writer saturated; {dropped} observations dropped so far"
                );
            }
            Ok(0)
        }
        Err(std::sync::mpsc::TrySendError::Disconnected(DetachedMessage::Persist(admitted))) => {
            persist_admitted(&admitted)
        }
        Err(std::sync::mpsc::TrySendError::Disconnected(DetachedMessage::Flush(_))) => Ok(0),
    }
}

/// Wait until every observation queued before this call is persisted, or
/// `timeout` passes. Returns whether the queue drained. Used at shutdown so a
/// serving process does not lose its last observations.
pub fn flush_detached_observations(timeout: std::time::Duration) -> bool {
    if !DETACHED_ENABLED.load(Ordering::Acquire) {
        return true;
    }
    let Some(sender) = detached_sender() else {
        return true;
    };
    let deadline = std::time::Instant::now() + timeout;
    let (done, drained) = std::sync::mpsc::sync_channel(1);
    let mut message = DetachedMessage::Flush(done);
    loop {
        match sender.try_send(message) {
            Ok(()) => break,
            Err(std::sync::mpsc::TrySendError::Full(returned)) => {
                if std::time::Instant::now() >= deadline {
                    return false;
                }
                message = returned;
                std::thread::sleep(std::time::Duration::from_millis(5));
            }
            // The writer is gone: whatever it still held is lost.
            Err(std::sync::mpsc::TrySendError::Disconnected(_)) => return false,
        }
    }
    let remaining = deadline.saturating_duration_since(std::time::Instant::now());
    drained.recv_timeout(remaining).is_ok()
}

/// Backward-compatible spelling for the explicit legacy compatibility adapter.
pub fn emit(kind: EventKind) -> u64 {
    emit_compatibility(kind)
}

pub fn events_since(after_id: u64) -> Vec<LeanCtxEvent> {
    bus().events_since(after_id)
}

pub fn latest_events(n: usize) -> Vec<LeanCtxEvent> {
    bus().latest_events(n)
}

#[derive(Default)]
struct FileEventCache {
    path: Option<std::path::PathBuf>,
    mtime: Option<std::time::SystemTime>,
    len: u64,
    events: Vec<LeanCtxEvent>,
}

fn load_historical_events(
    path: &std::path::Path,
) -> Result<Vec<LeanCtxEvent>, crate::core::context_os::ObservationPersistenceError> {
    static CACHE: OnceLock<Mutex<FileEventCache>> = OnceLock::new();
    let metadata = match std::fs::metadata(path) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(error) => return Err(error.into()),
    };
    let (mtime, len) = (metadata.modified().ok(), metadata.len());
    let cache = CACHE.get_or_init(|| Mutex::new(FileEventCache::default()));
    let mut guard = crate::core::context_os::observation::lock_observation(cache)?;
    let fresh = guard.path.as_deref() == Some(path) && guard.mtime == mtime && guard.len == len;
    if !fresh {
        let content = std::fs::read_to_string(path)?;
        let events = content
            .lines()
            .filter(|line| !line.trim().is_empty() && !is_retired_event_record(line))
            .map(serde_json::from_str)
            .collect::<Result<Vec<_>, _>>()?;
        // Replace the cache only after every nonempty record validates.
        guard.events = events;
        guard.path = Some(path.to_path_buf());
        guard.mtime = mtime;
        guard.len = len;
    }
    Ok(guard.events.clone())
}

/// Compatibility reader: an unavailable canonical store returns no events,
/// never apparently current history. Use the fallible API to inspect the error.
pub fn load_events_from_file(n: usize) -> Vec<LeanCtxEvent> {
    match try_load_events_from_file(n) {
        Ok(events) => events,
        Err(error) => {
            tracing::warn!("canonical event history unavailable: {error}");
            Vec::new()
        }
    }
}

/// [`load_events_from_file`] for a surface that SHOWS events — paths, names,
/// free text — to a host or a model, for a reader working in `project` (the
/// project of the running request when `None`). Every event passes the
/// serve-time rule of the observation store (`ObservationGate`): an event
/// that names a file is shown only while that file is admitted now; one
/// without a source on record follows the originless rule. Historical JSONL
/// records and rows written before admission existed are asked the same way.
/// Counts and timestamps need no gate: use [`load_events_from_file`].
///
/// With a reader project the whole call — the admission snapshot, every
/// payload check and every source read — runs inside ONE pinned policy view
/// of that project (`policy::runtime::with_project_source_view`): every read
/// sees the same project policy and role, and the view compares both again
/// when it ends. A view that cannot be opened, that is not the view of the
/// reader's project, or that reports a change shows NOTHING for this call.
/// A reader without a project has no view; only the originless rule applies.
pub fn load_served_events(n: usize, project: Option<&std::path::Path>) -> Vec<LeanCtxEvent> {
    use crate::core::context_os::observation::ObservationGate;

    let request = crate::core::admitted_sources::request_root();
    let reader = project
        .map(|root| root.to_string_lossy().into_owned())
        .or_else(|| (!request.is_empty()).then_some(request));
    let Some(reader) = reader else {
        return served_through(ObservationGate::for_project(None), n);
    };
    crate::core::policy::runtime::with_project_source_view(&reader, || {
        // The view pins the project the request is already bound to, if any.
        // The gate reads under that project; a reader that asked for another
        // one is not served from it.
        // Compared in the view's own spelling: it pins `Path::canonicalize`
        // of the project (`policy::runtime::canonical_project_root`).
        let pinned = crate::core::admitted_sources::request_root();
        let same = std::path::Path::new(&reader)
            .canonicalize()
            .is_ok_and(|asked| asked == std::path::Path::new(&pinned));
        if same {
            served_through(ObservationGate::for_project(None), n)
        } else {
            Vec::new()
        }
    })
    .unwrap_or_else(|error| {
        tracing::debug!("[events] nothing shown: {error}");
        Vec::new()
    })
}

fn served_through(
    mut gate: crate::core::context_os::observation::ObservationGate,
    n: usize,
) -> Vec<LeanCtxEvent> {
    let mut events = load_events_from_file(n);
    events.retain(|event| gate.serves(event));
    // The view pins project policy and role, not the gateway rules: those are
    // compared once more before anything is returned. Changed mid-call:
    // nothing is shown.
    if !gate.gateway_unchanged() {
        tracing::debug!("[events] nothing shown: gateway rules changed during the call");
        return Vec::new();
    }
    if gate.withheld() > 0 {
        tracing::debug!(
            "[events] {} observation(s) withheld: no current source authority",
            gate.withheld()
        );
    }
    events
}

/// Merge immutable historical JSONL with validated canonical observations.
pub fn try_load_events_from_file(
    n: usize,
) -> Result<Vec<LeanCtxEvent>, crate::core::context_os::ObservationPersistenceError> {
    let mut events = jsonl_path()
        .map(|path| load_historical_events(&path))
        .transpose()?
        .unwrap_or_default();

    let compatibility = crate::core::context_os::try_runtime()?
        .bus
        .read_legacy_observations(n.max(1))?;
    events.extend(compatibility);
    // Historical file order predates the canonical cursor. Block-allocated
    // IDs from concurrent producers do not encode append order.
    // Never discard distinct records just because historical/fallback IDs
    // collide. The legacy projection omits producer identity, so even equal
    // projected payloads cannot establish that two observations are duplicates.
    let start = events.len().saturating_sub(n);
    Ok(events[start..].to_vec())
}

pub fn emit_tool_call(
    tool: &str,
    tokens_original: u64,
    tokens_saved: u64,
    mode: Option<String>,
    duration_ms: u64,
    path: Option<String>,
) {
    emit(EventKind::ToolCall {
        tool: tool.to_string(),
        tokens_original,
        tokens_saved,
        mode,
        duration_ms,
        path,
    });
}

pub fn emit_cache_hit(path: &str, saved_tokens: u64) {
    emit(EventKind::CacheHit {
        path: path.to_string(),
        saved_tokens,
    });
}

pub fn emit_agent_action(agent_id: &str, action: &str, tool: Option<&str>) {
    emit(EventKind::AgentAction {
        agent_id: agent_id.to_string(),
        action: action.to_string(),
        tool: tool.map(std::string::ToString::to_string),
    });
}

pub fn emit_policy_violation(role: &str, tool: &str, reason: &str) {
    emit(EventKind::PolicyViolation {
        role: role.to_string(),
        tool: tool.to_string(),
        reason: reason.to_string(),
    });
}

pub fn emit_role_changed(from: &str, to: &str) {
    emit(EventKind::RoleChanged {
        from: from.to_string(),
        to: to.to_string(),
    });
}

pub fn emit_profile_changed(from: &str, to: &str) {
    emit(EventKind::ProfileChanged {
        from: from.to_string(),
        to: to.to_string(),
    });
}

pub fn emit_verification_warning(warning_kind: &str, detail: &str, severity: &str) {
    emit(EventKind::VerificationWarning {
        warning_kind: warning_kind.to_string(),
        detail: detail.to_string(),
        severity: severity.to_string(),
    });
}

pub fn emit_threshold_adapted(language: &str, arm: &str, old_threshold: f64, new_threshold: f64) {
    emit(EventKind::ThresholdAdapted {
        language: language.to_string(),
        arm: arm.to_string(),
        old_threshold,
        new_threshold,
    });
}

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

    /// Every parallel test in this process emits into one shared store, and
    /// the write lock wait is bounded on purpose (fire-and-forget emitters).
    /// On a loaded Windows runner that queue can exceed the bound, so only a
    /// `LockTimeout` is retried; any other persistence failure still fails.
    fn emit_persisted(kind: &EventKind) -> u64 {
        use crate::core::context_os::ObservationPersistenceError;
        // Five backoff attempts (~3 s) still lost the queue on a loaded
        // windows-latest lib run (2026-10-07, 12k tests in parallel); bound
        // the retry by wall clock instead.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
        let mut attempt = 0u32;
        loop {
            match try_emit(kind.clone()) {
                Ok(id) => return id,
                Err(ObservationPersistenceError::LockTimeout)
                    if std::time::Instant::now() < deadline =>
                {
                    std::thread::sleep(std::time::Duration::from_millis(100 << attempt.min(4)));
                    attempt += 1;
                }
                Err(ObservationPersistenceError::LockTimeout) => {
                    panic!("canonical persistence: write lock still contended after 30 s");
                }
                Err(error) => panic!("canonical persistence: {error:?}"),
            }
        }
    }

    #[test]
    fn emit_returns_positive_id() {
        // Keep real persistence, without queueing behind the whole suite's emitters.
        if crate::test_env::run_in_own_process("core::events::tests::emit_returns_positive_id", &[])
        {
            return;
        }
        let id = emit_persisted(&EventKind::ToolCall {
            tool: "ctx_read".to_string(),
            tokens_original: 1000,
            tokens_saved: 800,
            mode: Some("map".to_string()),
            duration_ms: 5,
            path: Some("src/main.rs".to_string()),
        });
        assert!(id > 0);
        let events = latest_events(100);
        assert!(events.iter().any(|e| e.id == id));
    }

    /// #2196: a tool handler emitting an observation must not wait on the
    /// observation writer. With the writer held (a long transaction or a
    /// contended store), the synchronous path blocks for the full lock bound
    /// and fails; a serving process returns at once and persists the
    /// observation after the writer is released.
    #[test]
    fn serving_emit_does_not_wait_for_a_held_observation_lock() {
        if crate::test_env::run_in_own_process(
            "core::events::tests::serving_emit_does_not_wait_for_a_held_observation_lock",
            &[],
        ) {
            return;
        }
        let kind = || EventKind::ToolCall {
            tool: "ctx_read".to_string(),
            tokens_original: 10,
            tokens_saved: 5,
            mode: Some("full".to_string()),
            duration_ms: 1,
            path: None,
        };
        let runtime = crate::core::context_os::try_runtime().expect("runtime");
        let writer = runtime.bus.hold_observation_writer();

        // The pre-#2196 dispatch path: the caller sleeps out the lock bound.
        let started = std::time::Instant::now();
        let blocked = try_emit(kind());
        let synchronous_wait = started.elapsed();
        assert!(
            matches!(
                blocked,
                Err(crate::core::context_os::ObservationPersistenceError::LockTimeout)
            ),
            "synchronous emit with the writer held: {blocked:?}"
        );
        assert!(synchronous_wait >= std::time::Duration::from_secs(4));

        enable_detached_observations();
        let started = std::time::Instant::now();
        let id = emit_compatibility(kind());
        let detached_wait = started.elapsed();
        assert!(id > 0, "the observation was admitted and queued");
        assert!(
            detached_wait < std::time::Duration::from_secs(1),
            "dispatch waited {detached_wait:?} on a held observation lock"
        );
        assert!(!latest_events(1_000).iter().any(|e| e.id == id));

        drop(writer);
        assert!(flush_detached_observations(std::time::Duration::from_secs(
            30
        )));
        assert!(latest_events(1_000).iter().any(|e| e.id == id));
        let stats = detached_observation_stats();
        assert_eq!((stats.queued, stats.persisted, stats.dropped), (1, 1, 0));
    }

    /// Project an event into the in-memory ring only. The filter tests are
    /// about `events_since`, not persistence: routing them through the shared
    /// SQLite observation store made them queue behind every other emitting
    /// test, and on windows-latest that queue outlasted even a 30 s retry.
    fn record_in_ring(kind: EventKind) -> u64 {
        let id = next_event_id();
        bus().record(LeanCtxEvent {
            id,
            timestamp: chrono::Local::now()
                .format("%Y-%m-%dT%H:%M:%S%.3f")
                .to_string(),
            kind,
        });
        id
    }

    #[test]
    fn events_since_filters_correctly() {
        let id1 = record_in_ring(EventKind::CacheHit {
            path: "filter_test_a.rs".to_string(),
            saved_tokens: 100,
        });
        let id2 = record_in_ring(EventKind::CacheHit {
            path: "filter_test_b.rs".to_string(),
            saved_tokens: 200,
        });

        let after = events_since(id1);
        assert!(after.iter().any(|e| e.id == id2));
        assert!(after.iter().all(|e| e.id > id1));
    }

    #[test]
    fn fallback_ids_stay_unique_and_increasing_when_the_clock_steps_back() {
        let last = AtomicU64::new(0);
        let first = fallback_event_id(&last, 1_000);
        let same_ms = fallback_event_id(&last, 1_000);
        let stepped_back = fallback_event_id(&last, 990);
        let later = fallback_event_id(&last, 2_000);
        assert_eq!(first, 1_000_000);
        assert!(same_ms > first);
        assert!(
            stepped_back > same_ms,
            "a backwards clock step must not reuse or reorder ids"
        );
        assert_eq!(later, 2_000_000);
    }

    /// The (path, mtime, len) cache must never serve stale events: appending a
    /// line changes the file length, which has nanosecond-independent
    /// granularity (unlike mtime), so new events show up on the next poll.
    /// An isolated journal keeps unrelated canonical emissions from evicting
    /// these historical fixtures from the merged public reader's result window.
    #[test]
    fn load_events_from_file_sees_appended_events() {
        let directory = tempfile::tempdir().expect("isolated historical journal");
        let path = directory.path().join("events.jsonl");

        let line_a = r#"{"id":900001,"timestamp":"2026-06-12T08:00:00.000","kind":{"type":"CacheHit","path":"cached_a.rs","saved_tokens":42}}"#;
        std::fs::write(&path, format!("{line_a}\n")).expect("write events.jsonl");

        let first = load_historical_events(&path).expect("read initial historical events");
        assert!(
            first.iter().any(|e| e.id == 900_001),
            "initial load should parse the seeded event"
        );

        // Second call with unchanged file exercises the cached branch.
        let cached = load_historical_events(&path).expect("read cached historical events");
        assert_eq!(cached.len(), first.len());

        let line_b = r#"{"id":900002,"timestamp":"2026-06-12T08:00:01.000","kind":{"type":"CacheHit","path":"cached_b.rs","saved_tokens":7}}"#;
        {
            use std::io::Write;
            let mut f = std::fs::OpenOptions::new()
                .append(true)
                .open(&path)
                .expect("append events.jsonl");
            writeln!(f, "{line_b}").expect("append line");
        }

        let second = load_historical_events(&path).expect("read appended historical events");
        assert!(
            second.iter().any(|e| e.id == 900_002),
            "append must invalidate the cache and surface the new event"
        );
    }

    #[test]
    fn events_jsonl_writer_child() {
        let Some(path) = std::env::var_os("__LEAN_CTX_EVENTS_TEST_PATH") else {
            return;
        };
        let writer = std::env::var("__LEAN_CTX_EVENTS_TEST_WRITER")
            .expect("writer id")
            .parse::<u64>()
            .expect("numeric writer id");
        for sequence in 0..100 {
            let event = LeanCtxEvent {
                id: writer * 100 + sequence,
                timestamp: "2026-07-15T12:00:00.000".to_string(),
                kind: EventKind::CacheHit {
                    path: format!("writer-{writer}.rs"),
                    saved_tokens: sequence,
                },
            };
            append_jsonl_at(std::path::Path::new(&path), &event).expect("append event");
        }
    }

    fn test_event(id: u64) -> LeanCtxEvent {
        LeanCtxEvent {
            id,
            timestamp: "2026-07-15T12:00:00.000".to_string(),
            kind: EventKind::CacheHit {
                path: "event-id-test.rs".to_string(),
                saved_tokens: 1,
            },
        }
    }

    #[test]
    fn retired_event_kinds_are_skipped_and_still_reserve_ids() {
        let dir = tempfile::tempdir().expect("temp dir");
        let journal = dir.path().join("events.jsonl");
        let retired = r#"{"id":90,"timestamp":"2026-07-15T12:00:00.000","kind":{"type":"BudgetWarning","role":"coder","dimension":"tokens","used":"1","limit":"2","percent":50}}"#;
        assert!(is_retired_event_record(retired));
        let current = serde_json::to_string(&test_event(42)).unwrap();
        assert!(!is_retired_event_record(&current));
        std::fs::write(&journal, format!("{current}\n{retired}\n")).unwrap();

        let events = load_historical_events(&journal).expect("retired kinds must not fail reads");
        assert_eq!(events, vec![test_event(42)]);
        assert_eq!(max_event_id_in_journal(&journal), 90);
    }

    #[test]
    fn concurrent_processes_append_complete_json_lines() {
        let dir = tempfile::tempdir().expect("temp dir");
        let path = dir.path().join("events.jsonl");
        let executable = std::env::current_exe().expect("test executable");

        let mut children = Vec::new();
        for writer in 0..4 {
            children.push(
                std::process::Command::new(&executable)
                    .args([
                        "--exact",
                        "core::events::tests::events_jsonl_writer_child",
                        "--nocapture",
                    ])
                    .env("__LEAN_CTX_EVENTS_TEST_PATH", &path)
                    .env("__LEAN_CTX_EVENTS_TEST_WRITER", writer.to_string())
                    .spawn()
                    .expect("spawn writer"),
            );
        }
        for mut child in children {
            assert!(child.wait().expect("wait for writer").success());
        }

        let content = std::fs::read_to_string(&path).expect("read events");
        let lines: Vec<_> = content.lines().collect();
        assert_eq!(lines.len(), 400, "every append must produce one line");
        for line in lines {
            serde_json::from_str::<LeanCtxEvent>(line)
                .unwrap_or_else(|error| panic!("invalid JSONL line: {error}: {line}"));
        }
    }

    #[test]
    fn concurrent_processes_serialize_rotation_and_append() {
        let dir = tempfile::tempdir().expect("temp dir");
        let path = dir.path().join("events.jsonl");
        let seed = serde_json::to_string(&LeanCtxEvent {
            id: 1,
            timestamp: "2026-07-15T12:00:00.000".to_string(),
            kind: EventKind::CacheHit {
                path: "seed.rs".to_string(),
                saved_tokens: 1,
            },
        })
        .expect("serialize seed");
        std::fs::write(&path, format!("{seed}\n").repeat(JSONL_MAX_LINES))
            .expect("seed rotation threshold");

        let executable = std::env::current_exe().expect("test executable");
        let mut children = Vec::new();
        for writer in 0..4 {
            children.push(
                std::process::Command::new(&executable)
                    .args(["--exact", "core::events::tests::events_jsonl_writer_child"])
                    .env("__LEAN_CTX_EVENTS_TEST_PATH", &path)
                    .env("__LEAN_CTX_EVENTS_TEST_WRITER", writer.to_string())
                    .spawn()
                    .expect("spawn writer"),
            );
        }
        for mut child in children {
            assert!(child.wait().expect("wait for writer").success());
        }

        let old =
            std::fs::read_to_string(path.with_extension("jsonl.old")).expect("rotated journal");
        assert_eq!(old.lines().count(), JSONL_MAX_LINES);
        assert!(
            old.lines()
                .all(|line| serde_json::from_str::<LeanCtxEvent>(line).is_ok()),
            "rotated journal must contain complete JSON lines"
        );

        let current = std::fs::read_to_string(&path).expect("current journal");
        assert_eq!(current.lines().count(), 400);
        assert!(
            current
                .lines()
                .all(|line| serde_json::from_str::<LeanCtxEvent>(line).is_ok()),
            "replacement journal must contain complete JSON lines"
        );
    }

    #[test]
    fn persistent_event_id_writer_child() {
        let Some(dir) = std::env::var_os("__LEAN_CTX_EVENT_ID_TEST_DIR") else {
            return;
        };
        let writer = std::env::var("__LEAN_CTX_EVENT_ID_TEST_WRITER").expect("writer id");
        let dir = std::path::PathBuf::from(dir);
        let sequence = dir.join("events.seq");
        let journal = dir.join("events.jsonl");
        let (first, last) =
            reserve_event_id_block_at(&sequence, &journal, 100).expect("reserve event id block");
        let ids: Vec<String> = (first..=last).map(|id| id.to_string()).collect();
        std::fs::write(dir.join(format!("writer-{writer}.ids")), ids.join("\n"))
            .expect("write allocated ids");
    }

    #[test]
    fn persistent_event_id_bootstraps_above_existing_journals() {
        let dir = tempfile::tempdir().expect("temp dir");
        let sequence = dir.path().join("events.seq");
        let journal = dir.path().join("events.jsonl");
        let old = journal.with_extension("jsonl.old");
        std::fs::write(
            &journal,
            format!("{}\n", serde_json::to_string(&test_event(41)).unwrap()),
        )
        .unwrap();
        std::fs::write(
            &old,
            format!("{}\n", serde_json::to_string(&test_event(73)).unwrap()),
        )
        .unwrap();

        assert_eq!(next_event_id_at(&sequence, &journal).unwrap(), 74);
        assert_eq!(next_event_id_at(&sequence, &journal).unwrap(), 75);
    }

    #[test]
    fn corrupt_sequence_recovers_above_existing_journal() {
        let dir = tempfile::tempdir().expect("temp dir");
        let sequence = dir.path().join("events.seq");
        let journal = dir.path().join("events.jsonl");
        std::fs::write(&sequence, "76:broken").unwrap();
        std::fs::write(
            &journal,
            format!("{}\n", serde_json::to_string(&test_event(80)).unwrap()),
        )
        .unwrap();

        assert_eq!(next_event_id_at(&sequence, &journal).unwrap(), 81);
        assert_eq!(read_persisted_sequence(&sequence), Some(81));
    }

    #[test]
    fn concurrent_processes_allocate_unique_event_ids() {
        let dir = tempfile::tempdir().expect("temp dir");
        let executable = std::env::current_exe().expect("test executable");
        let mut children = Vec::new();
        for writer in 0..4 {
            children.push(
                std::process::Command::new(&executable)
                    .args([
                        "--exact",
                        "core::events::tests::persistent_event_id_writer_child",
                    ])
                    .env("__LEAN_CTX_EVENT_ID_TEST_DIR", dir.path())
                    .env("__LEAN_CTX_EVENT_ID_TEST_WRITER", writer.to_string())
                    .spawn()
                    .expect("spawn event id writer"),
            );
        }
        for mut child in children {
            assert!(child.wait().expect("wait for event id writer").success());
        }

        let mut ids = Vec::new();
        for writer in 0..4 {
            let content = std::fs::read_to_string(dir.path().join(format!("writer-{writer}.ids")))
                .expect("read allocated ids");
            ids.extend(content.lines().map(|line| line.parse::<u64>().unwrap()));
        }
        ids.sort_unstable();
        ids.dedup();
        assert_eq!(ids.len(), 400, "every process must receive unique ids");
        assert_eq!(ids.first(), Some(&1));
        assert_eq!(ids.last(), Some(&400));
        let sequence_path = dir.path().join("events.seq");
        assert_eq!(read_persisted_sequence(&sequence_path), Some(400));
        assert_eq!(
            std::fs::read_to_string(sequence_path)
                .expect("read sequence journal")
                .lines()
                .count(),
            4,
            "four writers should persist four block reservations, not 400 ids"
        );
    }
}