supercode_harness/
codex_peer.rs

1//! Live stock-Codex session discovery.
2//!
3//! Codex does not publish a peer registry or a supported attachment endpoint,
4//! but its process keeps every rollout it currently owns open. This module
5//! joins that process-owned file descriptor back to the persisted catalog
6//! path. The rollout's last explicit lifecycle event then distinguishes an
7//! executing turn from a merely running session. No timing or CPU heuristic
8//! is used.
9
10use std::collections::{HashMap, HashSet};
11use std::fs::File;
12use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
13use std::path::{Path, PathBuf};
14use std::time::{Duration, Instant};
15
16use serde_json::Value;
17
18const LIFECYCLE_SCAN_BYTES: u64 = 1024 * 1024;
19const LIFECYCLE_OVERLAP_BYTES: u64 = 64 * 1024;
20const OWNERSHIP_REFRESH_INTERVAL: Duration = Duration::from_secs(1);
21const OWNERSHIP_MISS_CONFIRMATIONS: u8 = 2;
22
23/// Activity proven for a rollout owned by stock Codex.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum CodexPeerStatus {
26    /// Codex owns the rollout, but no active turn is proven.
27    Running,
28    /// The latest lifecycle boundary completed or aborted a turn.
29    Idle,
30    /// The latest lifecycle boundary starts a task.
31    Busy,
32}
33
34impl CodexPeerStatus {
35    /// Stable wire spelling shared by the harness protocol.
36    pub const fn as_str(self) -> &'static str {
37        match self {
38            Self::Running => "running",
39            Self::Idle => "idle",
40            Self::Busy => "busy",
41        }
42    }
43}
44
45/// Every Codex rollout currently held open by a stock `codex` process, with
46/// the narrowest activity state its own event stream proves.
47pub fn live_rollouts(sessions_root: &Path) -> HashMap<PathBuf, CodexPeerStatus> {
48    CodexPeerTracker::default().sample(sessions_root)
49}
50
51/// Cached process-ownership sampler for latency-sensitive activity streams.
52///
53/// Process/file-descriptor discovery is materially more expensive than
54/// reading a bounded lifecycle tail. Ownership is therefore refreshed once a
55/// second while known open rollouts are re-read on every activity tick.
56#[derive(Debug, Default)]
57pub(crate) struct CodexPeerTracker {
58    root: Option<PathBuf>,
59    open_rollouts: Vec<PathBuf>,
60    ownership_misses: HashMap<PathBuf, u8>,
61    lifecycle: HashMap<PathBuf, CodexLifecycleCursor>,
62    refreshed_at: Option<Instant>,
63}
64
65#[derive(Debug, Default)]
66struct CodexLifecycleCursor {
67    offset: u64,
68    status: Option<CodexPeerStatus>,
69}
70
71impl CodexPeerTracker {
72    pub(crate) fn sample(&mut self, sessions_root: &Path) -> HashMap<PathBuf, CodexPeerStatus> {
73        let root = normalized_path(sessions_root);
74        let refresh = self.root.as_ref() != Some(&root)
75            || self
76                .refreshed_at
77                .is_none_or(|at| at.elapsed() >= OWNERSHIP_REFRESH_INTERVAL);
78        if refresh {
79            let observed = platform_open_rollouts()
80                .into_iter()
81                .map(|path| normalized_path(&path))
82                .collect();
83            self.open_rollouts =
84                reconcile_open_rollouts(&self.open_rollouts, observed, &mut self.ownership_misses);
85            self.root = Some(root.clone());
86            self.refreshed_at = Some(Instant::now());
87            self.lifecycle
88                .retain(|path, _| self.open_rollouts.contains(path));
89        }
90        let mut statuses = HashMap::new();
91        for path in &self.open_rollouts {
92            if !(path.starts_with(&root)
93                && path.extension().and_then(|extension| extension.to_str()) == Some("jsonl"))
94            {
95                continue;
96            }
97            let cursor = self.lifecycle.entry(path.clone()).or_default();
98            let status = sample_lifecycle_status(path, cursor).unwrap_or(CodexPeerStatus::Running);
99            statuses.insert(path.clone(), status);
100        }
101        statuses
102    }
103}
104
105/// One process-table sample is negative evidence, not a lifecycle boundary.
106/// Keep a previously open rollout through one miss; only consecutive misses
107/// retire it. This absorbs transient `pgrep`/`lsof` snapshots without putting
108/// a wall-clock guess into session state, while a genuinely exited process is
109/// removed on the next independent ownership observation.
110fn reconcile_open_rollouts(
111    previous: &[PathBuf],
112    observed: Vec<PathBuf>,
113    misses: &mut HashMap<PathBuf, u8>,
114) -> Vec<PathBuf> {
115    let observed = observed.into_iter().collect::<HashSet<_>>();
116    let previous = previous.iter().cloned().collect::<HashSet<_>>();
117    let mut reconciled = observed.clone();
118
119    for path in &observed {
120        misses.remove(path);
121    }
122    for path in previous.difference(&observed) {
123        let count = misses.entry(path.clone()).or_insert(0);
124        *count = count.saturating_add(1);
125        if *count < OWNERSHIP_MISS_CONFIRMATIONS {
126            reconciled.insert(path.clone());
127        } else {
128            misses.remove(path);
129        }
130    }
131    misses.retain(|path, _| previous.contains(path) && !observed.contains(path));
132
133    let mut reconciled = reconciled.into_iter().collect::<Vec<_>>();
134    reconciled.sort();
135    reconciled
136}
137
138/// Activity for a discovered catalog path owned by a currently running Codex.
139pub fn rollout_status(
140    live: &HashMap<PathBuf, CodexPeerStatus>,
141    path: &Path,
142) -> Option<CodexPeerStatus> {
143    live.get(&normalized_path(path)).copied()
144}
145
146/// Lightweight native identity and direct parent from Codex's first
147/// `session_meta` record. Activity aggregation uses this to treat a process-
148/// owned subagent rollout as work inside its root conversation.
149pub(crate) fn rollout_lineage(path: &Path) -> Option<(String, Option<String>)> {
150    let mut header = String::new();
151    BufReader::new(File::open(path).ok()?.take(256 * 1024))
152        .read_line(&mut header)
153        .ok()?;
154    let value = serde_json::from_str::<Value>(&header).ok()?;
155    if value.get("type").and_then(Value::as_str) != Some("session_meta") {
156        return None;
157    }
158    let payload = value.get("payload")?;
159    let session_id = payload.get("id")?.as_str()?.to_string();
160    let parent_session_id = payload
161        .pointer("/source/subagent/thread_spawn/parent_thread_id")
162        .or_else(|| payload.get("parent_thread_id"))
163        .and_then(Value::as_str)
164        .map(str::to_string);
165    Some((session_id, parent_session_id))
166}
167
168fn sample_lifecycle_status(
169    path: &Path,
170    cursor: &mut CodexLifecycleCursor,
171) -> Option<CodexPeerStatus> {
172    let mut file = File::open(path).ok()?;
173    let length = file.metadata().ok()?.len();
174    if length < cursor.offset {
175        cursor.offset = 0;
176        cursor.status = None;
177    }
178    if cursor.offset == 0 {
179        cursor.status = latest_lifecycle_status_between(&mut file, 0, length);
180    } else if length > cursor.offset {
181        if let Some(status) = latest_lifecycle_status_between(&mut file, cursor.offset, length) {
182            cursor.status = Some(status);
183        }
184    }
185    cursor.offset = length;
186    cursor.status
187}
188
189fn latest_lifecycle_status_between(
190    file: &mut File,
191    floor: u64,
192    upper: u64,
193) -> Option<CodexPeerStatus> {
194    let mut end = upper;
195    while end > floor {
196        let start = end.saturating_sub(LIFECYCLE_SCAN_BYTES).max(floor);
197        file.seek(SeekFrom::Start(start)).ok()?;
198        let mut tail = vec![0; (end - start) as usize];
199        file.read_exact(&mut tail).ok()?;
200        if let Some(status) = lifecycle_status_in_tail(&tail, start == floor) {
201            return Some(status);
202        }
203        if start == floor {
204            break;
205        }
206        // A lifecycle record is small, but an adjacent tool record can be enormous. Overlap keeps
207        // a boundary record whole without ever allocating in proportion to the rollout.
208        end = start.saturating_add(LIFECYCLE_OVERLAP_BYTES);
209    }
210    None
211}
212
213fn lifecycle_status_in_tail(
214    tail: &[u8],
215    starts_at_record_boundary: bool,
216) -> Option<CodexPeerStatus> {
217    const BOUNDARIES: [&str; 3] = [
218        "\"type\":\"task_started\"",
219        "\"type\":\"task_complete\"",
220        "\"type\":\"turn_aborted\"",
221    ];
222
223    // The read can begin in the middle of a large UTF-8 JSON string. Ignore
224    // that first fragment, then use the standard library's substring search
225    // to jump directly between lifecycle candidates instead of inspecting
226    // every byte of every tool payload with a naive sliding window.
227    let complete_start = if starts_at_record_boundary {
228        0
229    } else {
230        tail.iter()
231            .position(|byte| *byte == b'\n')
232            .map_or(tail.len(), |newline| newline + 1)
233    };
234    let text = std::str::from_utf8(&tail[complete_start..]).ok()?;
235    let mut search_end = text.len();
236    while let Some(candidate) = BOUNDARIES
237        .iter()
238        .filter_map(|boundary| text[..search_end].rfind(boundary))
239        .max()
240    {
241        let line_start = text[..candidate]
242            .rfind('\n')
243            .map_or(0, |newline| newline + 1);
244        let line_end = text[candidate..]
245            .find('\n')
246            .map_or(text.len(), |newline| candidate + newline);
247        let Ok(event) = serde_json::from_str::<Value>(&text[line_start..line_end]) else {
248            search_end = candidate;
249            continue;
250        };
251        if event.get("type").and_then(Value::as_str) != Some("event_msg") {
252            search_end = candidate;
253            continue;
254        }
255        match event
256            .get("payload")
257            .and_then(|payload| payload.get("type"))
258            .and_then(Value::as_str)
259        {
260            Some("task_started") => return Some(CodexPeerStatus::Busy),
261            Some("task_complete" | "turn_aborted") => return Some(CodexPeerStatus::Idle),
262            _ => {}
263        }
264        search_end = candidate;
265    }
266    None
267}
268
269fn normalized_path(path: &Path) -> PathBuf {
270    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
271}
272
273#[cfg(target_os = "macos")]
274fn platform_open_rollouts() -> Vec<PathBuf> {
275    macos_open_rollouts_native().unwrap_or_else(macos_open_rollouts_with_commands)
276}
277
278#[cfg(target_os = "macos")]
279fn macos_open_rollouts_with_commands() -> Vec<PathBuf> {
280    use std::process::Command;
281
282    // Darwin's pgrep omits every ancestor of the caller unless `-a` is set.
283    // Discovery commonly runs underneath the very Codex session it must
284    // report (for example inside a Supercode-powered widget), so omitting
285    // ancestors makes the current session uniquely invisible.
286    let Ok(processes) = Command::new("/usr/bin/pgrep")
287        .args(["-a", "-x", "codex"])
288        .output()
289    else {
290        return Vec::new();
291    };
292    let pids = String::from_utf8_lossy(&processes.stdout)
293        .lines()
294        .filter_map(|line| line.trim().parse::<u32>().ok())
295        .take(128)
296        .map(|pid| pid.to_string())
297        .collect::<Vec<_>>();
298    if pids.is_empty() {
299        return Vec::new();
300    }
301    let Ok(files) = Command::new("/usr/sbin/lsof")
302        .args(["-Fn", "-a", "-p", &pids.join(",")])
303        .output()
304    else {
305        return Vec::new();
306    };
307    String::from_utf8_lossy(&files.stdout)
308        .lines()
309        .filter_map(|line| line.strip_prefix('n'))
310        .filter(|path| path.ends_with(".jsonl"))
311        .map(PathBuf::from)
312        .collect()
313}
314
315#[cfg(target_os = "macos")]
316fn macos_open_rollouts_native() -> Option<Vec<PathBuf>> {
317    use std::mem::size_of;
318
319    const PROCESS_NAME_BYTES: usize = 64;
320    const INITIAL_PID_CAPACITY: usize = 2_048;
321
322    let mut pids = vec![0_i32; INITIAL_PID_CAPACITY];
323    let count = loop {
324        let count = unsafe {
325            proc_listallpids(
326                pids.as_mut_ptr().cast(),
327                i32::try_from(pids.len() * size_of::<i32>()).ok()?,
328            )
329        };
330        if count <= 0 {
331            return None;
332        }
333        if usize::try_from(count).ok()? < pids.len() {
334            break count;
335        }
336        pids.resize(pids.len() * 2, 0);
337    };
338    pids.truncate(usize::try_from(count).ok()?.min(pids.len()));
339
340    let codex_pids = pids
341        .into_iter()
342        .filter(|pid| *pid > 0)
343        .filter(|pid| {
344            let mut name = [0_u8; PROCESS_NAME_BYTES];
345            let length = unsafe {
346                proc_name(
347                    *pid,
348                    name.as_mut_ptr().cast(),
349                    u32::try_from(name.len()).expect("small process-name buffer"),
350                )
351            };
352            usize::try_from(length)
353                .ok()
354                .and_then(|length| name.get(..length))
355                == Some(b"codex".as_slice())
356        })
357        .collect::<Vec<_>>();
358    if codex_pids.is_empty() {
359        return Some(Vec::new());
360    }
361
362    macos_open_jsonl_for_pids(&codex_pids)
363}
364
365#[cfg(target_os = "macos")]
366fn macos_open_jsonl_for_pids(pids: &[i32]) -> Option<Vec<PathBuf>> {
367    use std::mem::{size_of, MaybeUninit};
368    use std::os::unix::ffi::OsStrExt;
369
370    const INITIAL_FD_CAPACITY: usize = 256;
371    const MAX_FD_CAPACITY: usize = 65_536;
372    const PROC_PIDLISTFDS: i32 = 1;
373    const PROC_PIDFDVNODEPATHINFO: i32 = 2;
374    const PROX_FDTYPE_VNODE: u32 = 1;
375
376    let mut successful_fd_reads = 0_usize;
377    let mut failed_fd_reads = 0_usize;
378    let mut paths = Vec::new();
379    for pid in pids.iter().copied() {
380        let mut capacity = INITIAL_FD_CAPACITY;
381        let descriptors = loop {
382            let mut descriptors = vec![ProcFdInfo::default(); capacity];
383            let bytes = unsafe {
384                proc_pidinfo(
385                    pid,
386                    PROC_PIDLISTFDS,
387                    0,
388                    descriptors.as_mut_ptr().cast(),
389                    i32::try_from(descriptors.len() * size_of::<ProcFdInfo>()).ok()?,
390                )
391            };
392            if bytes <= 0 {
393                failed_fd_reads += 1;
394                break None;
395            }
396            let bytes = usize::try_from(bytes).ok()?;
397            if bytes < descriptors.len() * size_of::<ProcFdInfo>() {
398                descriptors.truncate(bytes / size_of::<ProcFdInfo>());
399                break Some(descriptors);
400            }
401            if capacity >= MAX_FD_CAPACITY {
402                return None;
403            }
404            capacity = (capacity * 2).min(MAX_FD_CAPACITY);
405        };
406        let Some(descriptors) = descriptors else {
407            continue;
408        };
409        successful_fd_reads += 1;
410        for descriptor in descriptors
411            .into_iter()
412            .filter(|descriptor| descriptor.proc_fdtype == PROX_FDTYPE_VNODE)
413        {
414            let mut info = MaybeUninit::<VnodeFdInfoWithPath>::zeroed();
415            let bytes = unsafe {
416                proc_pidfdinfo(
417                    pid,
418                    descriptor.proc_fd,
419                    PROC_PIDFDVNODEPATHINFO,
420                    info.as_mut_ptr().cast(),
421                    i32::try_from(size_of::<VnodeFdInfoWithPath>()).expect("fixed native struct"),
422                )
423            };
424            if usize::try_from(bytes).ok() != Some(size_of::<VnodeFdInfoWithPath>()) {
425                continue;
426            }
427            let info = unsafe { info.assume_init() };
428            let path_length = info
429                .pvip
430                .vip_path
431                .iter()
432                .position(|byte| *byte == 0)
433                .unwrap_or(info.pvip.vip_path.len());
434            let path = PathBuf::from(std::ffi::OsStr::from_bytes(
435                &info.pvip.vip_path[..path_length],
436            ));
437            if path.extension().and_then(|extension| extension.to_str()) == Some("jsonl") {
438                paths.push(path);
439            }
440        }
441    }
442    if successful_fd_reads == 0 || failed_fd_reads != 0 {
443        return None;
444    }
445    paths.sort();
446    paths.dedup();
447    Some(paths)
448}
449
450#[cfg(target_os = "macos")]
451#[repr(C)]
452#[derive(Debug, Clone, Copy, Default)]
453struct ProcFdInfo {
454    proc_fd: i32,
455    proc_fdtype: u32,
456}
457
458#[cfg(target_os = "macos")]
459#[repr(C)]
460struct ProcFileInfo {
461    fi_openflags: u32,
462    fi_status: u32,
463    fi_offset: i64,
464    fi_type: i32,
465    fi_guardflags: u32,
466}
467
468#[cfg(target_os = "macos")]
469#[repr(C)]
470struct VinfoStat {
471    vst_dev: u32,
472    vst_mode: u16,
473    vst_nlink: u16,
474    vst_ino: u64,
475    vst_uid: u32,
476    vst_gid: u32,
477    vst_atime: i64,
478    vst_atimensec: i64,
479    vst_mtime: i64,
480    vst_mtimensec: i64,
481    vst_ctime: i64,
482    vst_ctimensec: i64,
483    vst_birthtime: i64,
484    vst_birthtimensec: i64,
485    vst_size: i64,
486    vst_blocks: i64,
487    vst_blksize: i32,
488    vst_flags: u32,
489    vst_gen: u32,
490    vst_rdev: u32,
491    vst_qspare: [i64; 2],
492}
493
494#[cfg(target_os = "macos")]
495#[repr(C)]
496struct VnodeInfo {
497    vi_stat: VinfoStat,
498    vi_type: i32,
499    vi_pad: i32,
500    vi_fsid: [i32; 2],
501}
502
503#[cfg(target_os = "macos")]
504#[repr(C)]
505struct VnodeInfoPath {
506    vip_vi: VnodeInfo,
507    vip_path: [u8; 1_024],
508}
509
510#[cfg(target_os = "macos")]
511#[repr(C)]
512struct VnodeFdInfoWithPath {
513    pfi: ProcFileInfo,
514    pvip: VnodeInfoPath,
515}
516
517#[cfg(target_os = "macos")]
518#[link(name = "proc")]
519unsafe extern "C" {
520    fn proc_listallpids(buffer: *mut std::ffi::c_void, buffersize: i32) -> i32;
521    fn proc_pidinfo(
522        pid: i32,
523        flavor: i32,
524        arg: u64,
525        buffer: *mut std::ffi::c_void,
526        buffersize: i32,
527    ) -> i32;
528    fn proc_pidfdinfo(
529        pid: i32,
530        fd: i32,
531        flavor: i32,
532        buffer: *mut std::ffi::c_void,
533        buffersize: i32,
534    ) -> i32;
535    fn proc_name(pid: i32, buffer: *mut std::ffi::c_void, buffersize: u32) -> i32;
536}
537
538#[cfg(target_os = "linux")]
539fn platform_open_rollouts() -> Vec<PathBuf> {
540    let Ok(processes) = std::fs::read_dir("/proc") else {
541        return Vec::new();
542    };
543    let mut paths = Vec::new();
544    for process in processes.flatten() {
545        let pid = process.file_name();
546        if !pid.as_encoded_bytes().iter().all(u8::is_ascii_digit) {
547            continue;
548        }
549        let process_root = process.path();
550        if std::fs::read_to_string(process_root.join("comm"))
551            .ok()
552            .is_none_or(|name| name.trim() != "codex")
553        {
554            continue;
555        }
556        let Ok(descriptors) = std::fs::read_dir(process_root.join("fd")) else {
557            continue;
558        };
559        paths.extend(
560            descriptors
561                .flatten()
562                .filter_map(|descriptor| std::fs::read_link(descriptor.path()).ok())
563                .filter(|path| {
564                    path.extension().and_then(|extension| extension.to_str()) == Some("jsonl")
565                }),
566        );
567    }
568    paths
569}
570
571#[cfg(not(any(target_os = "macos", target_os = "linux")))]
572fn platform_open_rollouts() -> Vec<PathBuf> {
573    Vec::new()
574}
575
576#[cfg(test)]
577mod tests {
578    use std::fs::{remove_file, OpenOptions};
579    use std::io::Write;
580
581    use super::*;
582
583    #[cfg(target_os = "macos")]
584    #[test]
585    fn native_macos_fd_layout_and_open_jsonl_discovery_match_the_sdk() {
586        assert_eq!(std::mem::size_of::<ProcFdInfo>(), 8);
587        assert_eq!(std::mem::size_of::<ProcFileInfo>(), 24);
588        assert_eq!(std::mem::size_of::<VnodeInfo>(), 152);
589        assert_eq!(std::mem::size_of::<VnodeFdInfoWithPath>(), 1_200);
590
591        let path = std::env::temp_dir().join(format!(
592            "supercode-native-open-rollout-{}.jsonl",
593            std::process::id()
594        ));
595        let file = File::create(&path).unwrap();
596        let open = macos_open_jsonl_for_pids(&[std::process::id() as i32])
597            .expect("the current process's descriptor table should be readable");
598        assert!(open.contains(&normalized_path(&path)));
599        drop(file);
600        remove_file(path).unwrap();
601    }
602
603    #[test]
604    fn long_tool_heavy_turn_is_found_once_then_followed_incrementally() {
605        let path = std::env::temp_dir().join(format!(
606            "supercode-codex-long-turn-{}-{}.jsonl",
607            std::process::id(),
608            std::thread::current().name().unwrap_or("test")
609        ));
610        let mut file = File::create(&path).unwrap();
611        writeln!(
612            file,
613            r#"{{"type":"event_msg","payload":{{"type":"task_started"}}}}"#
614        )
615        .unwrap();
616        write!(
617            file,
618            r#"{{"type":"response_item","payload":"{}"}}"#,
619            "x".repeat(6 * 1024 * 1024)
620        )
621        .unwrap();
622        writeln!(file).unwrap();
623        file.flush().unwrap();
624
625        let mut cursor = CodexLifecycleCursor::default();
626        assert_eq!(
627            sample_lifecycle_status(&path, &mut cursor),
628            Some(CodexPeerStatus::Busy)
629        );
630        let first_offset = cursor.offset;
631
632        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
633        writeln!(
634            file,
635            r#"{{"type":"event_msg","payload":{{"type":"item_completed"}}}}"#
636        )
637        .unwrap();
638        file.flush().unwrap();
639        assert_eq!(
640            sample_lifecycle_status(&path, &mut cursor),
641            Some(CodexPeerStatus::Busy)
642        );
643        assert!(cursor.offset > first_offset);
644
645        writeln!(
646            file,
647            r#"{{"type":"event_msg","payload":{{"type":"task_complete"}}}}"#
648        )
649        .unwrap();
650        file.flush().unwrap();
651        assert_eq!(
652            sample_lifecycle_status(&path, &mut cursor),
653            Some(CodexPeerStatus::Idle)
654        );
655        remove_file(path).unwrap();
656    }
657
658    #[test]
659    fn open_rollout_requires_consecutive_misses_before_retirement() {
660        let rollout = PathBuf::from("/tmp/session.jsonl");
661        let mut misses = HashMap::new();
662
663        let observed = reconcile_open_rollouts(&[], vec![rollout.clone()], &mut misses);
664        assert_eq!(observed, vec![rollout.clone()]);
665
666        let retained = reconcile_open_rollouts(&observed, vec![], &mut misses);
667        assert_eq!(retained, vec![rollout.clone()]);
668        assert_eq!(misses.get(&rollout), Some(&1));
669
670        let recovered = reconcile_open_rollouts(&retained, vec![rollout.clone()], &mut misses);
671        assert_eq!(recovered, vec![rollout.clone()]);
672        assert!(misses.is_empty());
673
674        let retained = reconcile_open_rollouts(&recovered, vec![], &mut misses);
675        assert_eq!(retained, vec![rollout.clone()]);
676        let retired = reconcile_open_rollouts(&retained, vec![], &mut misses);
677        assert!(retired.is_empty());
678        assert!(misses.is_empty());
679    }
680}
supercode_harness/codex_peer.rs

supercode_harness/
codex_peer.rs
