magi-code 0.102.1

//! Passive, opt-in maintenance. Scheduling observes only committed session records.
mod generation;
pub(crate) use generation::summarize_records;

use crate::{
    config::{EffectiveConfig, Settings},
    sessions::{Session, SessionEvent, SessionEventKind},
};
use std::{
    path::PathBuf,
    sync::{Arc, Mutex},
    thread,
};

#[derive(Clone)]
struct Job {
    config: EffectiveConfig,
    settings: Settings,
    cwd: PathBuf,
}

#[derive(Clone)]
pub(crate) struct Trigger {
    pub(crate) through_event_id: String,
    pub(crate) turn_ended: bool,
}

pub(crate) struct Observer {
    job: Job,
    count: usize,
    pending: Option<Trigger>,
    running: bool,
    waiting_for_response: bool,
    latest_event_id: Option<String>,
}

pub(crate) type ObserverSlot = Arc<Mutex<Option<Observer>>>;

pub(crate) fn configure(
    session: &Session,
    config: &EffectiveConfig,
    settings: &Settings,
    cwd: &std::path::Path,
) {
    let Ok(mut slot) = session.maintenance.lock() else {
        return;
    };
    if !settings.psm.enabled && slot.is_none() {
        return;
    }
    let job = Job {
        config: config.clone(),
        settings: settings.clone(),
        cwd: cwd.to_path_buf(),
    };
    if let Some(observer) = slot.as_mut() {
        observer.job = job;
        if !settings.psm.enabled {
            observer.pending = None;
            observer.count = 0;
            observer.waiting_for_response = false;
        }
    } else if settings.psm.enabled {
        *slot = Some(Observer {
            job,
            count: 0,
            pending: None,
            running: false,
            waiting_for_response: false,
            latest_event_id: None,
        });
    }
}

/// Called after durable append, including nested tools, steering, and failed/cancelled turns.
pub(crate) fn observe(session: &Session, events: &[SessionEvent]) {
    let Ok(mut slot) = session.maintenance.lock() else {
        return;
    };
    let Some(observer) = slot.as_mut() else {
        return;
    };
    if !observer.job.settings.psm.enabled {
        return;
    }
    for event in events {
        observer.latest_event_id = event.event_id.clone();
        let tracked = is_tracked(&event.event_type);
        if tracked {
            observer.count = observer.count.saturating_add(1);
        }
        let turn_ended = event.kind() == Some(SessionEventKind::TurnStatus)
            && event.payload["status"]
                .as_str()
                .is_some_and(|status| status != "incomplete");
        let response_ready =
            observer.waiting_for_response && event.kind() == Some(SessionEventKind::ToolResult);
        if ((tracked && observer.count >= observer.job.settings.psm.event_threshold)
            || turn_ended
            || response_ready)
            && let Some(id) = &event.event_id
        {
            observer.pending = Some(Trigger {
                through_event_id: id.clone(),
                turn_ended,
            });
            observer.count = 0;
        }
    }
    start_worker(session, observer);
}

fn start_worker(session: &Session, observer: &mut Observer) {
    if observer.running || observer.pending.is_none() {
        return;
    }
    observer.running = true;
    let worker_session = session.clone();
    if thread::Builder::new()
        .name("progressive-session-maintenance".into())
        .spawn(move || run(worker_session))
        .is_err()
    {
        observer.running = false;
    }
}

/// The outer runner owns real turn completion; successful turns need not append a status record.
pub(crate) struct TurnGuard<'a>(pub(crate) Option<&'a Session>);

impl Drop for TurnGuard<'_> {
    fn drop(&mut self) {
        let Some(session) = self.0 else { return };
        let Ok(mut slot) = session.maintenance.lock() else {
            return;
        };
        let Some(observer) = slot.as_mut() else {
            return;
        };
        if !observer.job.settings.psm.enabled {
            return;
        }
        let Some(id) = observer.latest_event_id.clone() else {
            return;
        };
        observer.pending = Some(Trigger {
            through_event_id: id,
            turn_ended: true,
        });
        observer.count = 0;
        start_worker(session, observer);
    }
}

pub(crate) fn is_tracked(kind: &str) -> bool {
    matches!(
        kind,
        "user_input" | "assistant_output" | "tool_call" | "tool_result"
    )
}

fn run(session: Session) {
    while let Some((job, trigger)) = session.maintenance.lock().ok().and_then(|mut slot| {
        let observer = slot.as_mut()?;
        let trigger = observer.pending.take()?;
        Some((observer.job.clone(), trigger))
    }) {
        let mut status = crate::sessions::progressive_summaries::Status::new(
            &session,
            &trigger,
            &job.config,
            &job.settings,
        );
        let result = crate::sessions::progressive_summaries::maintain(
            &session,
            &trigger,
            &job.config,
            &job.settings,
            &job.cwd,
            &mut status,
        );
        status.finish(&result);
        let Ok(mut slot) = session.maintenance.lock() else {
            return;
        };
        let Some(observer) = slot.as_mut() else {
            return;
        };
        match result {
            Ok(waiting) => observer.waiting_for_response = waiting,
            Err(_) => {
                // No cursor advance on failure. A later planned trigger re-reads the gap.
                observer.waiting_for_response = false;
            }
        }
        if observer.pending.is_none() {
            observer.running = false;
            return;
        }
    }
    if let Ok(mut slot) = session.maintenance.lock()
        && let Some(observer) = slot.as_mut()
    {
        observer.running = false;
    }
}