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;
11use std::fs::File;
12use std::io::{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);
21
22/// Activity proven for a rollout owned by stock Codex.
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub enum CodexPeerStatus {
25    /// Codex owns the rollout, but no active turn is proven.
26    Running,
27    /// The latest lifecycle boundary completed or aborted a turn.
28    Idle,
29    /// The latest lifecycle boundary starts a task.
30    Busy,
31}
32
33impl CodexPeerStatus {
34    /// Stable wire spelling shared by the harness protocol.
35    pub const fn as_str(self) -> &'static str {
36        match self {
37            Self::Running => "running",
38            Self::Idle => "idle",
39            Self::Busy => "busy",
40        }
41    }
42}
43
44/// Every Codex rollout currently held open by a stock `codex` process, with
45/// the narrowest activity state its own event stream proves.
46pub fn live_rollouts(sessions_root: &Path) -> HashMap<PathBuf, CodexPeerStatus> {
47    CodexPeerTracker::default().sample(sessions_root)
48}
49
50/// Cached process-ownership sampler for latency-sensitive activity streams.
51///
52/// Process/file-descriptor discovery is materially more expensive than
53/// reading a bounded lifecycle tail. Ownership is therefore refreshed once a
54/// second while known open rollouts are re-read on every activity tick.
55#[derive(Debug, Default)]
56pub(crate) struct CodexPeerTracker {
57    root: Option<PathBuf>,
58    open_rollouts: Vec<PathBuf>,
59    lifecycle: HashMap<PathBuf, CodexLifecycleCursor>,
60    refreshed_at: Option<Instant>,
61}
62
63#[derive(Debug, Default)]
64struct CodexLifecycleCursor {
65    offset: u64,
66    status: Option<CodexPeerStatus>,
67}
68
69impl CodexPeerTracker {
70    pub(crate) fn sample(&mut self, sessions_root: &Path) -> HashMap<PathBuf, CodexPeerStatus> {
71        let root = normalized_path(sessions_root);
72        let refresh = self.root.as_ref() != Some(&root)
73            || self
74                .refreshed_at
75                .is_none_or(|at| at.elapsed() >= OWNERSHIP_REFRESH_INTERVAL);
76        if refresh {
77            self.open_rollouts = platform_open_rollouts()
78                .into_iter()
79                .map(|path| normalized_path(&path))
80                .collect();
81            self.root = Some(root.clone());
82            self.refreshed_at = Some(Instant::now());
83            self.lifecycle
84                .retain(|path, _| self.open_rollouts.contains(path));
85        }
86        let mut statuses = HashMap::new();
87        for path in &self.open_rollouts {
88            if !(path.starts_with(&root)
89                && path.extension().and_then(|extension| extension.to_str()) == Some("jsonl"))
90            {
91                continue;
92            }
93            let cursor = self.lifecycle.entry(path.clone()).or_default();
94            let status = sample_lifecycle_status(path, cursor).unwrap_or(CodexPeerStatus::Running);
95            statuses.insert(path.clone(), status);
96        }
97        statuses
98    }
99}
100
101/// Activity for a discovered catalog path owned by a currently running Codex.
102pub fn rollout_status(
103    live: &HashMap<PathBuf, CodexPeerStatus>,
104    path: &Path,
105) -> Option<CodexPeerStatus> {
106    live.get(&normalized_path(path)).copied()
107}
108
109fn sample_lifecycle_status(
110    path: &Path,
111    cursor: &mut CodexLifecycleCursor,
112) -> Option<CodexPeerStatus> {
113    let mut file = File::open(path).ok()?;
114    let length = file.metadata().ok()?.len();
115    if length < cursor.offset {
116        cursor.offset = 0;
117        cursor.status = None;
118    }
119    if cursor.offset == 0 {
120        cursor.status = latest_lifecycle_status_between(&mut file, 0, length);
121    } else if length > cursor.offset {
122        if let Some(status) = latest_lifecycle_status_between(&mut file, cursor.offset, length) {
123            cursor.status = Some(status);
124        }
125    }
126    cursor.offset = length;
127    cursor.status
128}
129
130fn latest_lifecycle_status_between(
131    file: &mut File,
132    floor: u64,
133    upper: u64,
134) -> Option<CodexPeerStatus> {
135    let mut end = upper;
136    while end > floor {
137        let start = end.saturating_sub(LIFECYCLE_SCAN_BYTES).max(floor);
138        file.seek(SeekFrom::Start(start)).ok()?;
139        let mut tail = vec![0; (end - start) as usize];
140        file.read_exact(&mut tail).ok()?;
141        if let Some(status) = lifecycle_status_in_tail(&tail, start == floor) {
142            return Some(status);
143        }
144        if start == floor {
145            break;
146        }
147        // A lifecycle record is small, but an adjacent tool record can be enormous. Overlap keeps
148        // a boundary record whole without ever allocating in proportion to the rollout.
149        end = start.saturating_add(LIFECYCLE_OVERLAP_BYTES);
150    }
151    None
152}
153
154fn lifecycle_status_in_tail(
155    tail: &[u8],
156    starts_at_record_boundary: bool,
157) -> Option<CodexPeerStatus> {
158    const BOUNDARIES: [&str; 3] = [
159        "\"type\":\"task_started\"",
160        "\"type\":\"task_complete\"",
161        "\"type\":\"turn_aborted\"",
162    ];
163
164    // The read can begin in the middle of a large UTF-8 JSON string. Ignore
165    // that first fragment, then use the standard library's substring search
166    // to jump directly between lifecycle candidates instead of inspecting
167    // every byte of every tool payload with a naive sliding window.
168    let complete_start = if starts_at_record_boundary {
169        0
170    } else {
171        tail.iter()
172            .position(|byte| *byte == b'\n')
173            .map_or(tail.len(), |newline| newline + 1)
174    };
175    let text = std::str::from_utf8(&tail[complete_start..]).ok()?;
176    let mut search_end = text.len();
177    while let Some(candidate) = BOUNDARIES
178        .iter()
179        .filter_map(|boundary| text[..search_end].rfind(boundary))
180        .max()
181    {
182        let line_start = text[..candidate]
183            .rfind('\n')
184            .map_or(0, |newline| newline + 1);
185        let line_end = text[candidate..]
186            .find('\n')
187            .map_or(text.len(), |newline| candidate + newline);
188        let Ok(event) = serde_json::from_str::<Value>(&text[line_start..line_end]) else {
189            search_end = candidate;
190            continue;
191        };
192        if event.get("type").and_then(Value::as_str) != Some("event_msg") {
193            search_end = candidate;
194            continue;
195        }
196        match event
197            .get("payload")
198            .and_then(|payload| payload.get("type"))
199            .and_then(Value::as_str)
200        {
201            Some("task_started") => return Some(CodexPeerStatus::Busy),
202            Some("task_complete" | "turn_aborted") => return Some(CodexPeerStatus::Idle),
203            _ => {}
204        }
205        search_end = candidate;
206    }
207    None
208}
209
210fn normalized_path(path: &Path) -> PathBuf {
211    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
212}
213
214#[cfg(target_os = "macos")]
215fn platform_open_rollouts() -> Vec<PathBuf> {
216    use std::process::Command;
217
218    // Darwin's pgrep omits every ancestor of the caller unless `-a` is set.
219    // Discovery commonly runs underneath the very Codex session it must
220    // report (for example inside a Supercode-powered widget), so omitting
221    // ancestors makes the current session uniquely invisible.
222    let Ok(processes) = Command::new("/usr/bin/pgrep")
223        .args(["-a", "-x", "codex"])
224        .output()
225    else {
226        return Vec::new();
227    };
228    let pids = String::from_utf8_lossy(&processes.stdout)
229        .lines()
230        .filter_map(|line| line.trim().parse::<u32>().ok())
231        .take(128)
232        .map(|pid| pid.to_string())
233        .collect::<Vec<_>>();
234    if pids.is_empty() {
235        return Vec::new();
236    }
237    let Ok(files) = Command::new("/usr/sbin/lsof")
238        .args(["-Fn", "-a", "-p", &pids.join(",")])
239        .output()
240    else {
241        return Vec::new();
242    };
243    String::from_utf8_lossy(&files.stdout)
244        .lines()
245        .filter_map(|line| line.strip_prefix('n'))
246        .filter(|path| path.ends_with(".jsonl"))
247        .map(PathBuf::from)
248        .collect()
249}
250
251#[cfg(target_os = "linux")]
252fn platform_open_rollouts() -> Vec<PathBuf> {
253    let Ok(processes) = std::fs::read_dir("/proc") else {
254        return Vec::new();
255    };
256    let mut paths = Vec::new();
257    for process in processes.flatten() {
258        let pid = process.file_name();
259        if !pid.as_encoded_bytes().iter().all(u8::is_ascii_digit) {
260            continue;
261        }
262        let process_root = process.path();
263        if std::fs::read_to_string(process_root.join("comm"))
264            .ok()
265            .is_none_or(|name| name.trim() != "codex")
266        {
267            continue;
268        }
269        let Ok(descriptors) = std::fs::read_dir(process_root.join("fd")) else {
270            continue;
271        };
272        paths.extend(
273            descriptors
274                .flatten()
275                .filter_map(|descriptor| std::fs::read_link(descriptor.path()).ok())
276                .filter(|path| {
277                    path.extension().and_then(|extension| extension.to_str()) == Some("jsonl")
278                }),
279        );
280    }
281    paths
282}
283
284#[cfg(not(any(target_os = "macos", target_os = "linux")))]
285fn platform_open_rollouts() -> Vec<PathBuf> {
286    Vec::new()
287}
288
289#[cfg(test)]
290mod tests {
291    use std::fs::{remove_file, OpenOptions};
292    use std::io::Write;
293
294    use super::*;
295
296    #[test]
297    fn long_tool_heavy_turn_is_found_once_then_followed_incrementally() {
298        let path = std::env::temp_dir().join(format!(
299            "supercode-codex-long-turn-{}-{}.jsonl",
300            std::process::id(),
301            std::thread::current().name().unwrap_or("test")
302        ));
303        let mut file = File::create(&path).unwrap();
304        writeln!(
305            file,
306            r#"{{"type":"event_msg","payload":{{"type":"task_started"}}}}"#
307        )
308        .unwrap();
309        write!(
310            file,
311            r#"{{"type":"response_item","payload":"{}"}}"#,
312            "x".repeat(6 * 1024 * 1024)
313        )
314        .unwrap();
315        writeln!(file).unwrap();
316        file.flush().unwrap();
317
318        let mut cursor = CodexLifecycleCursor::default();
319        assert_eq!(
320            sample_lifecycle_status(&path, &mut cursor),
321            Some(CodexPeerStatus::Busy)
322        );
323        let first_offset = cursor.offset;
324
325        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
326        writeln!(
327            file,
328            r#"{{"type":"event_msg","payload":{{"type":"item_completed"}}}}"#
329        )
330        .unwrap();
331        file.flush().unwrap();
332        assert_eq!(
333            sample_lifecycle_status(&path, &mut cursor),
334            Some(CodexPeerStatus::Busy)
335        );
336        assert!(cursor.offset > first_offset);
337
338        writeln!(
339            file,
340            r#"{{"type":"event_msg","payload":{{"type":"task_complete"}}}}"#
341        )
342        .unwrap();
343        file.flush().unwrap();
344        assert_eq!(
345            sample_lifecycle_status(&path, &mut cursor),
346            Some(CodexPeerStatus::Idle)
347        );
348        remove_file(path).unwrap();
349    }
350}
supercode_harness/codex_peer.rs

supercode_harness/
codex_peer.rs
