supercode_harness/

codex_peer.rs

//! 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()
}

/// A rollout's session id and, for a subagent thread, its parent's id.
pub fn rollout_session(path: &Path) -> Option<(String, Option<String>)> {
    rollout_lineage(path)
}

/// The working directory a rollout's conversation runs in, from its first
/// `session_meta` record.
pub fn rollout_cwd(path: &Path) -> Option<PathBuf> {
    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()?;
    value
        .pointer("/payload/cwd")
        .and_then(Value::as_str)
        .map(PathBuf::from)
}

/// 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> {
    macos_open_rollouts_native().unwrap_or_else(macos_open_rollouts_with_commands)
}

#[cfg(target_os = "macos")]
fn macos_open_rollouts_with_commands() -> 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 = "macos")]
fn macos_open_rollouts_native() -> Option<Vec<PathBuf>> {
    use std::mem::size_of;

    const PROCESS_NAME_BYTES: usize = 64;
    const INITIAL_PID_CAPACITY: usize = 2_048;

    let mut pids = vec![0_i32; INITIAL_PID_CAPACITY];
    let count = loop {
        let count = unsafe {
            proc_listallpids(
                pids.as_mut_ptr().cast(),
                i32::try_from(pids.len() * size_of::<i32>()).ok()?,
            )
        };
        if count <= 0 {
            return None;
        }
        if usize::try_from(count).ok()? < pids.len() {
            break count;
        }
        pids.resize(pids.len() * 2, 0);
    };
    pids.truncate(usize::try_from(count).ok()?.min(pids.len()));

    let codex_pids = pids
        .into_iter()
        .filter(|pid| *pid > 0)
        .filter(|pid| {
            let mut name = [0_u8; PROCESS_NAME_BYTES];
            let length = unsafe {
                proc_name(
                    *pid,
                    name.as_mut_ptr().cast(),
                    u32::try_from(name.len()).expect("small process-name buffer"),
                )
            };
            usize::try_from(length)
                .ok()
                .and_then(|length| name.get(..length))
                == Some(b"codex".as_slice())
        })
        .collect::<Vec<_>>();
    if codex_pids.is_empty() {
        return Some(Vec::new());
    }

    macos_open_jsonl_for_pids(&codex_pids)
}

#[cfg(target_os = "macos")]
fn macos_open_jsonl_for_pids(pids: &[i32]) -> Option<Vec<PathBuf>> {
    use std::mem::{size_of, MaybeUninit};
    use std::os::unix::ffi::OsStrExt;

    const INITIAL_FD_CAPACITY: usize = 256;
    const MAX_FD_CAPACITY: usize = 65_536;
    const PROC_PIDLISTFDS: i32 = 1;
    const PROC_PIDFDVNODEPATHINFO: i32 = 2;
    const PROX_FDTYPE_VNODE: u32 = 1;

    let mut successful_fd_reads = 0_usize;
    let mut failed_fd_reads = 0_usize;
    let mut paths = Vec::new();
    for pid in pids.iter().copied() {
        let mut capacity = INITIAL_FD_CAPACITY;
        let descriptors = loop {
            let mut descriptors = vec![ProcFdInfo::default(); capacity];
            let bytes = unsafe {
                proc_pidinfo(
                    pid,
                    PROC_PIDLISTFDS,
                    0,
                    descriptors.as_mut_ptr().cast(),
                    i32::try_from(descriptors.len() * size_of::<ProcFdInfo>()).ok()?,
                )
            };
            if bytes <= 0 {
                failed_fd_reads += 1;
                break None;
            }
            let bytes = usize::try_from(bytes).ok()?;
            if bytes < descriptors.len() * size_of::<ProcFdInfo>() {
                descriptors.truncate(bytes / size_of::<ProcFdInfo>());
                break Some(descriptors);
            }
            if capacity >= MAX_FD_CAPACITY {
                return None;
            }
            capacity = (capacity * 2).min(MAX_FD_CAPACITY);
        };
        let Some(descriptors) = descriptors else {
            continue;
        };
        successful_fd_reads += 1;
        for descriptor in descriptors
            .into_iter()
            .filter(|descriptor| descriptor.proc_fdtype == PROX_FDTYPE_VNODE)
        {
            let mut info = MaybeUninit::<VnodeFdInfoWithPath>::zeroed();
            let bytes = unsafe {
                proc_pidfdinfo(
                    pid,
                    descriptor.proc_fd,
                    PROC_PIDFDVNODEPATHINFO,
                    info.as_mut_ptr().cast(),
                    i32::try_from(size_of::<VnodeFdInfoWithPath>()).expect("fixed native struct"),
                )
            };
            if usize::try_from(bytes).ok() != Some(size_of::<VnodeFdInfoWithPath>()) {
                continue;
            }
            let info = unsafe { info.assume_init() };
            let path_length = info
                .pvip
                .vip_path
                .iter()
                .position(|byte| *byte == 0)
                .unwrap_or(info.pvip.vip_path.len());
            let path = PathBuf::from(std::ffi::OsStr::from_bytes(
                &info.pvip.vip_path[..path_length],
            ));
            if path.extension().and_then(|extension| extension.to_str()) == Some("jsonl") {
                paths.push(path);
            }
        }
    }
    if successful_fd_reads == 0 || failed_fd_reads != 0 {
        return None;
    }
    paths.sort();
    paths.dedup();
    Some(paths)
}

#[cfg(target_os = "macos")]
#[repr(C)]
#[derive(Debug, Clone, Copy, Default)]
struct ProcFdInfo {
    proc_fd: i32,
    proc_fdtype: u32,
}

#[cfg(target_os = "macos")]
#[repr(C)]
struct ProcFileInfo {
    fi_openflags: u32,
    fi_status: u32,
    fi_offset: i64,
    fi_type: i32,
    fi_guardflags: u32,
}

#[cfg(target_os = "macos")]
#[repr(C)]
struct VinfoStat {
    vst_dev: u32,
    vst_mode: u16,
    vst_nlink: u16,
    vst_ino: u64,
    vst_uid: u32,
    vst_gid: u32,
    vst_atime: i64,
    vst_atimensec: i64,
    vst_mtime: i64,
    vst_mtimensec: i64,
    vst_ctime: i64,
    vst_ctimensec: i64,
    vst_birthtime: i64,
    vst_birthtimensec: i64,
    vst_size: i64,
    vst_blocks: i64,
    vst_blksize: i32,
    vst_flags: u32,
    vst_gen: u32,
    vst_rdev: u32,
    vst_qspare: [i64; 2],
}

#[cfg(target_os = "macos")]
#[repr(C)]
struct VnodeInfo {
    vi_stat: VinfoStat,
    vi_type: i32,
    vi_pad: i32,
    vi_fsid: [i32; 2],
}

#[cfg(target_os = "macos")]
#[repr(C)]
struct VnodeInfoPath {
    vip_vi: VnodeInfo,
    vip_path: [u8; 1_024],
}

#[cfg(target_os = "macos")]
#[repr(C)]
struct VnodeFdInfoWithPath {
    pfi: ProcFileInfo,
    pvip: VnodeInfoPath,
}

#[cfg(target_os = "macos")]
#[link(name = "proc")]
unsafe extern "C" {
    fn proc_listallpids(buffer: *mut std::ffi::c_void, buffersize: i32) -> i32;
    fn proc_pidinfo(
        pid: i32,
        flavor: i32,
        arg: u64,
        buffer: *mut std::ffi::c_void,
        buffersize: i32,
    ) -> i32;
    fn proc_pidfdinfo(
        pid: i32,
        fd: i32,
        flavor: i32,
        buffer: *mut std::ffi::c_void,
        buffersize: i32,
    ) -> i32;
    fn proc_name(pid: i32, buffer: *mut std::ffi::c_void, buffersize: u32) -> i32;
}

#[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()
}

/// The Codex session a process runs: the root rollout (one with no parent
/// thread) among the files `pid` holds open, as `(session id, rollout path)`.
/// `None` when `pid` holds no Codex rollout.
///
/// This is how a command run from a Codex shell learns which session ran it:
/// its ancestry reaches the `codex` process, and that process owns exactly
/// the rollout of its conversation (subagent rollouts name their parent).
///
/// Codex's shared app-server daemon holds the rollouts of every thread it has
/// loaded; a process holding more than one conversation names none of them.
pub fn session_of_process(pid: u32) -> Option<(String, PathBuf)> {
    let mut roots: Vec<(String, PathBuf)> = Vec::new();
    let mut child = None;
    for path in rollouts_open_by(pid) {
        let Some((session_id, parent)) = rollout_lineage(&path) else {
            continue;
        };
        if parent.is_none() {
            if !roots.iter().any(|(id, _)| *id == session_id) {
                roots.push((session_id, path));
            }
        } else {
            child.get_or_insert((session_id, path));
        }
    }
    match roots.len() {
        0 => child,
        1 => roots.pop(),
        _ => None,
    }
}

/// Whether `pid` holds the rollout of Codex session `session_id` open.
pub fn holds_session(pid: u32, session_id: &str) -> bool {
    rollouts_open_by(pid)
        .iter()
        .filter_map(|path| rollout_lineage(path))
        .any(|(id, _)| id == session_id)
}

#[cfg(target_os = "macos")]
fn rollouts_open_by(pid: u32) -> Vec<PathBuf> {
    i32::try_from(pid)
        .ok()
        .and_then(|pid| macos_open_jsonl_for_pids(&[pid]))
        .unwrap_or_default()
}

#[cfg(target_os = "linux")]
fn rollouts_open_by(pid: u32) -> Vec<PathBuf> {
    let Ok(descriptors) = std::fs::read_dir(format!("/proc/{pid}/fd")) else {
        return Vec::new();
    };
    descriptors
        .flatten()
        .filter_map(|descriptor| std::fs::read_link(descriptor.path()).ok())
        .filter(|path| path.extension().and_then(|extension| extension.to_str()) == Some("jsonl"))
        .collect()
}

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