openlatch-client 0.6.23

//! Replay of `~/.openlatch/logs/fallback.jsonl` — hook-binary events
//! captured while the daemon was unreachable.
//!
//! Security note: hook entries are **pre-privacy-filter**. Every envelope
//! must go through `handlers::process_envelope` on replay — skipping the
//! filter would leak unredacted content to the cloud. Progress is
//! persisted to a sibling `.offset` file so mid-replay crashes resume
//! cleanly.

use std::io::{BufRead, BufReader, Seek, SeekFrom};
use std::path::Path;
use std::sync::Arc;

use crate::core::cloud::offset::{
    advance_past_oldest, count_entries_from, read_offset, write_offset, AdvanceStats,
};
use crate::core::envelope::EventEnvelope;
use crate::daemon::handlers;
use crate::daemon::AppState;

/// Name of the fallback log written by `openlatch-hook` when the daemon is
/// unreachable. Matches the path resolution in
/// `src/app/openlatch_hook/main.rs::openlatch_log_dir()`.
const FALLBACK_FILENAME: &str = "fallback.jsonl";
const OFFSET_FILENAME: &str = "fallback.jsonl.offset";
/// The capture-time cutoff of the last organization switch (PLT-7 C-8), as
/// Unix milliseconds. An entry captured before it belongs to the old
/// organization and is never replayed, even one the hook appended after apply
/// moved the cursor (it built the envelope, failed to reach the daemon, then
/// wrote it).
const SWITCH_CUTOFF_FILENAME: &str = "fallback.switch-cutoff";

/// The replay loop. Returns immediately when `state.cloud_state` is `None`
/// (cloud forwarding disabled — nothing to replay to). Otherwise loops
/// forever, replaying on every `fallback_drain_notify` signal plus once at startup.
///
/// A plain future rather than a spawn, so the daemon can run it under its task
/// supervisor (`core::supervision::task`) and a panic mid-replay does not end
/// offline-event recovery for the rest of the process lifetime.
pub async fn run(state: Arc<AppState>) {
    let Some(cloud_state) = state.cloud_state.clone() else {
        return;
    };
    let notify = cloud_state.fallback_drain_notify.clone();
    // Startup replay: pick up entries left by a previous daemon run
    // (offline boot is the motivating case).
    replay_once(&state).await;
    loop {
        notify.notified().await;
        replay_once(&state).await;
    }
}

/// Single pass over `fallback.jsonl`. Reads from the persisted offset to
/// end of file, feeds each valid envelope through `process_envelope`, and
/// commits progress via the offset file. Stops on the first I/O error so a
/// transient filesystem issue doesn't retry-storm.
pub async fn replay_once(state: &Arc<AppState>) {
    // Emergency-mode early-return: when the cloud worker has flagged the
    // channel as saturated, every replay event would race live hooks for
    // capacity. Skip this pass entirely so the channel drains as fast as
    // the worker can post — the next `fallback_drain_notify` signal will retry.
    if let Some(cs) = state.cloud_state.as_ref() {
        if cs.is_emergency_mode() {
            tracing::debug!("fallback replay: skipping pass — cloud channel in emergency mode");
            return;
        }
    }

    // PLT-7 D-11 — a pass holds the switch's I/O gate from its first read to its
    // cursor commit, so pausing for an organization switch waits for it, and no
    // pass starts while uploads are paused.
    let switch = state.cloud_state.as_ref().map(|cs| cs.switch.clone());
    let _pass = match switch.as_ref() {
        Some(switch) => match switch.begin_io_pass().await {
            Some(pass) => Some(pass),
            None => {
                tracing::debug!(
                    "fallback replay: skipping pass — uploads paused for an organization switch"
                );
                return;
            }
        },
        None => None,
    };
    let pass_generation = switch.as_ref().map(|s| s.generation());
    // A cursor this pass computed is void once a switch applied after it
    // started: apply moved the cursor to the end, and this one would move it
    // back over the old organization's events.
    let switched_since_start =
        || matches!((switch.as_ref(), pass_generation), (Some(s), Some(g)) if s.is_stale(g));

    let log_dir = crate::config::openlatch_dir().join("logs");
    let path = log_dir.join(FALLBACK_FILENAME);
    let offset_path = log_dir.join(OFFSET_FILENAME);

    if !path.exists() {
        // Remove any stale offset file if fallback.jsonl was cleared
        // manually. Keeps the directory tidy.
        let _ = std::fs::remove_file(&offset_path);
        return;
    }

    // Enforce the fallback size cap BEFORE the replay pass so a long
    // outage can't poison a fresh restart by handing the worker tens of
    // thousands of stale events all at once. Drop-oldest is implemented
    // by advancing the persisted offset, not by rewriting the file —
    // cheaper, atomic, and reuses the existing replay machinery.
    let max_bytes = state.config.cloud.fallback_max_bytes;
    if max_bytes > 0 {
        enforce_fallback_size_cap(&path, &offset_path, max_bytes);
    }

    let start_offset = read_offset(&offset_path);
    let file = match std::fs::File::open(&path) {
        Ok(f) => f,
        Err(e) => {
            tracing::warn!(
                code = "OL-1205",
                error = %e,
                path = %path.display(),
                "fallback replay: could not open fallback.jsonl"
            );
            return;
        }
    };

    let total_len = file.metadata().map(|m| m.len()).unwrap_or(0);
    if start_offset >= total_len {
        // Nothing new since last replay. Clean up both files so a long-
        // idle daemon doesn't carry empty artifacts around forever.
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&offset_path);
        return;
    }

    let mut reader = BufReader::new(file);
    if reader.seek(SeekFrom::Start(start_offset)).is_err() {
        tracing::warn!(
            code = "OL-1205",
            path = %path.display(),
            "fallback replay: seek failed — discarding offset and restarting from 0"
        );
        let _ = std::fs::remove_file(&offset_path);
        return;
    }

    let switch_cutoff = read_switch_cutoff(&log_dir);
    let mut replayed: u64 = 0;
    let mut corrupt: u64 = 0;
    let mut consumed_offset = start_offset;
    let mut halted = false;

    let pending_hint = count_entries_from(&path, start_offset);
    tracing::info!(
        start_offset,
        total_bytes = total_len,
        pending = pending_hint,
        "fallback replay: starting"
    );
    crate::telemetry::capture_global(crate::telemetry::Event::fallback_replay_started(
        pending_hint,
    ));

    for line in reader.lines() {
        let Ok(raw) = line else {
            halted = true;
            break;
        };
        let line_len = raw.len() as u64 + 1; // +1 for newline
        if raw.trim().is_empty() {
            consumed_offset += line_len;
            continue;
        }

        let envelope: EventEnvelope = match serde_json::from_str(&raw) {
            Ok(e) => e,
            Err(e) => {
                tracing::debug!(
                    error = %e,
                    "fallback replay: discarding unparsable line"
                );
                corrupt += 1;
                consumed_offset += line_len;
                continue;
            }
        };

        // Captured before the last organization switch: the old organization's
        // event, never replayed under the new key.
        if switch_cutoff.is_some_and(|cutoff| envelope.time.timestamp_millis() < cutoff) {
            tracing::debug!(
                "fallback replay: dropping an entry captured before the organization switch"
            );
            consumed_offset += line_len;
            continue;
        }

        // If the cloud worker has exited (e.g. daemon shutting down) the
        // replay path's `send().await` would return `Err` and we'd advance
        // the offset past an event that never reached the channel. Detect
        // that here so the offset stays anchored on the unsent line and
        // the next daemon start picks up exactly where we left off.
        if let Some(tx) = state.cloud_tx.as_ref() {
            if tx.is_closed() {
                tracing::info!(
                    "fallback replay: cloud channel closed — halting, will resume from offset {consumed_offset} on next start"
                );
                halted = true;
                break;
            }
        }

        // Push through the same pipeline as a live hook so privacy
        // filtering and enrichment both fire. We intentionally drop the
        // returned verdict — the agent that produced this envelope is
        // long gone; we only care that the event reaches the cloud and
        // the audit log.
        let started = std::time::Instant::now();
        let _ = handlers::process_envelope_for_replay(state.clone(), envelope, started).await;
        replayed += 1;
        consumed_offset += line_len;
    }

    if switched_since_start() {
        tracing::debug!("fallback replay: an organization switch applied mid-pass; discarding this pass's cursor");
        return;
    }

    if halted {
        tracing::warn!(
            code = "OL-1205",
            replayed,
            corrupt,
            "fallback replay: halted on read error — remainder will retry next cycle"
        );
        write_offset(&offset_path, consumed_offset);
        return;
    }

    if consumed_offset >= total_len {
        // Clean drain — delete both files.
        if let Err(e) = std::fs::remove_file(&path) {
            tracing::warn!(
                code = "OL-1205",
                error = %e,
                "fallback replay: could not remove drained fallback.jsonl"
            );
        }
        let _ = std::fs::remove_file(&offset_path);
    } else {
        write_offset(&offset_path, consumed_offset);
    }

    if replayed > 0 || corrupt > 0 {
        let remaining = count_entries_from(&path, consumed_offset);
        tracing::info!(
            replayed,
            corrupt,
            remaining,
            consumed_bytes = consumed_offset - start_offset,
            "fallback replay: completed"
        );
        crate::telemetry::capture_global(crate::telemetry::Event::fallback_replay_completed(
            replayed, corrupt, remaining,
        ));
    }
}

/// Drop the unread tail of `logs/fallback.jsonl` for an organization switch
/// (PLT-7 C-8 apply step 1): the replay cursor moves to the end of the file, so
/// no event the hook captured for the old organization is ever replayed under
/// the new key. Returns how many entries were dropped.
///
/// Called by apply with uploads paused and the replay quiesced, so no pass is
/// reading the file. It first records the capture-time cutoff, so an entry the
/// hook appends later for the old organization is dropped too. When the cursor
/// cannot be persisted the file itself is removed — every byte before the
/// cursor was already replayed. `log_dir` is `<openlatch_dir>/logs`.
///
/// # Errors
///
/// When neither the skip nor the removal can be verified, or the cutoff cannot
/// be written: apply must then stay paused, since the tail would replay under
/// the new key.
pub(crate) fn skip_unread_tail_in(log_dir: &Path) -> Result<u64, String> {
    write_switch_cutoff(log_dir, chrono::Utc::now().timestamp_millis())?;
    let path = log_dir.join(FALLBACK_FILENAME);
    let offset_path = log_dir.join(OFFSET_FILENAME);
    let end = match std::fs::metadata(&path) {
        Ok(metadata) => metadata.len(),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(e) => return Err(format!("cannot read {}: {e}", path.display())),
    };
    let start = read_offset(&offset_path);
    if start >= end {
        return Ok(0);
    }
    let dropped = count_entries_from(&path, start);
    write_offset(&offset_path, end);
    if read_offset(&offset_path) != end {
        match std::fs::remove_file(&path) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                return Err(format!(
                    "cannot drop the unread tail of {}: {e}",
                    path.display()
                ))
            }
        }
    }
    Ok(dropped)
}

fn write_switch_cutoff(log_dir: &Path, cutoff_ms: i64) -> Result<(), String> {
    let path = log_dir.join(SWITCH_CUTOFF_FILENAME);
    std::fs::create_dir_all(log_dir)
        .and_then(|()| crate::core::fs_secure::write_owner_only(&path, &cutoff_ms.to_string()))
        .map_err(|e| format!("cannot record the switch cutoff at {}: {e}", path.display()))
}

fn read_switch_cutoff(log_dir: &Path) -> Option<i64> {
    std::fs::read_to_string(log_dir.join(SWITCH_CUTOFF_FILENAME))
        .ok()
        .and_then(|s| s.trim().parse().ok())
}

/// How many entries of `logs/fallback.jsonl` are not replayed yet.
pub(crate) fn unread_entries_in(log_dir: &Path) -> u64 {
    let path = log_dir.join(FALLBACK_FILENAME);
    let start = read_offset(&log_dir.join(OFFSET_FILENAME));
    match std::fs::metadata(&path) {
        Ok(m) if m.len() > start => count_entries_from(&path, start),
        _ => 0,
    }
}

/// Cap the unread tail of `fallback.jsonl` to `max_bytes` by advancing
/// the persisted offset past oldest entries — same algorithm the outbox
/// uses, factored into `core::cloud::offset::advance_past_oldest`.
fn enforce_fallback_size_cap(path: &Path, offset_path: &Path, max_bytes: u64) {
    let Ok(metadata) = std::fs::metadata(path) else {
        return;
    };
    let total_len = metadata.len();
    let start_offset = read_offset(offset_path);
    if start_offset >= total_len {
        return;
    }
    let unread = total_len - start_offset;
    if unread <= max_bytes {
        return;
    }
    let excess = unread - max_bytes;
    let Some(AdvanceStats {
        new_offset,
        dropped,
    }) = advance_past_oldest(path, start_offset, excess)
    else {
        return;
    };
    tracing::warn!(
        code = "OL-1205",
        dropped,
        size_before = unread,
        max_bytes,
        new_offset,
        "fallback overflow: advancing offset past oldest entries to stay under size cap"
    );
    crate::telemetry::capture_global(crate::telemetry::Event::fallback_overflow(
        dropped, unread, max_bytes,
    ));
    write_offset(offset_path, new_offset);
}

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

    fn write_line(file: &mut std::fs::File, line: &str) -> u64 {
        use std::io::Write;
        let bytes = format!("{line}\n");
        file.write_all(bytes.as_bytes()).unwrap();
        bytes.len() as u64
    }

    #[test]
    fn enforce_size_cap_is_noop_when_unread_within_budget() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        let mut f = std::fs::File::create(&path).unwrap();
        for i in 0..10 {
            write_line(&mut f, &format!("{{\"id\":\"evt_{i}\"}}"));
        }
        drop(f);

        // Far above file size — nothing to evict.
        enforce_fallback_size_cap(&path, &offset_path, 10_000_000);

        // No offset file should be written.
        assert!(!offset_path.exists(), "no eviction → no offset file");
    }

    #[test]
    fn enforce_size_cap_advances_offset_past_oldest_when_over_budget() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        let mut f = std::fs::File::create(&path).unwrap();
        let mut sizes: Vec<u64> = Vec::new();
        for i in 0..20 {
            sizes.push(write_line(&mut f, &format!("{{\"id\":\"evt_{i:04}\"}}")));
        }
        drop(f);
        let total: u64 = sizes.iter().sum();
        // Cap at ~half the file — at least one line must be evicted.
        let cap = total / 2;

        enforce_fallback_size_cap(&path, &offset_path, cap);

        let new_offset = read_offset(&offset_path);
        assert!(new_offset > 0, "offset must advance past oldest entries");
        let unread = total - new_offset;
        assert!(
            unread <= cap,
            "unread tail ({unread}) must be ≤ cap ({cap}) after eviction"
        );
    }

    #[test]
    fn fallback_replay_skip_unread_tail_moves_the_cursor_to_the_end() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        let mut f = std::fs::File::create(&path).unwrap();
        let first = write_line(&mut f, "{\"id\":\"evt_read\"}");
        let mut total = first;
        for i in 0..3 {
            total += write_line(&mut f, &format!("{{\"id\":\"evt_unread_{i}\"}}"));
        }
        drop(f);
        write_offset(&offset_path, first);

        assert_eq!(unread_entries_in(tmp.path()), 3);
        assert_eq!(skip_unread_tail_in(tmp.path()), Ok(3));
        assert_eq!(unread_entries_in(tmp.path()), 0);
        assert_eq!(read_offset(&offset_path), total);
        // Nothing left to drop the second time.
        assert_eq!(skip_unread_tail_in(tmp.path()), Ok(0));
        // The cutoff is recorded, so a late append from before the switch is caught.
        assert!(read_switch_cutoff(tmp.path()).is_some());
    }

    #[test]
    fn fallback_replay_skip_unread_tail_without_a_file_drops_nothing() {
        let tmp = TempDir::new().unwrap();
        assert_eq!(skip_unread_tail_in(tmp.path()), Ok(0));
        assert!(!tmp.path().join(OFFSET_FILENAME).exists());
    }

    #[test]
    fn enforce_size_cap_handles_missing_file() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        // Must not panic when the file doesn't exist.
        enforce_fallback_size_cap(&path, &offset_path, 1000);
        assert!(!offset_path.exists());
    }

    #[test]
    fn enforce_size_cap_zero_disables_cap() {
        // The caller is responsible for the `max_bytes > 0` gate; this test
        // exists to assert that even if the function is called with 0, the
        // saturating arithmetic doesn't underflow. With max_bytes = 0,
        // `unread > 0` evicts everything — verify it does so without
        // panicking and lands on a sane offset.
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        let mut f = std::fs::File::create(&path).unwrap();
        write_line(&mut f, "{\"id\":\"evt_0\"}");
        drop(f);
        enforce_fallback_size_cap(&path, &offset_path, 0);
        // Either nothing evicted or the entire file evicted — both are
        // acceptable; what matters is no panic / underflow.
        let _ = read_offset(&offset_path);
    }
}

/// The replay pass end to end: `replay_once` over a real `logs/fallback.jsonl`,
/// observed only through what leaves it — the events handed to the cloud
/// channel and the files left on disk. Runs under `OPENLATCH_DIR_ENV_LOCK`
/// against a temporary state directory.
#[cfg(test)]
mod replay_pass {
    use super::*;
    use crate::cloud::CloudEvent;
    use tempfile::TempDir;
    use tokio::sync::mpsc;

    /// Points `OPENLATCH_DIR` at a temporary directory for one test and restores it.
    struct StateDir {
        dir: TempDir,
        previous: Option<String>,
        _lock: std::sync::MutexGuard<'static, ()>,
    }

    impl StateDir {
        fn new() -> Self {
            let lock = crate::config::OPENLATCH_DIR_ENV_LOCK
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let dir = TempDir::new().unwrap();
            let previous = std::env::var("OPENLATCH_DIR").ok();
            std::env::set_var("OPENLATCH_DIR", dir.path());
            std::fs::create_dir_all(dir.path().join("logs")).unwrap();
            Self {
                dir,
                previous,
                _lock: lock,
            }
        }

        fn logs(&self) -> std::path::PathBuf {
            self.dir.path().join("logs")
        }

        fn fallback(&self) -> std::path::PathBuf {
            self.logs().join(FALLBACK_FILENAME)
        }

        fn offset(&self) -> std::path::PathBuf {
            self.logs().join(OFFSET_FILENAME)
        }

        fn write_fallback(&self, lines: &[String]) {
            let body: String = lines.iter().map(|l| format!("{l}\n")).collect();
            std::fs::write(self.fallback(), body).unwrap();
        }
    }

    impl Drop for StateDir {
        fn drop(&mut self) {
            match &self.previous {
                Some(v) => std::env::set_var("OPENLATCH_DIR", v),
                None => std::env::remove_var("OPENLATCH_DIR"),
            }
        }
    }

    fn envelope_line(id: &str, time: &str, command: &str) -> String {
        serde_json::json!({
            "specversion": "1.0",
            "id": id,
            "source": "claude-code",
            "type": "pre_tool_use",
            "time": time,
            "subject": "sess_replay",
            "data": {
                "session_id": "sess_replay",
                "tool_name": "Bash",
                "tool_input": {"command": command}
            }
        })
        .to_string()
    }

    fn line(id: &str) -> String {
        // The command is unique per entry: the daemon's dedup would otherwise
        // collapse identical hook calls and hide what replay itself does.
        envelope_line(id, "2026-09-30T00:00:00Z", &format!("ls {id}"))
    }

    /// An `AppState` forwarding to the returned receiver; `max_bytes` is the
    /// fallback size cap (0 disables it).
    fn state_with_channel(max_bytes: u64) -> (Arc<AppState>, mpsc::Receiver<CloudEvent>) {
        let mut app =
            Arc::into_inner(crate::daemon::admin::tests::test_app_state(None, None)).unwrap();
        let mut config = crate::core::config::Config::defaults();
        config.agent_id = Some("agt_replay_test".to_string());
        config.cloud.fallback_max_bytes = max_bytes;
        app.config = Arc::new(config);
        let (tx, rx) = mpsc::channel(64);
        app.cloud_tx = Some(tx);
        (Arc::new(app), rx)
    }

    fn drained_ids(rx: &mut mpsc::Receiver<CloudEvent>) -> Vec<String> {
        let mut ids = Vec::new();
        while let Ok(event) = rx.try_recv() {
            ids.push(event.envelope["id"].as_str().unwrap().to_string());
        }
        ids
    }

    #[tokio::test]
    async fn a_clean_pass_forwards_every_entry_in_order_then_removes_both_files() {
        let dir = StateDir::new();
        dir.write_fallback(&[line("evt-1"), line("evt-2"), line("evt-3")]);
        let (state, mut rx) = state_with_channel(0);

        replay_once(&state).await;

        assert_eq!(drained_ids(&mut rx), ["evt-1", "evt-2", "evt-3"]);
        assert!(!dir.fallback().exists(), "a drained log is deleted");
        assert!(!dir.offset().exists(), "a drained log leaves no cursor");
    }

    #[tokio::test]
    async fn a_pass_resumes_at_the_persisted_offset_and_never_resends_earlier_entries() {
        let dir = StateDir::new();
        let first = line("evt-old");
        dir.write_fallback(&[first.clone(), line("evt-new")]);
        write_offset(&dir.offset(), first.len() as u64 + 1);
        let (state, mut rx) = state_with_channel(0);

        replay_once(&state).await;

        assert_eq!(drained_ids(&mut rx), ["evt-new"]);
        assert!(!dir.fallback().exists());
        assert!(!dir.offset().exists());
    }

    #[tokio::test]
    async fn a_cursor_at_the_end_of_the_log_sends_nothing_and_cleans_up() {
        let dir = StateDir::new();
        let only = line("evt-done");
        dir.write_fallback(std::slice::from_ref(&only));
        write_offset(&dir.offset(), only.len() as u64 + 1);
        let (state, mut rx) = state_with_channel(0);

        replay_once(&state).await;

        assert!(drained_ids(&mut rx).is_empty());
        assert!(!dir.fallback().exists());
        assert!(!dir.offset().exists());
    }

    #[tokio::test]
    async fn a_stale_cursor_is_removed_when_the_log_is_gone() {
        let dir = StateDir::new();
        write_offset(&dir.offset(), 42);
        let (state, mut rx) = state_with_channel(0);

        replay_once(&state).await;

        assert!(drained_ids(&mut rx).is_empty());
        assert!(!dir.offset().exists());
    }

    #[tokio::test]
    async fn corrupt_and_blank_lines_are_skipped_without_blocking_the_valid_ones() {
        let dir = StateDir::new();
        dir.write_fallback(&[
            "{not json".to_string(),
            String::new(),
            "   ".to_string(),
            line("evt-ok"),
            "{\"id\":\"missing-required-fields\"}".to_string(),
        ]);
        let (state, mut rx) = state_with_channel(0);

        replay_once(&state).await;

        assert_eq!(drained_ids(&mut rx), ["evt-ok"]);
        assert!(
            !dir.fallback().exists(),
            "skipped lines still count as consumed, so the log drains"
        );
    }

    #[tokio::test]
    async fn entries_captured_before_the_switch_cutoff_are_never_replayed() {
        let dir = StateDir::new();
        let cutoff = chrono::DateTime::parse_from_rfc3339("2026-09-30T12:00:00Z")
            .unwrap()
            .timestamp_millis();
        write_switch_cutoff(&dir.logs(), cutoff).unwrap();
        dir.write_fallback(&[
            envelope_line("evt-before", "2026-09-30T11:59:59.999Z", "ls before"),
            envelope_line("evt-at-cutoff", "2026-09-30T12:00:00Z", "ls at"),
            envelope_line("evt-after", "2026-09-30T12:00:01Z", "ls after"),
        ]);
        let (state, mut rx) = state_with_channel(0);

        replay_once(&state).await;

        assert_eq!(
            drained_ids(&mut rx),
            ["evt-at-cutoff", "evt-after"],
            "strictly older than the cutoff is dropped; the cutoff instant itself is kept"
        );
        assert!(!dir.fallback().exists());
    }

    #[tokio::test]
    async fn a_closed_cloud_channel_halts_the_pass_and_keeps_the_log_for_the_next_start() {
        let dir = StateDir::new();
        dir.write_fallback(&[line("evt-1"), line("evt-2")]);
        let (state, rx) = state_with_channel(0);
        drop(rx);

        replay_once(&state).await;

        assert!(dir.fallback().exists(), "unsent entries must survive");
        assert_eq!(
            read_offset(&dir.offset()),
            0,
            "the cursor stays anchored on the first unsent entry"
        );
    }

    #[tokio::test]
    async fn the_size_cap_drops_the_oldest_entries_before_the_pass() {
        let dir = StateDir::new();
        let lines: Vec<String> = (0..4).map(|i| line(&format!("evt-{i}"))).collect();
        let one = lines[0].len() as u64 + 1;
        dir.write_fallback(&lines);
        // Room for exactly the last two entries.
        let (state, mut rx) = state_with_channel(one * 2);

        replay_once(&state).await;

        assert_eq!(drained_ids(&mut rx), ["evt-2", "evt-3"]);
        assert!(!dir.fallback().exists());
    }

    #[tokio::test]
    async fn a_zero_size_cap_disables_eviction() {
        let dir = StateDir::new();
        let lines: Vec<String> = (0..4).map(|i| line(&format!("evt-{i}"))).collect();
        dir.write_fallback(&lines);
        let (state, mut rx) = state_with_channel(0);

        replay_once(&state).await;

        assert_eq!(drained_ids(&mut rx).len(), 4);
    }

    #[tokio::test]
    async fn replayed_content_is_privacy_filtered_and_carries_no_current_verdict_stamp() {
        let dir = StateDir::new();
        let secret = "AKIAABCDEFGHIJKLMNOP";
        dir.write_fallback(&[envelope_line(
            "evt-secret",
            "2026-09-30T00:00:00Z",
            &format!("echo {secret}"),
        )]);
        let (state, mut rx) = state_with_channel(0);

        replay_once(&state).await;

        let event = rx.try_recv().expect("the entry is forwarded");
        let body = event.envelope.to_string();
        assert!(
            !body.contains(secret),
            "unredacted secret reached the cloud channel: {body}"
        );
        assert_eq!(event.agent_id, "agt_replay_test");
        let top = event.envelope.as_object().unwrap();
        assert!(
            top.keys()
                .all(|k| !k.starts_with("olverdict") && !k.starts_with("olpolicy")),
            "a replayed event must not be stamped with today's verdict: {top:?}"
        );
    }
}