agentmux_launcher/

event_log.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0
//
// Phase D.2 — event log: in-memory ring buffer of recent reducer
// events, used by D.3's replay protocol; plus an optional disk
// persistence stream at `<data-dir>/launcher-events.log` for
// crash forensics.
//
// Two roles, kept clean:
//
// 1. **In-memory ring** is the source of truth for replay during
//    the launcher's lifetime. `GetEvents { since: u64 }` reads
//    from it. Bounded by `MAX_RING_EVENTS` to stop unbounded
//    memory growth in long sessions; oldest events evict first.
//    A subscriber that fell behind further than the ring's
//    coverage gets an `EventList` of whatever's still in the ring
//    + a flag indicating they may have missed some — it's their
//    job to recover by treating subsequent events as authoritative.
//
// 2. **Disk file** is purely forensic. Append-only JSON-lines.
//    Survives launcher crash so an operator can post-mortem
//    "what was the launcher doing right before it died."
//    Fire-and-forget: write failures log a warning but never
//    block the in-memory path. Rotated when it exceeds
//    `MAX_DISK_BYTES` (renames current to `.old`, starts fresh).
//
// Why both: in-memory satisfies D.3's resync needs at zero I/O
// cost in the happy path. Disk satisfies the "what happened
// before the crash" debugging story without complicating the
// replay reader (which would otherwise need fallback-to-disk
// logic when the ring's coverage doesn't reach back far enough).
//
// Phase D.3 adds the wire protocol that consumes this module.

use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::Mutex;

use agentmux_common::ipc::Event;
use tokio::io::AsyncWriteExt;

/// Cap on in-memory ring size. 4096 events is comfortable for
/// realistic resync windows (~minutes of activity at typical
/// reducer event rates: 10–50 events per user action). Tunable
/// upward if forensics in long sessions show truncation; downward
/// would only be useful if we measured noticeable memory pressure
/// from this (we don't — Event is small and there are 4096 of them).
const MAX_RING_EVENTS: usize = 4096;

/// Cap on disk file size before rotation. 8 MiB ≈ 4–8K events
/// depending on event variant. Two-file rotation: when current
/// exceeds this, rename to `.old` (overwriting any prior `.old`)
/// and start fresh. Total worst-case footprint: 2 × MAX_DISK_BYTES.
const MAX_DISK_BYTES: u64 = 8 * 1024 * 1024;

/// Append-only ring + optional disk persistence.
///
/// Cloneable via `Arc<EventLog>` from the IPC server context;
/// `append` and `events_since` are cheap (Mutex held for
/// microseconds — Vec<Event> push / scan, no I/O on the
/// in-memory path).
///
/// Disk writes happen on a dedicated tokio task that subscribes
/// to the broadcast bus separately from the in-memory append.
/// This means the in-memory ring is updated synchronously from
/// the IPC server's dispatch path; disk persistence runs at its
/// own pace and may lag.
#[derive(Debug)]
pub struct EventLog {
    ring: Mutex<VecDeque<Event>>,
    disk_path: Option<PathBuf>,
}

impl EventLog {
    /// Construct an event log. `disk_path = None` disables disk
    /// persistence (used in tests where filesystem state is
    /// inconvenient). The on-disk file is created on first append;
    /// no upfront I/O.
    pub fn new(disk_path: Option<PathBuf>) -> Self {
        Self {
            ring: Mutex::new(VecDeque::with_capacity(MAX_RING_EVENTS)),
            disk_path,
        }
    }

    /// Append an event to the in-memory ring. Evicts the oldest
    /// entry when the ring is at capacity. Synchronous,
    /// O(1)-amortized.
    pub fn append(&self, event: Event) {
        let mut ring = self.ring.lock().expect("event-log ring mutex poisoned");
        if ring.len() == MAX_RING_EVENTS {
            ring.pop_front();
        }
        ring.push_back(event);
    }

    /// Snapshot of all events currently in the ring with version >
    /// `since`. Returned in insertion order (oldest first), so the
    /// caller applies them sequentially.
    ///
    /// Phase D.3 — used by `Command::GetEvents { since }` to
    /// produce the replay slice. The snapshot is taken at-call-time;
    /// events arriving after this returns are NOT included (the
    /// subscriber sees them on the live broadcast stream).
    pub fn events_since(&self, since: u64) -> Vec<Event> {
        let ring = self.ring.lock().expect("event-log ring mutex poisoned");
        ring.iter()
            .filter(|e| event_version(e) > since)
            .cloned()
            .collect()
    }

    /// True if the requested `since` version is older than the
    /// oldest event in the ring (i.e. the subscriber missed events
    /// that have already been evicted). Caller should treat the
    /// resulting `events_since` slice as best-effort and may need
    /// to re-fetch a snapshot to recover canonical state.
    pub fn replay_truncated(&self, since: u64) -> bool {
        let ring = self.ring.lock().expect("event-log ring mutex poisoned");
        match ring.front() {
            // Phase E.1a (codex P2 #608) — saturating_add guards
            // against `since: u64::MAX` overflow. Wire input is
            // externally reachable; debug builds would panic, release
            // would wrap. Saturating means "since == u64::MAX" trivially
            // returns false (oldest can't exceed u64::MAX), which is
            // the correct semantic — there's no possible event newer
            // than that, so there can't be a gap.
            Some(oldest) => event_version(oldest) > since.saturating_add(1),
            None => false,
        }
    }

    /// Disk path for the writer task to flush to. None when disk
    /// persistence is disabled.
    pub fn disk_path(&self) -> Option<&PathBuf> {
        self.disk_path.as_ref()
    }
}

/// Background task: write events to the disk file as they arrive
/// on the broadcast bus. Rotates when the file exceeds
/// `MAX_DISK_BYTES`.
///
/// Spawned once per launcher run from main.rs. Holds a tokio
/// broadcast receiver and the EventLog's disk path. Failures
/// log a warning and drop the event from the disk stream
/// (in-memory ring is unaffected — disk is forensics-only).
pub async fn run_disk_writer(
    log: std::sync::Arc<EventLog>,
    mut events_rx: tokio::sync::broadcast::Receiver<Event>,
) {
    let path = match log.disk_path() {
        Some(p) => p.clone(),
        None => return, // disk persistence disabled — exit task
    };
    let rotated_path = path.with_extension("log.old");

    let mut file = match open_for_append(&path).await {
        Ok(f) => f,
        Err(e) => {
            crate::log(&format!(
                "[event-log] cannot open {} for append: {} — disk persistence disabled",
                path.display(),
                e
            ));
            return;
        }
    };
    let mut bytes_written = file.metadata().await.map(|m| m.len()).unwrap_or(0);

    loop {
        match events_rx.recv().await {
            Ok(event) => {
                let mut buf = match serde_json::to_vec(&event) {
                    Ok(b) => b,
                    Err(e) => {
                        crate::log(&format!("[event-log] serialize failed: {}", e));
                        continue;
                    }
                };
                buf.push(b'\n');
                if bytes_written + buf.len() as u64 > MAX_DISK_BYTES {
                    // Rotate: drop the writer, rename current →
                    // .old (overwriting), reopen fresh. Failures
                    // here are non-fatal; we log and keep writing
                    // to the existing file (it'll just exceed cap).
                    drop(file);
                    if let Err(e) = tokio::fs::rename(&path, &rotated_path).await {
                        crate::log(&format!("[event-log] rotation rename failed: {} — continuing without rotation", e));
                    }
                    file = match open_for_append(&path).await {
                        Ok(f) => f,
                        Err(e) => {
                            crate::log(&format!(
                                "[event-log] post-rotation open failed: {} — disk persistence stopping",
                                e
                            ));
                            return;
                        }
                    };
                    bytes_written = 0;
                }
                if let Err(e) = file.write_all(&buf).await {
                    crate::log(&format!("[event-log] write failed: {} — dropping event from disk stream", e));
                    continue;
                }
                // Phase E.1a — durable: fsync per append so events
                // written before a crash survive it. Required by
                // Phase E §6.4 for srv's bootstrap-replay correctness.
                // Latency cost: ~ms per event vs microseconds without
                // sync. Acceptable for our event volume (~10–50 per
                // user action). Trade-off considered: batched fsync
                // (group commit). Skipping for E.1a — premature
                // optimization; revisit if profiling shows a hot path.
                if let Err(e) = file.sync_data().await {
                    crate::log(&format!("[event-log] sync_data failed: {} — event written but not fsynced", e));
                }
                bytes_written += buf.len() as u64;
            }
            Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                crate::log(&format!("[event-log] disk writer lagged, missed {} events", n));
            }
            Err(tokio::sync::broadcast::error::RecvError::Closed) => {
                let _ = file.flush().await;
                return;
            }
        }
    }
}

async fn open_for_append(path: &std::path::Path) -> std::io::Result<tokio::fs::File> {
    tokio::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .await
}

/// Extract the `version` field from any Event variant. Mirrors
/// reducer.rs::extract_version (which is test-only; the prod code
/// path needs its own copy here). When a new variant is added that
/// carries a version, add it here too — the unreachable arm catches
/// future-variant compile errors.
fn event_version(e: &Event) -> u64 {
    match e {
        Event::ProcessSpawned { version, .. }
        | Event::ProcessExited { version, .. }
        | Event::LifecyclePhaseChanged { version, .. }
        | Event::Registered { version, .. }
        | Event::Pong { version, .. }
        | Event::WindowOpened { version, .. }
        | Event::WindowClosed { version, .. }
        | Event::PoolWindowAdded { version, .. }
        | Event::PoolWindowRemoved { version, .. }
        | Event::PoolWindowPromoted { version, .. }
        | Event::PanesReaped { version, .. }
        | Event::PoolDrained { version, .. }
        | Event::PoolNotLast { version, .. }
        | Event::WindowInstanceAssigned { version, .. }
        | Event::WindowInstanceReleased { version, .. }
        | Event::BackendWindowIdRegistered { version, .. }
        | Event::BackendWindowIdUnregistered { version, .. }
        | Event::DriftDetected { version, .. }
        | Event::HwndDriftDetected { version, .. }
        | Event::CorrectiveWindowMove { version, .. }
        | Event::HostShouldQuit { version, .. }
        | Event::Snapshot { version, .. }
        | Event::EventList { version, .. }
        | Event::SagaStarted { version, .. }
        | Event::SagaCompleted { version, .. }
        | Event::SagaFailed { version, .. }
        | Event::SrvSnapshot { version, .. }
        | Event::WorkspaceCreated { version, .. }
        | Event::WorkspaceDeleted { version, .. }
        | Event::TabCreated { version, .. }
        | Event::TabDeleted { version, .. }
        | Event::ActiveTabChanged { version, .. }
        | Event::TabReordered { version, .. }
        | Event::SrvWindowOpened { version, .. }
        | Event::SrvWindowClosed { version, .. }
        | Event::SrvWindowWorkspaceChanged { version, .. }
        | Event::TabsReorderedBulk { version, .. }
        | Event::WorkspaceRenamed { version, .. }
        | Event::TabRenamed { version, .. }
        | Event::WorkspaceMetaUpdated { version, .. }
        | Event::TabMetaUpdated { version, .. }
        | Event::BlockMetaUpdated { version, .. }
        | Event::TabMoved { version, .. }
        | Event::BlockMoved { version, .. }
        | Event::BlockCreated { version, .. }
        | Event::BlockDeleted { version, .. }
        | Event::FocusedNodeChanged { version, .. }
        | Event::MagnifiedNodeChanged { version, .. }
        | Event::SagaActionFailed { version, .. }
        | Event::Error { version, .. }
        // Phase E.4.B — layout tree events.
        | Event::LayoutNodeInserted { version, .. }
        | Event::LayoutNodeInsertedAtIndex { version, .. }
        | Event::LayoutNodeDeleted { version, .. }
        | Event::LayoutNodeMoved { version, .. }
        | Event::LayoutNodesSwapped { version, .. }
        | Event::LayoutNodesResized { version, .. }
        | Event::LayoutNodeReplaced { version, .. }
        | Event::LayoutSplitHorizontalApplied { version, .. }
        | Event::LayoutSplitVerticalApplied { version, .. }
        | Event::LayoutCleared { version, .. }
        | Event::LayoutTreeReplaced { version, .. }
        | Event::WindowMetaUpdated { version, .. } => *version,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use agentmux_common::ipc::{ClientKind, LifecyclePhase};

    fn lifecycle_event(v: u64) -> Event {
        Event::LifecyclePhaseChanged {
            from: LifecyclePhase::Starting,
            to: LifecyclePhase::Running,
            version: v,
        }
    }

    fn registered_event(v: u64) -> Event {
        Event::Registered {
            client_id: 1,
            launcher_pid: 1,
            launcher_version: "test".into(),
            version: v,
        }
    }

    fn process_spawned_event(v: u64) -> Event {
        Event::ProcessSpawned {
            pid: 100,
            kind: ClientKind::Host,
            client_version: "0.0.0".into(),
            version: v,
        }
    }

    #[test]
    fn append_grows_ring_until_cap_then_evicts_oldest() {
        let log = EventLog::new(None);
        for v in 1..=(MAX_RING_EVENTS as u64 + 5) {
            log.append(lifecycle_event(v));
        }
        let ring = log.ring.lock().unwrap();
        assert_eq!(ring.len(), MAX_RING_EVENTS);
        // Oldest 5 should have been evicted; first should be v=6.
        assert_eq!(event_version(ring.front().unwrap()), 6);
        // Newest is v = MAX_RING_EVENTS + 5.
        assert_eq!(
            event_version(ring.back().unwrap()),
            MAX_RING_EVENTS as u64 + 5
        );
    }

    #[test]
    fn events_since_returns_only_versions_strictly_greater() {
        let log = EventLog::new(None);
        for v in [1u64, 2, 3, 4, 5] {
            log.append(lifecycle_event(v));
        }
        let replay = log.events_since(2);
        assert_eq!(replay.len(), 3, "expected v=3,4,5; got {:?}", replay);
        assert_eq!(event_version(&replay[0]), 3);
        assert_eq!(event_version(&replay[2]), 5);
    }

    #[test]
    fn events_since_zero_returns_all() {
        let log = EventLog::new(None);
        for v in [1u64, 2, 3] {
            log.append(registered_event(v));
        }
        let replay = log.events_since(0);
        assert_eq!(replay.len(), 3);
    }

    #[test]
    fn events_since_at_or_above_max_returns_empty() {
        let log = EventLog::new(None);
        log.append(process_spawned_event(5));
        log.append(process_spawned_event(6));
        let replay = log.events_since(10);
        assert!(replay.is_empty());
    }

    #[test]
    fn replay_truncated_detects_missed_events() {
        let log = EventLog::new(None);
        // Fill past capacity to force eviction.
        for v in 1..=(MAX_RING_EVENTS as u64 + 100) {
            log.append(lifecycle_event(v));
        }
        // Subscriber asks for events since v=5; ring's oldest is now v=101.
        // The subscriber missed v=6..=100 — the replay slice covers v=101..,
        // and `replay_truncated(5)` reports the gap.
        assert!(log.replay_truncated(5));
        // Subscriber asking from a version newer than ring's oldest sees
        // no truncation.
        let oldest = MAX_RING_EVENTS as u64 + 100 - MAX_RING_EVENTS as u64 + 1;
        assert!(!log.replay_truncated(oldest));
    }

    #[test]
    fn replay_truncated_on_empty_log_is_false() {
        let log = EventLog::new(None);
        assert!(!log.replay_truncated(0));
        assert!(!log.replay_truncated(100));
    }
}