supercode-harness 0.4.13

//! Live stock-Codex session discovery.
//!
//! Codex does not publish a peer registry or a supported attachment endpoint,
//! but its process keeps every rollout it currently owns open. This module
//! joins that process-owned file descriptor back to the persisted catalog
//! path. The rollout's last explicit lifecycle event then distinguishes an
//! executing turn from a merely running session. No timing or CPU heuristic
//! is used.

use std::collections::{HashMap, HashSet};
use std::fs::File;
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use serde_json::Value;

const LIFECYCLE_SCAN_BYTES: u64 = 1024 * 1024;
const LIFECYCLE_OVERLAP_BYTES: u64 = 64 * 1024;
const OWNERSHIP_REFRESH_INTERVAL: Duration = Duration::from_secs(1);
const OWNERSHIP_MISS_CONFIRMATIONS: u8 = 2;

/// Activity proven for a rollout owned by stock Codex.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CodexPeerStatus {
    /// Codex owns the rollout, but no active turn is proven.
    Running,
    /// The latest lifecycle boundary completed or aborted a turn.
    Idle,
    /// The latest lifecycle boundary starts a task.
    Busy,
}

impl CodexPeerStatus {
    /// Stable wire spelling shared by the harness protocol.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Running => "running",
            Self::Idle => "idle",
            Self::Busy => "busy",
        }
    }
}

/// Every Codex rollout currently held open by a stock `codex` process, with
/// the narrowest activity state its own event stream proves.
pub fn live_rollouts(sessions_root: &Path) -> HashMap<PathBuf, CodexPeerStatus> {
    CodexPeerTracker::default().sample(sessions_root)
}

/// Cached process-ownership sampler for latency-sensitive activity streams.
///
/// Process/file-descriptor discovery is materially more expensive than
/// reading a bounded lifecycle tail. Ownership is therefore refreshed once a
/// second while known open rollouts are re-read on every activity tick.
#[derive(Debug, Default)]
pub(crate) struct CodexPeerTracker {
    root: Option<PathBuf>,
    open_rollouts: Vec<PathBuf>,
    ownership_misses: HashMap<PathBuf, u8>,
    lifecycle: HashMap<PathBuf, CodexLifecycleCursor>,
    refreshed_at: Option<Instant>,
}

#[derive(Debug, Default)]
struct CodexLifecycleCursor {
    offset: u64,
    status: Option<CodexPeerStatus>,
}

impl CodexPeerTracker {
    pub(crate) fn sample(&mut self, sessions_root: &Path) -> HashMap<PathBuf, CodexPeerStatus> {
        let root = normalized_path(sessions_root);
        let refresh = self.root.as_ref() != Some(&root)
            || self
                .refreshed_at
                .is_none_or(|at| at.elapsed() >= OWNERSHIP_REFRESH_INTERVAL);
        if refresh {
            let observed = platform_open_rollouts()
                .into_iter()
                .map(|path| normalized_path(&path))
                .collect();
            self.open_rollouts =
                reconcile_open_rollouts(&self.open_rollouts, observed, &mut self.ownership_misses);
            self.root = Some(root.clone());
            self.refreshed_at = Some(Instant::now());
            self.lifecycle
                .retain(|path, _| self.open_rollouts.contains(path));
        }
        let mut statuses = HashMap::new();
        for path in &self.open_rollouts {
            if !(path.starts_with(&root)
                && path.extension().and_then(|extension| extension.to_str()) == Some("jsonl"))
            {
                continue;
            }
            let cursor = self.lifecycle.entry(path.clone()).or_default();
            let status = sample_lifecycle_status(path, cursor).unwrap_or(CodexPeerStatus::Running);
            statuses.insert(path.clone(), status);
        }
        statuses
    }
}

/// One process-table sample is negative evidence, not a lifecycle boundary.
/// Keep a previously open rollout through one miss; only consecutive misses
/// retire it. This absorbs transient `pgrep`/`lsof` snapshots without putting
/// a wall-clock guess into session state, while a genuinely exited process is
/// removed on the next independent ownership observation.
fn reconcile_open_rollouts(
    previous: &[PathBuf],
    observed: Vec<PathBuf>,
    misses: &mut HashMap<PathBuf, u8>,
) -> Vec<PathBuf> {
    let observed = observed.into_iter().collect::<HashSet<_>>();
    let previous = previous.iter().cloned().collect::<HashSet<_>>();
    let mut reconciled = observed.clone();

    for path in &observed {
        misses.remove(path);
    }
    for path in previous.difference(&observed) {
        let count = misses.entry(path.clone()).or_insert(0);
        *count = count.saturating_add(1);
        if *count < OWNERSHIP_MISS_CONFIRMATIONS {
            reconciled.insert(path.clone());
        } else {
            misses.remove(path);
        }
    }
    misses.retain(|path, _| previous.contains(path) && !observed.contains(path));

    let mut reconciled = reconciled.into_iter().collect::<Vec<_>>();
    reconciled.sort();
    reconciled
}

/// Activity for a discovered catalog path owned by a currently running Codex.
pub fn rollout_status(
    live: &HashMap<PathBuf, CodexPeerStatus>,
    path: &Path,
) -> Option<CodexPeerStatus> {
    live.get(&normalized_path(path)).copied()
}

/// Lightweight native identity and direct parent from Codex's first
/// `session_meta` record. Activity aggregation uses this to treat a process-
/// owned subagent rollout as work inside its root conversation.
pub(crate) fn rollout_lineage(path: &Path) -> Option<(String, Option<String>)> {
    let mut header = String::new();
    BufReader::new(File::open(path).ok()?.take(256 * 1024))
        .read_line(&mut header)
        .ok()?;
    let value = serde_json::from_str::<Value>(&header).ok()?;
    if value.get("type").and_then(Value::as_str) != Some("session_meta") {
        return None;
    }
    let payload = value.get("payload")?;
    let session_id = payload.get("id")?.as_str()?.to_string();
    let parent_session_id = payload
        .pointer("/source/subagent/thread_spawn/parent_thread_id")
        .or_else(|| payload.get("parent_thread_id"))
        .and_then(Value::as_str)
        .map(str::to_string);
    Some((session_id, parent_session_id))
}

fn sample_lifecycle_status(
    path: &Path,
    cursor: &mut CodexLifecycleCursor,
) -> Option<CodexPeerStatus> {
    let mut file = File::open(path).ok()?;
    let length = file.metadata().ok()?.len();
    if length < cursor.offset {
        cursor.offset = 0;
        cursor.status = None;
    }
    if cursor.offset == 0 {
        cursor.status = latest_lifecycle_status_between(&mut file, 0, length);
    } else if length > cursor.offset {
        if let Some(status) = latest_lifecycle_status_between(&mut file, cursor.offset, length) {
            cursor.status = Some(status);
        }
    }
    cursor.offset = length;
    cursor.status
}

fn latest_lifecycle_status_between(
    file: &mut File,
    floor: u64,
    upper: u64,
) -> Option<CodexPeerStatus> {
    let mut end = upper;
    while end > floor {
        let start = end.saturating_sub(LIFECYCLE_SCAN_BYTES).max(floor);
        file.seek(SeekFrom::Start(start)).ok()?;
        let mut tail = vec![0; (end - start) as usize];
        file.read_exact(&mut tail).ok()?;
        if let Some(status) = lifecycle_status_in_tail(&tail, start == floor) {
            return Some(status);
        }
        if start == floor {
            break;
        }
        // A lifecycle record is small, but an adjacent tool record can be enormous. Overlap keeps
        // a boundary record whole without ever allocating in proportion to the rollout.
        end = start.saturating_add(LIFECYCLE_OVERLAP_BYTES);
    }
    None
}

fn lifecycle_status_in_tail(
    tail: &[u8],
    starts_at_record_boundary: bool,
) -> Option<CodexPeerStatus> {
    const BOUNDARIES: [&str; 3] = [
        "\"type\":\"task_started\"",
        "\"type\":\"task_complete\"",
        "\"type\":\"turn_aborted\"",
    ];

    // The read can begin in the middle of a large UTF-8 JSON string. Ignore
    // that first fragment, then use the standard library's substring search
    // to jump directly between lifecycle candidates instead of inspecting
    // every byte of every tool payload with a naive sliding window.
    let complete_start = if starts_at_record_boundary {
        0
    } else {
        tail.iter()
            .position(|byte| *byte == b'\n')
            .map_or(tail.len(), |newline| newline + 1)
    };
    let text = std::str::from_utf8(&tail[complete_start..]).ok()?;
    let mut search_end = text.len();
    while let Some(candidate) = BOUNDARIES
        .iter()
        .filter_map(|boundary| text[..search_end].rfind(boundary))
        .max()
    {
        let line_start = text[..candidate]
            .rfind('\n')
            .map_or(0, |newline| newline + 1);
        let line_end = text[candidate..]
            .find('\n')
            .map_or(text.len(), |newline| candidate + newline);
        let Ok(event) = serde_json::from_str::<Value>(&text[line_start..line_end]) else {
            search_end = candidate;
            continue;
        };
        if event.get("type").and_then(Value::as_str) != Some("event_msg") {
            search_end = candidate;
            continue;
        }
        match event
            .get("payload")
            .and_then(|payload| payload.get("type"))
            .and_then(Value::as_str)
        {
            Some("task_started") => return Some(CodexPeerStatus::Busy),
            Some("task_complete" | "turn_aborted") => return Some(CodexPeerStatus::Idle),
            _ => {}
        }
        search_end = candidate;
    }
    None
}

fn normalized_path(path: &Path) -> PathBuf {
    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
}

#[cfg(target_os = "macos")]
fn platform_open_rollouts() -> Vec<PathBuf> {
    use std::process::Command;

    // Darwin's pgrep omits every ancestor of the caller unless `-a` is set.
    // Discovery commonly runs underneath the very Codex session it must
    // report (for example inside a Supercode-powered widget), so omitting
    // ancestors makes the current session uniquely invisible.
    let Ok(processes) = Command::new("/usr/bin/pgrep")
        .args(["-a", "-x", "codex"])
        .output()
    else {
        return Vec::new();
    };
    let pids = String::from_utf8_lossy(&processes.stdout)
        .lines()
        .filter_map(|line| line.trim().parse::<u32>().ok())
        .take(128)
        .map(|pid| pid.to_string())
        .collect::<Vec<_>>();
    if pids.is_empty() {
        return Vec::new();
    }
    let Ok(files) = Command::new("/usr/sbin/lsof")
        .args(["-Fn", "-a", "-p", &pids.join(",")])
        .output()
    else {
        return Vec::new();
    };
    String::from_utf8_lossy(&files.stdout)
        .lines()
        .filter_map(|line| line.strip_prefix('n'))
        .filter(|path| path.ends_with(".jsonl"))
        .map(PathBuf::from)
        .collect()
}

#[cfg(target_os = "linux")]
fn platform_open_rollouts() -> Vec<PathBuf> {
    let Ok(processes) = std::fs::read_dir("/proc") else {
        return Vec::new();
    };
    let mut paths = Vec::new();
    for process in processes.flatten() {
        let pid = process.file_name();
        if !pid.as_encoded_bytes().iter().all(u8::is_ascii_digit) {
            continue;
        }
        let process_root = process.path();
        if std::fs::read_to_string(process_root.join("comm"))
            .ok()
            .is_none_or(|name| name.trim() != "codex")
        {
            continue;
        }
        let Ok(descriptors) = std::fs::read_dir(process_root.join("fd")) else {
            continue;
        };
        paths.extend(
            descriptors
                .flatten()
                .filter_map(|descriptor| std::fs::read_link(descriptor.path()).ok())
                .filter(|path| {
                    path.extension().and_then(|extension| extension.to_str()) == Some("jsonl")
                }),
        );
    }
    paths
}

#[cfg(not(any(target_os = "macos", target_os = "linux")))]
fn platform_open_rollouts() -> Vec<PathBuf> {
    Vec::new()
}

#[cfg(test)]
mod tests {
    use std::fs::{remove_file, OpenOptions};
    use std::io::Write;

    use super::*;

    #[test]
    fn long_tool_heavy_turn_is_found_once_then_followed_incrementally() {
        let path = std::env::temp_dir().join(format!(
            "supercode-codex-long-turn-{}-{}.jsonl",
            std::process::id(),
            std::thread::current().name().unwrap_or("test")
        ));
        let mut file = File::create(&path).unwrap();
        writeln!(
            file,
            r#"{{"type":"event_msg","payload":{{"type":"task_started"}}}}"#
        )
        .unwrap();
        write!(
            file,
            r#"{{"type":"response_item","payload":"{}"}}"#,
            "x".repeat(6 * 1024 * 1024)
        )
        .unwrap();
        writeln!(file).unwrap();
        file.flush().unwrap();

        let mut cursor = CodexLifecycleCursor::default();
        assert_eq!(
            sample_lifecycle_status(&path, &mut cursor),
            Some(CodexPeerStatus::Busy)
        );
        let first_offset = cursor.offset;

        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
        writeln!(
            file,
            r#"{{"type":"event_msg","payload":{{"type":"item_completed"}}}}"#
        )
        .unwrap();
        file.flush().unwrap();
        assert_eq!(
            sample_lifecycle_status(&path, &mut cursor),
            Some(CodexPeerStatus::Busy)
        );
        assert!(cursor.offset > first_offset);

        writeln!(
            file,
            r#"{{"type":"event_msg","payload":{{"type":"task_complete"}}}}"#
        )
        .unwrap();
        file.flush().unwrap();
        assert_eq!(
            sample_lifecycle_status(&path, &mut cursor),
            Some(CodexPeerStatus::Idle)
        );
        remove_file(path).unwrap();
    }

    #[test]
    fn open_rollout_requires_consecutive_misses_before_retirement() {
        let rollout = PathBuf::from("/tmp/session.jsonl");
        let mut misses = HashMap::new();

        let observed = reconcile_open_rollouts(&[], vec![rollout.clone()], &mut misses);
        assert_eq!(observed, vec![rollout.clone()]);

        let retained = reconcile_open_rollouts(&observed, vec![], &mut misses);
        assert_eq!(retained, vec![rollout.clone()]);
        assert_eq!(misses.get(&rollout), Some(&1));

        let recovered = reconcile_open_rollouts(&retained, vec![rollout.clone()], &mut misses);
        assert_eq!(recovered, vec![rollout.clone()]);
        assert!(misses.is_empty());

        let retained = reconcile_open_rollouts(&recovered, vec![], &mut misses);
        assert_eq!(retained, vec![rollout.clone()]);
        let retired = reconcile_open_rollouts(&retained, vec![], &mut misses);
        assert!(retired.is_empty());
        assert!(misses.is_empty());
    }
}