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::{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);
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
109/// Lightweight native identity and direct parent from Codex's first
110/// `session_meta` record. Activity aggregation uses this to treat a process-
111/// owned subagent rollout as work inside its root conversation.
112pub(crate) fn rollout_lineage(path: &Path) -> Option<(String, Option<String>)> {
113    let mut header = String::new();
114    BufReader::new(File::open(path).ok()?.take(256 * 1024))
115        .read_line(&mut header)
116        .ok()?;
117    let value = serde_json::from_str::<Value>(&header).ok()?;
118    if value.get("type").and_then(Value::as_str) != Some("session_meta") {
119        return None;
120    }
121    let payload = value.get("payload")?;
122    let session_id = payload.get("id")?.as_str()?.to_string();
123    let parent_session_id = payload
124        .pointer("/source/subagent/thread_spawn/parent_thread_id")
125        .or_else(|| payload.get("parent_thread_id"))
126        .and_then(Value::as_str)
127        .map(str::to_string);
128    Some((session_id, parent_session_id))
129}
130
131fn sample_lifecycle_status(
132    path: &Path,
133    cursor: &mut CodexLifecycleCursor,
134) -> Option<CodexPeerStatus> {
135    let mut file = File::open(path).ok()?;
136    let length = file.metadata().ok()?.len();
137    if length < cursor.offset {
138        cursor.offset = 0;
139        cursor.status = None;
140    }
141    if cursor.offset == 0 {
142        cursor.status = latest_lifecycle_status_between(&mut file, 0, length);
143    } else if length > cursor.offset {
144        if let Some(status) = latest_lifecycle_status_between(&mut file, cursor.offset, length) {
145            cursor.status = Some(status);
146        }
147    }
148    cursor.offset = length;
149    cursor.status
150}
151
152fn latest_lifecycle_status_between(
153    file: &mut File,
154    floor: u64,
155    upper: u64,
156) -> Option<CodexPeerStatus> {
157    let mut end = upper;
158    while end > floor {
159        let start = end.saturating_sub(LIFECYCLE_SCAN_BYTES).max(floor);
160        file.seek(SeekFrom::Start(start)).ok()?;
161        let mut tail = vec![0; (end - start) as usize];
162        file.read_exact(&mut tail).ok()?;
163        if let Some(status) = lifecycle_status_in_tail(&tail, start == floor) {
164            return Some(status);
165        }
166        if start == floor {
167            break;
168        }
169        // A lifecycle record is small, but an adjacent tool record can be enormous. Overlap keeps
170        // a boundary record whole without ever allocating in proportion to the rollout.
171        end = start.saturating_add(LIFECYCLE_OVERLAP_BYTES);
172    }
173    None
174}
175
176fn lifecycle_status_in_tail(
177    tail: &[u8],
178    starts_at_record_boundary: bool,
179) -> Option<CodexPeerStatus> {
180    const BOUNDARIES: [&str; 3] = [
181        "\"type\":\"task_started\"",
182        "\"type\":\"task_complete\"",
183        "\"type\":\"turn_aborted\"",
184    ];
185
186    // The read can begin in the middle of a large UTF-8 JSON string. Ignore
187    // that first fragment, then use the standard library's substring search
188    // to jump directly between lifecycle candidates instead of inspecting
189    // every byte of every tool payload with a naive sliding window.
190    let complete_start = if starts_at_record_boundary {
191        0
192    } else {
193        tail.iter()
194            .position(|byte| *byte == b'\n')
195            .map_or(tail.len(), |newline| newline + 1)
196    };
197    let text = std::str::from_utf8(&tail[complete_start..]).ok()?;
198    let mut search_end = text.len();
199    while let Some(candidate) = BOUNDARIES
200        .iter()
201        .filter_map(|boundary| text[..search_end].rfind(boundary))
202        .max()
203    {
204        let line_start = text[..candidate]
205            .rfind('\n')
206            .map_or(0, |newline| newline + 1);
207        let line_end = text[candidate..]
208            .find('\n')
209            .map_or(text.len(), |newline| candidate + newline);
210        let Ok(event) = serde_json::from_str::<Value>(&text[line_start..line_end]) else {
211            search_end = candidate;
212            continue;
213        };
214        if event.get("type").and_then(Value::as_str) != Some("event_msg") {
215            search_end = candidate;
216            continue;
217        }
218        match event
219            .get("payload")
220            .and_then(|payload| payload.get("type"))
221            .and_then(Value::as_str)
222        {
223            Some("task_started") => return Some(CodexPeerStatus::Busy),
224            Some("task_complete" | "turn_aborted") => return Some(CodexPeerStatus::Idle),
225            _ => {}
226        }
227        search_end = candidate;
228    }
229    None
230}
231
232fn normalized_path(path: &Path) -> PathBuf {
233    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
234}
235
236#[cfg(target_os = "macos")]
237fn platform_open_rollouts() -> Vec<PathBuf> {
238    use std::process::Command;
239
240    // Darwin's pgrep omits every ancestor of the caller unless `-a` is set.
241    // Discovery commonly runs underneath the very Codex session it must
242    // report (for example inside a Supercode-powered widget), so omitting
243    // ancestors makes the current session uniquely invisible.
244    let Ok(processes) = Command::new("/usr/bin/pgrep")
245        .args(["-a", "-x", "codex"])
246        .output()
247    else {
248        return Vec::new();
249    };
250    let pids = String::from_utf8_lossy(&processes.stdout)
251        .lines()
252        .filter_map(|line| line.trim().parse::<u32>().ok())
253        .take(128)
254        .map(|pid| pid.to_string())
255        .collect::<Vec<_>>();
256    if pids.is_empty() {
257        return Vec::new();
258    }
259    let Ok(files) = Command::new("/usr/sbin/lsof")
260        .args(["-Fn", "-a", "-p", &pids.join(",")])
261        .output()
262    else {
263        return Vec::new();
264    };
265    String::from_utf8_lossy(&files.stdout)
266        .lines()
267        .filter_map(|line| line.strip_prefix('n'))
268        .filter(|path| path.ends_with(".jsonl"))
269        .map(PathBuf::from)
270        .collect()
271}
272
273#[cfg(target_os = "linux")]
274fn platform_open_rollouts() -> Vec<PathBuf> {
275    let Ok(processes) = std::fs::read_dir("/proc") else {
276        return Vec::new();
277    };
278    let mut paths = Vec::new();
279    for process in processes.flatten() {
280        let pid = process.file_name();
281        if !pid.as_encoded_bytes().iter().all(u8::is_ascii_digit) {
282            continue;
283        }
284        let process_root = process.path();
285        if std::fs::read_to_string(process_root.join("comm"))
286            .ok()
287            .is_none_or(|name| name.trim() != "codex")
288        {
289            continue;
290        }
291        let Ok(descriptors) = std::fs::read_dir(process_root.join("fd")) else {
292            continue;
293        };
294        paths.extend(
295            descriptors
296                .flatten()
297                .filter_map(|descriptor| std::fs::read_link(descriptor.path()).ok())
298                .filter(|path| {
299                    path.extension().and_then(|extension| extension.to_str()) == Some("jsonl")
300                }),
301        );
302    }
303    paths
304}
305
306#[cfg(not(any(target_os = "macos", target_os = "linux")))]
307fn platform_open_rollouts() -> Vec<PathBuf> {
308    Vec::new()
309}
310
311#[cfg(test)]
312mod tests {
313    use std::fs::{remove_file, OpenOptions};
314    use std::io::Write;
315
316    use super::*;
317
318    #[test]
319    fn long_tool_heavy_turn_is_found_once_then_followed_incrementally() {
320        let path = std::env::temp_dir().join(format!(
321            "supercode-codex-long-turn-{}-{}.jsonl",
322            std::process::id(),
323            std::thread::current().name().unwrap_or("test")
324        ));
325        let mut file = File::create(&path).unwrap();
326        writeln!(
327            file,
328            r#"{{"type":"event_msg","payload":{{"type":"task_started"}}}}"#
329        )
330        .unwrap();
331        write!(
332            file,
333            r#"{{"type":"response_item","payload":"{}"}}"#,
334            "x".repeat(6 * 1024 * 1024)
335        )
336        .unwrap();
337        writeln!(file).unwrap();
338        file.flush().unwrap();
339
340        let mut cursor = CodexLifecycleCursor::default();
341        assert_eq!(
342            sample_lifecycle_status(&path, &mut cursor),
343            Some(CodexPeerStatus::Busy)
344        );
345        let first_offset = cursor.offset;
346
347        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
348        writeln!(
349            file,
350            r#"{{"type":"event_msg","payload":{{"type":"item_completed"}}}}"#
351        )
352        .unwrap();
353        file.flush().unwrap();
354        assert_eq!(
355            sample_lifecycle_status(&path, &mut cursor),
356            Some(CodexPeerStatus::Busy)
357        );
358        assert!(cursor.offset > first_offset);
359
360        writeln!(
361            file,
362            r#"{{"type":"event_msg","payload":{{"type":"task_complete"}}}}"#
363        )
364        .unwrap();
365        file.flush().unwrap();
366        assert_eq!(
367            sample_lifecycle_status(&path, &mut cursor),
368            Some(CodexPeerStatus::Idle)
369        );
370        remove_file(path).unwrap();
371    }
372}
supercode_harness/codex_peer.rs

supercode_harness/
codex_peer.rs
