supercode_interchange/

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::native_store::load_native_store_family;
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>,
        /// Total normalized messages in the source-side display projection.
        total_message_count: usize,
    },
    /// 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,
                total_message_count,
            } => json!({
                "type": "messages_appended",
                "sequence": sequence,
                "session_id": session_id,
                "messages": messages.iter().map(message_json).collect::<Vec<_>>(),
                "total_message_count": total_message_count,
            }),
            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>,
    goose_sqlite: bool,
    fidelity: Fidelity,
    include_subagents: bool,
    message_limit: Option<usize>,
    max_message_chars: Option<usize>,
    display_history: bool,
    current: Session,
    fingerprint: Vec<PathStamp>,
    initial_pending: bool,
    next_sequence: u64,
}

#[derive(Clone, Copy)]
struct FollowerView {
    include_subagents: bool,
    message_limit: Option<usize>,
    max_message_chars: Option<usize>,
    display_history: bool,
}

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> {
        Self::open_locator_with_view(locator, fidelity, true, None, None, false)
    }

    /// Open a frontend-oriented follower whose snapshots contain only the
    /// selected parent and at most `message_limit` trailing messages.
    pub fn open_locator_with_view(
        locator: &SessionLocator,
        fidelity: Fidelity,
        include_subagents: bool,
        message_limit: Option<usize>,
        max_message_chars: Option<usize>,
        display_history: bool,
    ) -> Result<Self> {
        match &locator.storage {
            StorageLocator::File { path } => Self::open_with_options(
                path,
                None,
                false,
                fidelity,
                FollowerView {
                    include_subagents,
                    message_limit,
                    max_message_chars,
                    display_history,
                },
            ),
            StorageLocator::Sqlite { path, selector } => Self::open_with_options(
                path,
                Some(selector),
                locator.harness.as_str() == crate::HarnessId::GOOSE,
                fidelity,
                FollowerView {
                    include_subagents,
                    message_limit,
                    max_message_chars,
                    display_history,
                },
            ),
        }
    }

    /// 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> {
        Self::open_with_options(
            path,
            opencode_session,
            false,
            fidelity,
            FollowerView {
                include_subagents: true,
                message_limit: None,
                max_message_chars: None,
                display_history: false,
            },
        )
    }

    fn open_with_options(
        path: impl Into<PathBuf>,
        opencode_session: Option<&str>,
        goose_sqlite: bool,
        fidelity: Fidelity,
        view: FollowerView,
    ) -> 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 mut current = load_selected(&path, selected.as_deref(), goose_sqlite, fidelity, view)?;
        bound_session_view(&mut current, view.message_limit, view.max_message_chars);
        if sqlite && selected.is_none() {
            selected = current.meta.session_id.clone();
        }
        let fingerprint =
            source_fingerprint(&path, &current, selected.as_deref(), view.include_subagents)?;

        Ok(Self {
            path,
            opencode_session: selected,
            goose_sqlite,
            fidelity,
            include_subagents: view.include_subagents,
            message_limit: view.message_limit,
            max_message_chars: view.max_message_chars,
            display_history: view.display_history,
            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(),
            self.include_subagents,
        )?;
        if observed == self.fingerprint {
            return Ok(None);
        }

        let loaded = load_selected(
            &self.path,
            self.opencode_session.as_deref(),
            self.goose_sqlite,
            self.fidelity,
            FollowerView {
                include_subagents: self.include_subagents,
                message_limit: self.message_limit,
                max_message_chars: self.max_message_chars,
                display_history: self.display_history,
            },
        );
        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(mut session) => {
                bound_session_view(&mut session, self.message_limit, self.max_message_chars);
                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);
        let append_prefix = if identity_same && subagents_same {
            append_prefix_len(&self.current.messages, &session.messages)
        } else {
            0
        };
        if append_prefix > 0 && session.messages.len() > append_prefix {
            let messages = session.messages[append_prefix..].to_vec();
            let session_id = session.meta.session_id.clone();
            let total_message_count = session
                .imported_message_count
                .unwrap_or(session.messages.len())
                .max(session.messages.len());
            self.current = session;
            return Some(SessionWatchEvent::MessagesAppended {
                sequence: self.take_sequence(),
                session_id,
                messages,
                total_message_count,
            });
        }

        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>,
    goose_sqlite: bool,
    fidelity: Fidelity,
    view: FollowerView,
) -> Result<Session> {
    let sqlite = looks_like_sqlite(path);
    if sqlite {
        if goose_sqlite {
            let selector = selected.ok_or_else(|| {
                Error::Other("a Goose SQLite locator requires a session selector".to_string())
            })?;
            Ok(Session::from_goose_sqlite(path, selector)?)
        } else {
            Ok(Session::from_opencode_sqlite(path, selected)?)
        }
    } else if let Some(session) = load_native_store_family(path)? {
        Ok(session)
    } else if view.display_history {
        Ok(Session::load_display_view(
            path,
            fidelity,
            view.message_limit.unwrap_or(500),
        )?)
    } else if view.include_subagents {
        Ok(Session::load_with_fidelity(path, fidelity)?)
    } else {
        Ok(Session::load_parent_with_fidelity(path, fidelity)?)
    }
}

#[doc(hidden)]
pub fn bound_session_view(
    session: &mut Session,
    message_limit: Option<usize>,
    max_message_chars: Option<usize>,
) {
    if let Some(limit) = message_limit {
        if session.messages.len() > limit {
            session.messages.drain(..session.messages.len() - limit);
        }
    }

    let Some(max_chars) = max_message_chars else {
        return;
    };
    for message in &mut session.messages {
        if let Some(content) = &mut message.content {
            truncate_utf8(content, max_chars);
        }
        if let Some(parts) = &mut message.content_parts {
            for part in parts {
                truncate_value_strings(part, max_chars);
            }
        }
        if let Some(tool_calls) = &mut message.tool_calls {
            for call in tool_calls {
                truncate_utf8(&mut call.function.arguments, max_chars);
            }
        }
        for value in message.metadata.values_mut() {
            truncate_utf8(value, max_chars);
        }
    }
}

fn truncate_value_strings(value: &mut Value, max_chars: usize) {
    match value {
        Value::String(text) => truncate_utf8(text, max_chars),
        Value::Array(values) => {
            for value in values {
                truncate_value_strings(value, max_chars);
            }
        }
        Value::Object(values) => {
            for value in values.values_mut() {
                truncate_value_strings(value, max_chars);
            }
        }
        _ => {}
    }
}

fn truncate_utf8(value: &mut String, max_chars: usize) {
    let Some((byte_index, _)) = value.char_indices().nth(max_chars) else {
        return;
    };
    value.truncate(byte_index);
    value.push_str("\n…");
}

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))
}

/// Length of the already-known prefix in `next`. A bounded display window is
/// either a plain sliding tail, or an anchored tail whose first user row stays
/// pinned while records immediately after it slide. Both shapes prove that
/// consumers can append the remaining suffix without replacing visible rows.
fn append_prefix_len(current: &[ChatMessage], next: &[ChatMessage]) -> usize {
    let plain = (1..=current.len().min(next.len()))
        .rev()
        .find(|&length| current[current.len() - length..] == next[..length])
        .unwrap_or(0);
    let anchored = if current.first() == next.first() && next.len() > 1 {
        (1..=current.len().saturating_sub(1).min(next.len() - 1))
            .rev()
            .find(|&length| current[current.len() - length..] == next[1..1 + length])
            .map(|length| length + 1)
            .unwrap_or(0)
    } else {
        0
    };
    plain.max(anchored)
}

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::Gemini => "gemini",
        SessionSource::Goose => "goose",
        SessionSource::Native => "native",
    }
}

#[doc(hidden)]
pub 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(),
        "total_message_count": session.imported_message_count.unwrap_or(session.messages.len()).max(session.messages.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>,
    include_subagents: bool,
) -> Result<Vec<PathStamp>> {
    let mut stamps = vec![path_stamp(path)?];
    match session.meta.source {
        SessionSource::ClaudeCode if include_subagents => {
            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"))?);
            }
        }
        SessionSource::Native => {
            if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
                stamps.push(path_stamp(
                    &parent.join(format!("{}.sidecar.jsonl", stem.to_string_lossy())),
                )?);
                stamps.push(path_stamp(
                    &parent.join(format!("{}.meta.json", stem.to_string_lossy())),
                )?);
                collect_tree_stamps(
                    &parent.join(format!("{}.subagents", stem.to_string_lossy())),
                    &mut stamps,
                )?;
            }
        }
        _ => {}
    }
    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()),
    }
}

#[cfg(test)]
mod tests {
    use super::append_prefix_len;
    use crate::ChatMessage;

    #[test]
    fn bounded_append_overlap_handles_plain_and_user_anchored_windows() {
        let user = ChatMessage::user("anchor");
        let one = ChatMessage::assistant("one");
        let two = ChatMessage::assistant("two");
        let three = ChatMessage::assistant("three");
        let newest = ChatMessage::user("newest");

        assert_eq!(
            append_prefix_len(
                &[one.clone(), two.clone(), three.clone()],
                &[two.clone(), three.clone(), newest.clone()],
            ),
            2,
        );
        assert_eq!(
            append_prefix_len(
                &[user.clone(), one, two.clone(), three.clone()],
                &[user, two, three, newest],
            ),
            3,
        );
    }
}