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    use std::process::Command;
276
277    // Darwin's pgrep omits every ancestor of the caller unless `-a` is set.
278    // Discovery commonly runs underneath the very Codex session it must
279    // report (for example inside a Supercode-powered widget), so omitting
280    // ancestors makes the current session uniquely invisible.
281    let Ok(processes) = Command::new("/usr/bin/pgrep")
282        .args(["-a", "-x", "codex"])
283        .output()
284    else {
285        return Vec::new();
286    };
287    let pids = String::from_utf8_lossy(&processes.stdout)
288        .lines()
289        .filter_map(|line| line.trim().parse::<u32>().ok())
290        .take(128)
291        .map(|pid| pid.to_string())
292        .collect::<Vec<_>>();
293    if pids.is_empty() {
294        return Vec::new();
295    }
296    let Ok(files) = Command::new("/usr/sbin/lsof")
297        .args(["-Fn", "-a", "-p", &pids.join(",")])
298        .output()
299    else {
300        return Vec::new();
301    };
302    String::from_utf8_lossy(&files.stdout)
303        .lines()
304        .filter_map(|line| line.strip_prefix('n'))
305        .filter(|path| path.ends_with(".jsonl"))
306        .map(PathBuf::from)
307        .collect()
308}
309
310#[cfg(target_os = "linux")]
311fn platform_open_rollouts() -> Vec<PathBuf> {
312    let Ok(processes) = std::fs::read_dir("/proc") else {
313        return Vec::new();
314    };
315    let mut paths = Vec::new();
316    for process in processes.flatten() {
317        let pid = process.file_name();
318        if !pid.as_encoded_bytes().iter().all(u8::is_ascii_digit) {
319            continue;
320        }
321        let process_root = process.path();
322        if std::fs::read_to_string(process_root.join("comm"))
323            .ok()
324            .is_none_or(|name| name.trim() != "codex")
325        {
326            continue;
327        }
328        let Ok(descriptors) = std::fs::read_dir(process_root.join("fd")) else {
329            continue;
330        };
331        paths.extend(
332            descriptors
333                .flatten()
334                .filter_map(|descriptor| std::fs::read_link(descriptor.path()).ok())
335                .filter(|path| {
336                    path.extension().and_then(|extension| extension.to_str()) == Some("jsonl")
337                }),
338        );
339    }
340    paths
341}
342
343#[cfg(not(any(target_os = "macos", target_os = "linux")))]
344fn platform_open_rollouts() -> Vec<PathBuf> {
345    Vec::new()
346}
347
348#[cfg(test)]
349mod tests {
350    use std::fs::{remove_file, OpenOptions};
351    use std::io::Write;
352
353    use super::*;
354
355    #[test]
356    fn long_tool_heavy_turn_is_found_once_then_followed_incrementally() {
357        let path = std::env::temp_dir().join(format!(
358            "supercode-codex-long-turn-{}-{}.jsonl",
359            std::process::id(),
360            std::thread::current().name().unwrap_or("test")
361        ));
362        let mut file = File::create(&path).unwrap();
363        writeln!(
364            file,
365            r#"{{"type":"event_msg","payload":{{"type":"task_started"}}}}"#
366        )
367        .unwrap();
368        write!(
369            file,
370            r#"{{"type":"response_item","payload":"{}"}}"#,
371            "x".repeat(6 * 1024 * 1024)
372        )
373        .unwrap();
374        writeln!(file).unwrap();
375        file.flush().unwrap();
376
377        let mut cursor = CodexLifecycleCursor::default();
378        assert_eq!(
379            sample_lifecycle_status(&path, &mut cursor),
380            Some(CodexPeerStatus::Busy)
381        );
382        let first_offset = cursor.offset;
383
384        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
385        writeln!(
386            file,
387            r#"{{"type":"event_msg","payload":{{"type":"item_completed"}}}}"#
388        )
389        .unwrap();
390        file.flush().unwrap();
391        assert_eq!(
392            sample_lifecycle_status(&path, &mut cursor),
393            Some(CodexPeerStatus::Busy)
394        );
395        assert!(cursor.offset > first_offset);
396
397        writeln!(
398            file,
399            r#"{{"type":"event_msg","payload":{{"type":"task_complete"}}}}"#
400        )
401        .unwrap();
402        file.flush().unwrap();
403        assert_eq!(
404            sample_lifecycle_status(&path, &mut cursor),
405            Some(CodexPeerStatus::Idle)
406        );
407        remove_file(path).unwrap();
408    }
409
410    #[test]
411    fn open_rollout_requires_consecutive_misses_before_retirement() {
412        let rollout = PathBuf::from("/tmp/session.jsonl");
413        let mut misses = HashMap::new();
414
415        let observed = reconcile_open_rollouts(&[], vec![rollout.clone()], &mut misses);
416        assert_eq!(observed, vec![rollout.clone()]);
417
418        let retained = reconcile_open_rollouts(&observed, vec![], &mut misses);
419        assert_eq!(retained, vec![rollout.clone()]);
420        assert_eq!(misses.get(&rollout), Some(&1));
421
422        let recovered = reconcile_open_rollouts(&retained, vec![rollout.clone()], &mut misses);
423        assert_eq!(recovered, vec![rollout.clone()]);
424        assert!(misses.is_empty());
425
426        let retained = reconcile_open_rollouts(&recovered, vec![], &mut misses);
427        assert_eq!(retained, vec![rollout.clone()]);
428        let retired = reconcile_open_rollouts(&retained, vec![], &mut misses);
429        assert!(retired.is_empty());
430        assert!(misses.is_empty());
431    }
432}
supercode_harness/codex_peer.rs

supercode_harness/
codex_peer.rs
