magi_code/sessions/

manager.rs

use super::metadata::{SessionMetadataRecord, SessionMetadataSummary};
use super::read::validate_session_id;
use super::store::{prepare_session_root, primary_path};
use crate::persistence::CrossProcessFileLock;
use std::{
    collections::HashMap,
    fmt,
    path::{Path, PathBuf},
    sync::{Arc, Mutex, OnceLock, Weak},
};
use uuid::Uuid;

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SessionInternalDiagnostic {
    pub(crate) session_id: Option<String>,
    pub(crate) message: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SessionListReport {
    pub(crate) summaries: Vec<SessionMetadataSummary>,
    pub(crate) diagnostics: Vec<SessionInternalDiagnostic>,
}

#[derive(Debug, Clone)]
pub struct SessionManager {
    pub(in crate::sessions) root: PathBuf,
}

impl SessionManager {
    #[must_use]
    pub fn new(root: PathBuf) -> Self {
        Self { root }
    }

    pub(crate) fn create(&self) -> anyhow::Result<Session> {
        prepare_session_root(&self.root)?;
        let id = Uuid::new_v4().to_string();
        let path = self.path_for_valid_id(&id)?;
        // Do not create JSONL until first event append; avoids orphan session files.
        Ok(Session::new(id, path))
    }

    pub fn open(&self, id: impl Into<String>) -> anyhow::Result<Session> {
        let id = validate_session_id(id.into())?;
        Ok(Session::new(id.clone(), self.path_for_valid_id(&id)?))
    }

    pub(crate) fn open_existing(&self, id: impl Into<String>) -> anyhow::Result<Session> {
        let session = self.open(id)?;
        super::store::open_existing_primary(&self.root, &session.id)?
            .ok_or_else(|| anyhow::anyhow!("session JSONL is missing"))?;
        Ok(session)
    }

    #[cfg(test)]
    pub(crate) fn list(&self) -> anyhow::Result<Vec<Session>> {
        Ok(self
            .list_metadata_summaries()?
            .into_iter()
            .map(|summary| summary.session)
            .collect())
    }

    pub(crate) fn most_recent(&self) -> anyhow::Result<Option<Session>> {
        let report = self.list_metadata_report()?;
        if !report.diagnostics.is_empty() {
            anyhow::bail!("session discovery found unreadable or unsafe history");
        }
        Ok(report
            .summaries
            .into_iter()
            .last()
            .map(|summary| summary.session))
    }

    pub(crate) fn path_for_valid_id(&self, id: &str) -> anyhow::Result<PathBuf> {
        primary_path(&self.root, id)
    }
}

pub struct Session {
    pub(in crate::sessions) id: String,
    pub(in crate::sessions) path: PathBuf,
    pub(in crate::sessions) metadata_cache: Arc<Mutex<Option<SessionMetadataRecord>>>,
    pub(crate) maintenance: crate::progressive_session_maintenance::ObserverSlot,
    replay_state: Arc<SessionReplayState>,
    active_lease: Arc<Mutex<Option<Arc<ActiveLeaseSlot>>>>,
    // Shared only by clones of this admitted standalone owner, never by independent opens.
    standalone_writer: Option<Arc<CrossProcessFileLock>>,
}

struct SessionReplayState {
    generation: std::sync::atomic::AtomicU64,
}

type ActiveLeaseSlot = Mutex<Option<Arc<CrossProcessFileLock>>>;

static ACTIVE_SESSION_LEASES: OnceLock<Mutex<HashMap<PathBuf, Weak<ActiveLeaseSlot>>>> =
    OnceLock::new();
static SESSION_REPLAY_GENERATIONS: OnceLock<Mutex<HashMap<PathBuf, Weak<SessionReplayState>>>> =
    OnceLock::new();

impl fmt::Debug for Session {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("Session")
            .field("id", &self.id)
            .field("path", &self.path)
            .finish()
    }
}

impl Clone for Session {
    fn clone(&self) -> Self {
        Self {
            id: self.id.clone(),
            path: self.path.clone(),
            metadata_cache: Arc::clone(&self.metadata_cache),
            maintenance: Arc::clone(&self.maintenance),
            replay_state: Arc::clone(&self.replay_state),
            active_lease: Arc::clone(&self.active_lease),
            standalone_writer: self.standalone_writer.clone(),
        }
    }
}

impl PartialEq for Session {
    fn eq(&self, other: &Self) -> bool {
        self.id == other.id && self.path == other.path
    }
}

impl Eq for Session {}

#[derive(Debug)]
pub(crate) struct SessionWriterBusy;
impl std::fmt::Display for SessionWriterBusy {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("session is busy: another process owns the writer")
    }
}
impl std::error::Error for SessionWriterBusy {}
impl Session {
    pub(in crate::sessions) fn new(id: String, path: PathBuf) -> Self {
        let replay_state = replay_state_for_path(&path);
        Self {
            id,
            path,
            metadata_cache: Arc::new(Mutex::new(None)),
            maintenance: Arc::new(Mutex::new(None)),
            replay_state,
            active_lease: Arc::new(Mutex::new(None)),
            standalone_writer: None,
        }
    }

    pub(crate) fn admit_standalone_writer(mut self) -> anyhow::Result<Self> {
        if self.standalone_writer.is_none() {
            let writer = self
                .try_frontend_writer()?
                .ok_or_else(|| anyhow::Error::new(SessionWriterBusy))?;
            self.standalone_writer = Some(Arc::new(writer));
        }
        self.activate()
    }

    pub(crate) fn activate(self) -> anyhow::Result<Self> {
        self.ensure_active_lease()?;
        Ok(self)
    }

    pub(in crate::sessions) fn ensure_active_lease(&self) -> anyhow::Result<()> {
        let slot = {
            let mut local_slot = self
                .active_lease
                .lock()
                .map_err(|_| anyhow::anyhow!("active session lease slot was poisoned"))?;
            if let Some(slot) = local_slot.as_ref() {
                Arc::clone(slot)
            } else {
                let slot = active_lease_slot(&self.path)?;
                *local_slot = Some(Arc::clone(&slot));
                slot
            }
        };
        let mut lease = slot
            .lock()
            .map_err(|_| anyhow::anyhow!("active session lease was poisoned"))?;
        if lease.is_none() {
            let target = active_lease_target(&self.path);
            *lease = Some(Arc::new(
                CrossProcessFileLock::try_acquire(&target)?
                    .ok_or_else(|| anyhow::Error::new(SessionWriterBusy))?,
            ));
        }
        Ok(())
    }

    pub(in crate::sessions) fn metadata_cache(&self) -> &Mutex<Option<SessionMetadataRecord>> {
        &self.metadata_cache
    }
}

fn replay_state_for_path(path: &Path) -> Arc<SessionReplayState> {
    let key = normalize_active_lease_path(path);
    let registry = SESSION_REPLAY_GENERATIONS.get_or_init(|| Mutex::new(HashMap::new()));
    let Ok(mut states) = registry.lock() else {
        // A poisoned optimization registry must not make durable sessions unavailable.
        return Arc::new(SessionReplayState {
            generation: std::sync::atomic::AtomicU64::new(0),
        });
    };
    if let Some(state) = states.get(&key).and_then(Weak::upgrade) {
        return state;
    }
    states.retain(|_, state| state.strong_count() > 0);
    let state = Arc::new(SessionReplayState {
        generation: std::sync::atomic::AtomicU64::new(0),
    });
    states.insert(key, Arc::downgrade(&state));
    state
}

fn active_lease_slot(path: &Path) -> anyhow::Result<Arc<ActiveLeaseSlot>> {
    let key = normalize_active_lease_path(path);
    let registry = ACTIVE_SESSION_LEASES.get_or_init(|| Mutex::new(HashMap::new()));
    let mut slots = registry
        .lock()
        .map_err(|_| anyhow::anyhow!("active session lease registry was poisoned"))?;
    if let Some(slot) = slots.get(&key).and_then(Weak::upgrade) {
        return Ok(slot);
    }
    slots.retain(|_, slot| slot.strong_count() > 0);
    let slot = Arc::new(Mutex::new(None));
    slots.insert(key, Arc::downgrade(&slot));
    Ok(slot)
}

fn normalize_active_lease_path(path: &Path) -> PathBuf {
    if let Ok(canonical) = path.canonicalize() {
        return canonical;
    }
    if let (Some(parent), Some(file_name)) = (path.parent(), path.file_name())
        && let Ok(parent) = parent.canonicalize()
    {
        return parent.join(file_name);
    }
    path.to_path_buf()
}

pub(in crate::sessions) fn active_lease_target(path: &Path) -> PathBuf {
    path.with_extension("active")
}

impl Session {
    pub fn id(&self) -> &str {
        &self.id
    }

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

    pub(crate) fn replay_generation(&self) -> u64 {
        self.replay_state
            .generation
            .load(std::sync::atomic::Ordering::Acquire)
    }

    pub(in crate::sessions) fn mark_replay_changed(&self) {
        self.replay_state
            .generation
            .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
    }
}