supercode/

watch.rs

//! Passive, change-triggered following of local coding-harness sessions.
//!
//! This module deliberately observes persisted session state; it does not
//! attach to, control, or infer the liveness of the process writing it.

use std::fs::Metadata;
use std::path::{Path, PathBuf};
use std::time::UNIX_EPOCH;

use serde_json::{json, Value};

use crate::catalog::{SessionLocator, StorageLocator};
use crate::session::{looks_like_sqlite, Session, SessionSource};
use crate::{ChatMessage, Error, Fidelity, Result};

/// Why a watcher emitted a complete session snapshot.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionSnapshotReason {
    /// The first event emitted after opening the follower.
    Initial,
    /// Existing normalized history changed, disappeared, or branched.
    HistoryRewritten,
    /// Session identity or other non-message state changed.
    SourceChanged,
}

impl SessionSnapshotReason {
    fn as_str(self) -> &'static str {
        match self {
            Self::Initial => "initial",
            Self::HistoryRewritten => "history_rewritten",
            Self::SourceChanged => "source_changed",
        }
    }
}

/// A normalized event emitted while following a local session.
#[derive(Debug, Clone)]
pub enum SessionWatchEvent {
    /// A complete normalized view of the selected session.
    SessionSnapshot {
        /// Monotonically increasing sequence number, starting at one.
        sequence: u64,
        /// Why the full snapshot was necessary.
        reason: SessionSnapshotReason,
        /// The current normalized session.
        session: Box<Session>,
    },
    /// Messages appended without changing existing normalized history.
    MessagesAppended {
        /// Monotonically increasing sequence number.
        sequence: u64,
        /// Selected session id, when the source records one.
        session_id: Option<String>,
        /// Newly appended normalized messages.
        messages: Vec<ChatMessage>,
    },
    /// A recoverable read or parse problem. The follower remains usable.
    WatchError {
        /// Monotonically increasing sequence number.
        sequence: u64,
        /// Human-readable description of the problem.
        message: String,
    },
}

impl SessionWatchEvent {
    /// The event's monotonic sequence number.
    pub fn sequence(&self) -> u64 {
        match self {
            Self::SessionSnapshot { sequence, .. }
            | Self::MessagesAppended { sequence, .. }
            | Self::WatchError { sequence, .. } => *sequence,
        }
    }

    /// Render this event as one self-contained JSON value suitable for NDJSON.
    pub fn to_json(&self) -> Value {
        match self {
            Self::SessionSnapshot {
                sequence,
                reason,
                session,
            } => json!({
                "type": "session_snapshot",
                "sequence": sequence,
                "reason": reason.as_str(),
                "session": normalized_session_json(session),
            }),
            Self::MessagesAppended {
                sequence,
                session_id,
                messages,
            } => json!({
                "type": "messages_appended",
                "sequence": sequence,
                "session_id": session_id,
                "messages": messages.iter().map(message_json).collect::<Vec<_>>(),
            }),
            Self::WatchError { sequence, message } => json!({
                "type": "watch_error",
                "sequence": sequence,
                "recoverable": true,
                "message": message,
            }),
        }
    }
}

/// Poll-based follower for one persisted Claude Code, Codex, Pi, OpenCode, or Grok
/// session.
///
/// Polling first compares cheap filesystem stamps. The source is fully parsed
/// only after a relevant file changes. This keeps idle polling cheap while
/// retaining the existing, well-tested format loaders as the source of truth.
pub struct SessionFollower {
    path: PathBuf,
    opencode_session: Option<String>,
    fidelity: Fidelity,
    current: Session,
    fingerprint: Vec<PathStamp>,
    initial_pending: bool,
    next_sequence: u64,
}

impl SessionFollower {
    /// Open a persisted session using its durable catalog locator.
    pub fn open_locator(locator: &SessionLocator) -> Result<Self> {
        Self::open_locator_with_fidelity(locator, Fidelity::ByteLossless)
    }

    /// [`Self::open_locator`] at a declared fidelity.
    ///
    /// A read-only mirror follows at [`Fidelity::Semantic`] so a compacted
    /// transcript keeps streaming instead of turning every poll into a
    /// `watch_error`. See [`crate::Session::load_with_fidelity`].
    pub fn open_locator_with_fidelity(
        locator: &SessionLocator,
        fidelity: Fidelity,
    ) -> Result<Self> {
        match &locator.storage {
            StorageLocator::File { path } => Self::open_with_fidelity(path, None, fidelity),
            StorageLocator::Sqlite { path, selector } => {
                Self::open_with_fidelity(path, Some(selector), fidelity)
            }
        }
    }

    /// Open a local session for passive following.
    ///
    /// `opencode_session` is valid only for an OpenCode SQLite store. When it
    /// is omitted, the initially selected session is pinned for all later
    /// polls rather than following whichever database row becomes newest.
    pub fn open(path: impl Into<PathBuf>, opencode_session: Option<&str>) -> Result<Self> {
        Self::open_with_fidelity(path, opencode_session, Fidelity::ByteLossless)
    }

    /// [`Self::open`] at a declared fidelity.
    pub fn open_with_fidelity(
        path: impl Into<PathBuf>,
        opencode_session: Option<&str>,
        fidelity: Fidelity,
    ) -> Result<Self> {
        let path = path.into();
        let sqlite = looks_like_sqlite(&path);
        if opencode_session.is_some() && !sqlite {
            return Err(Error::Other(format!(
                "an OpenCode session selector requires a SQLite store; {} is not one",
                path.display()
            )));
        }

        let mut selected = opencode_session.map(str::to_owned);
        let current = load_selected(&path, selected.as_deref(), sqlite, fidelity)?;
        if sqlite && selected.is_none() {
            selected = current.meta.session_id.clone();
        }
        let fingerprint = source_fingerprint(&path, &current, selected.as_deref())?;

        Ok(Self {
            path,
            opencode_session: selected,
            fidelity,
            current,
            fingerprint,
            initial_pending: true,
            next_sequence: 1,
        })
    }

    /// Inspect the filesystem once and return the next event, if any.
    ///
    /// The first call always returns an initial snapshot. Later calls return
    /// `None` while the relevant filesystem stamps are unchanged.
    pub fn poll(&mut self) -> Result<Option<SessionWatchEvent>> {
        if self.initial_pending {
            self.initial_pending = false;
            return Ok(Some(self.snapshot(SessionSnapshotReason::Initial)));
        }

        let observed =
            source_fingerprint(&self.path, &self.current, self.opencode_session.as_deref())?;
        if observed == self.fingerprint {
            return Ok(None);
        }

        let sqlite = looks_like_sqlite(&self.path);
        let loaded = load_selected(
            &self.path,
            self.opencode_session.as_deref(),
            sqlite,
            self.fidelity,
        );
        self.fingerprint = observed;
        let next = match loaded {
            Ok(session) if session.parse_error_lines > 0 => {
                let count = session.parse_error_lines;
                Some(self.watch_error(format!(
                    "{} contains {count} malformed or truncated JSON line(s); retaining the last good snapshot",
                    self.path.display()
                )))
            }
            Err(error) => Some(self.watch_error(format!(
                "could not reload {}: {error}; retaining the last good snapshot",
                self.path.display()
            ))),
            Ok(session) => self.event_for_session(session),
        };
        Ok(next)
    }

    fn event_for_session(&mut self, session: Session) -> Option<SessionWatchEvent> {
        if normalized_session_eq(&self.current, &session) {
            self.current = session;
            return None;
        }

        let identity_same = session_identity_eq(&self.current, &session);
        let subagents_same = normalized_subagents_eq(&self.current, &session);
        if identity_same
            && subagents_same
            && session.messages.len() > self.current.messages.len()
            && session.messages.starts_with(&self.current.messages)
        {
            let messages = session.messages[self.current.messages.len()..].to_vec();
            let session_id = session.meta.session_id.clone();
            self.current = session;
            return Some(SessionWatchEvent::MessagesAppended {
                sequence: self.take_sequence(),
                session_id,
                messages,
            });
        }

        let reason = if identity_same {
            SessionSnapshotReason::HistoryRewritten
        } else {
            SessionSnapshotReason::SourceChanged
        };
        self.current = session;
        Some(self.snapshot(reason))
    }

    fn snapshot(&mut self, reason: SessionSnapshotReason) -> SessionWatchEvent {
        SessionWatchEvent::SessionSnapshot {
            sequence: self.take_sequence(),
            reason,
            session: Box::new(self.current.clone()),
        }
    }

    fn watch_error(&mut self, message: String) -> SessionWatchEvent {
        SessionWatchEvent::WatchError {
            sequence: self.take_sequence(),
            message,
        }
    }

    fn take_sequence(&mut self) -> u64 {
        let sequence = self.next_sequence;
        self.next_sequence += 1;
        sequence
    }
}

fn load_selected(
    path: &Path,
    selected: Option<&str>,
    sqlite: bool,
    fidelity: Fidelity,
) -> Result<Session> {
    if sqlite {
        Session::from_opencode_sqlite(path, selected)
    } else {
        Session::load_with_fidelity(path, fidelity)
    }
}

fn session_identity_eq(left: &Session, right: &Session) -> bool {
    left.meta.source == right.meta.source
        && left.meta.session_id == right.meta.session_id
        && left.meta.model == right.meta.model
        && left.meta.cwd == right.meta.cwd
        && left.meta.system_prompt == right.meta.system_prompt
        && left.meta.agent_id == right.meta.agent_id
        && left.meta.parent_tool_use_id == right.meta.parent_tool_use_id
        && left.meta.lineage == right.meta.lineage
}

fn normalized_session_eq(left: &Session, right: &Session) -> bool {
    session_identity_eq(left, right)
        && left.messages == right.messages
        && normalized_subagents_eq(left, right)
        && left.parse_error_lines == right.parse_error_lines
        && left.load_residue == right.load_residue
}

fn normalized_subagents_eq(left: &Session, right: &Session) -> bool {
    left.subagents.len() == right.subagents.len()
        && left
            .subagents
            .iter()
            .zip(&right.subagents)
            .all(|(left, right)| normalized_session_eq(left, right))
}

fn source_name(source: SessionSource) -> &'static str {
    match source {
        SessionSource::ClaudeCode => "claude_code",
        SessionSource::Codex => "codex",
        SessionSource::OpenCode => "opencode",
        SessionSource::Pi => "pi",
        SessionSource::Grok => "grok",
        SessionSource::Native => "native",
    }
}

fn message_json(message: &ChatMessage) -> Value {
    let mut value = serde_json::to_value(message).unwrap_or_else(|_| json!({}));
    if let Value::Object(object) = &mut value {
        object.insert("metadata".to_string(), json!(message.metadata));
    }
    value
}

/// Render a normalized session as a language-neutral JSON value.
pub fn normalized_session_json(session: &Session) -> Value {
    json!({
        "source": source_name(session.meta.source),
        "session_id": session.meta.session_id,
        "model": session.meta.model,
        "cwd": session.meta.cwd,
        "system_prompt": session.meta.system_prompt,
        "agent_id": session.meta.agent_id,
        "parent_tool_use_id": session.meta.parent_tool_use_id,
        "lineage": session.meta.lineage,
        "messages": session.messages.iter().map(message_json).collect::<Vec<_>>(),
        "subagents": session.subagents.iter().map(normalized_session_json).collect::<Vec<_>>(),
        "raw_record_count": session.raw.len(),
        "parse_error_lines": session.parse_error_lines,
        // Same pair `harness.v1.sessions.export` reports for an artifact: the
        // level reached, and exactly what was given up to reach it. `semantic`
        // with a non-empty residue means this is a read-only VIEW of a
        // transcript that cannot be losslessly reconstructed.
        "fidelity": session.load_fidelity(),
        "residue": session.load_residue,
    })
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct PathStamp {
    path: PathBuf,
    kind: StampKind,
    len: u64,
    modified_nanos: Option<u128>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StampKind {
    Missing,
    File,
    Directory,
    Other,
}

fn source_fingerprint(
    path: &Path,
    session: &Session,
    selected_session: Option<&str>,
) -> Result<Vec<PathStamp>> {
    let mut stamps = vec![path_stamp(path)?];
    match session.meta.source {
        SessionSource::ClaudeCode => {
            if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
                collect_tree_stamps(&parent.join(stem).join("subagents"), &mut stamps)?;
            }
        }
        SessionSource::OpenCode if looks_like_sqlite(path) => {
            stamps.push(path_stamp(&path_with_suffix(path, "-wal"))?);
            stamps.push(path_stamp(&path_with_suffix(path, "-shm"))?);
            if let (Some(parent), Some(session_id)) = (path.parent(), selected_session) {
                stamps.push(path_stamp(
                    &parent
                        .join("storage")
                        .join("session_diff")
                        .join(format!("{session_id}.json")),
                )?);
            }
        }
        SessionSource::Grok => {
            if let Some(parent) = path.parent() {
                // `chat_history.jsonl` is the resumable transcript. The
                // adjacent update stream and summary are cheap companion
                // stamps that make a running Grok session wake the follower
                // even while it is between committed transcript turns.
                stamps.push(path_stamp(&parent.join("updates.jsonl"))?);
                stamps.push(path_stamp(&parent.join("summary.json"))?);
            }
        }
        _ => {}
    }
    stamps.sort_by(|left, right| left.path.cmp(&right.path));
    Ok(stamps)
}

fn collect_tree_stamps(path: &Path, out: &mut Vec<PathStamp>) -> Result<()> {
    collect_tree_stamps_inner(path, out, true)
}

fn collect_tree_stamps_inner(path: &Path, out: &mut Vec<PathStamp>, follow: bool) -> Result<()> {
    let stamp = if follow {
        path_stamp(path)?
    } else {
        path_stamp_no_follow(path)?
    };
    let is_directory = stamp.kind == StampKind::Directory;
    out.push(stamp);
    if !is_directory {
        return Ok(());
    }

    let mut children = std::fs::read_dir(path)?.collect::<std::io::Result<Vec<_>>>()?;
    children.sort_by_key(|entry| entry.path());
    for child in children {
        collect_tree_stamps_inner(&child.path(), out, false)?;
    }
    Ok(())
}

fn path_stamp(path: &Path) -> Result<PathStamp> {
    path_stamp_with(path, |path| std::fs::metadata(path))
}

fn path_stamp_no_follow(path: &Path) -> Result<PathStamp> {
    path_stamp_with(path, |path| std::fs::symlink_metadata(path))
}

fn path_stamp_with(
    path: &Path,
    metadata: impl FnOnce(&Path) -> std::io::Result<Metadata>,
) -> Result<PathStamp> {
    match metadata(path) {
        Ok(metadata) => Ok(stamp_from_metadata(path, &metadata)),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(PathStamp {
            path: path.to_path_buf(),
            kind: StampKind::Missing,
            len: 0,
            modified_nanos: None,
        }),
        Err(error) => Err(error.into()),
    }
}

fn path_with_suffix(path: &Path, suffix: &str) -> PathBuf {
    let mut value = path.as_os_str().to_os_string();
    value.push(suffix);
    PathBuf::from(value)
}

fn stamp_from_metadata(path: &Path, metadata: &Metadata) -> PathStamp {
    let file_type = metadata.file_type();
    let kind = if file_type.is_file() {
        StampKind::File
    } else if file_type.is_dir() {
        StampKind::Directory
    } else {
        StampKind::Other
    };
    PathStamp {
        path: path.to_path_buf(),
        kind,
        len: metadata.len(),
        modified_nanos: metadata
            .modified()
            .ok()
            .and_then(|time| time.duration_since(UNIX_EPOCH).ok())
            .map(|duration| duration.as_nanos()),
    }
}