magi-code 0.77.1

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>>>,
    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<super::SessionWriterLease>>,
}

struct SessionReplayState {
    generation: std::sync::atomic::AtomicU64,
    last_terminal_status_generation: std::sync::atomic::AtomicU64,
    last_compaction_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),
            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 {}

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)),
            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::anyhow!("session is busy: another frontend owns the writer")
            })?;
            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::anyhow!("session is active in another process: {}", self.id)
                })?,
            ));
        }
        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),
            last_terminal_status_generation: std::sync::atomic::AtomicU64::new(0),
            last_compaction_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),
        last_terminal_status_generation: std::sync::atomic::AtomicU64::new(0),
        last_compaction_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(crate) fn terminal_status_recorded_since(&self, baseline_generation: u64) -> bool {
        let terminal_generation = self
            .replay_state
            .last_terminal_status_generation
            .load(std::sync::atomic::Ordering::Acquire);
        let compaction_generation = self
            .replay_state
            .last_compaction_generation
            .load(std::sync::atomic::Ordering::Acquire);
        terminal_generation > baseline_generation && terminal_generation > compaction_generation
    }

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

    pub(in crate::sessions) fn mark_terminal_status_recorded(&self, generation: u64) {
        self.replay_state
            .last_terminal_status_generation
            .store(generation, std::sync::atomic::Ordering::Release);
    }

    pub(in crate::sessions) fn mark_compaction_recorded(&self, generation: u64) {
        self.replay_state
            .last_compaction_generation
            .store(generation, std::sync::atomic::Ordering::Release);
    }

    #[cfg(test)]
    pub(crate) fn unchecked_for_test(id: String, path: PathBuf) -> Self {
        Self::new(id, path)
    }

    #[cfg(test)]
    pub(crate) fn release_active_lease_for_test(&self) {
        if let Some(slot) = self.active_lease.lock().unwrap().as_ref() {
            *slot.lock().unwrap() = None;
        }
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    use proptest::prelude::*;
    use tempfile::TempDir;

    fn valid_session_id_strategy() -> impl Strategy<Value = String> {
        proptest::string::string_regex("[A-Za-z0-9_-]{1,64}").unwrap()
    }

    fn invalid_session_id_strategy() -> impl Strategy<Value = String> {
        prop_oneof![
            Just(String::new()),
            Just(".".to_string()),
            any::<String>().prop_map(|value| format!("{value}..")),
            any::<String>().prop_map(|value| format!("{value}/{value}")),
            any::<String>().prop_map(|value| format!("{value}\\{value}")),
            any::<String>().prop_map(|value| format!("{value}.jsonl")),
            any::<String>().prop_map(|value| format!("{value}é")),
        ]
    }

    proptest! {
        #[test]
        fn path_for_valid_id_keeps_valid_ids_under_session_root(id in valid_session_id_strategy()) {
            let temp = TempDir::new().unwrap();
            let root = temp.path().join("sessions");
            let manager = SessionManager::new(root.clone());
            let path = manager.path_for_valid_id(&id).unwrap();
            let normalized_root = crate::path_utils::lexical_normalize(&root);
            let normalized_path = crate::path_utils::lexical_normalize(&path);
            let expected_file_name = format!("{id}.jsonl");

            prop_assert!(normalized_path.starts_with(&normalized_root));
            prop_assert_eq!(normalized_path.parent(), Some(normalized_root.as_path()));
            prop_assert_eq!(
                normalized_path.file_name().and_then(|file_name| file_name.to_str()),
                Some(expected_file_name.as_str())
            );

            let session = manager.open(id.clone()).unwrap();
            prop_assert_eq!(session.id(), id);
            prop_assert_eq!(session.path(), path.as_path());
        }

        #[test]
        fn open_and_path_for_valid_id_reject_generated_unsafe_ids(id in invalid_session_id_strategy()) {
            let temp = TempDir::new().unwrap();
            let manager = SessionManager::new(temp.path().join("sessions"));

            prop_assert!(manager.open(id.clone()).is_err());
            prop_assert!(manager.path_for_valid_id(&id).is_err());
        }
    }

    #[test]
    fn session_open_rejects_unsafe_ids_before_joining_paths() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        for id in [
            "",
            "..",
            "../escape",
            "nested/id",
            "nested\\id",
            "/absolute",
            "bad.jsonl",
        ] {
            assert!(manager.open(id).is_err(), "accepted unsafe id {id:?}");
        }
        assert!(manager.open("safe_ID-123").is_ok());
    }
}