agent_top_core/
collector.rs

1//! Joins the process table with the transcripts into a `Snapshot`.
2//!
3//! The collector knows no harness by name. Each one is a `HarnessAdapter`
4//! (RFC-101): it lists its transcripts, says which belong to which process,
5//! and opens a tracker for one. The collector walks the process forest, asks
6//! the adapter for each root, and builds the rows.
7
8use crate::harness::{self, AttributeContext, HarnessAdapter, RegistryHints, SessionSummary, SessionTracker, SpanRetention};
9use crate::model::*;
10use crate::process::{ProcessScanner, RawProc, build_forest};
11use std::collections::{HashMap, HashSet};
12use std::path::{Path, PathBuf};
13use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
14
15#[derive(Debug, Clone)]
16pub struct CollectorOptions {
17    /// How long after its last write a process-less transcript still shows as `stopped`.
18    pub stopped_window: Duration,
19    /// How often to re-list transcript directories.
20    pub fs_scan_interval: Duration,
21    /// A transcript idle for longer than this counts as idle even if the
22    /// harness never wrote an end-of-turn marker.
23    pub activity_timeout: Duration,
24}
25
26impl Default for CollectorOptions {
27    fn default() -> Self {
28        CollectorOptions {
29            stopped_window: Duration::from_secs(30 * 60),
30            fs_scan_interval: Duration::from_secs(5),
31            activity_timeout: Duration::from_secs(15 * 60),
32        }
33    }
34}
35
36pub struct Collector {
37    opts: CollectorOptions,
38    scanner: ProcessScanner,
39    adapters: Vec<Box<dyn HarnessAdapter>>,
40    trackers: HashMap<PathBuf, Box<dyn SessionTracker>>,
41    last_fs_scan: Option<Instant>,
42}
43
44impl Collector {
45    pub fn new(opts: CollectorOptions) -> Self {
46        Collector { opts, scanner: ProcessScanner::new(), adapters: harness::adapters(), trackers: HashMap::new(), last_fs_scan: None }
47    }
48
49    fn rescan_fs_if_due(&mut self) {
50        let due = self.last_fs_scan.map(|t| t.elapsed() >= self.opts.fs_scan_interval).unwrap_or(true);
51        if !due {
52            return;
53        }
54        self.last_fs_scan = Some(Instant::now());
55        let since = SystemTime::now().checked_sub(self.opts.stopped_window).unwrap_or(UNIX_EPOCH);
56        for a in &mut self.adapters {
57            a.rescan(since);
58        }
59    }
60
61    fn adapter(&self, harness: Harness) -> Option<&dyn HarnessAdapter> {
62        self.adapters.iter().find(|a| a.harness() == harness).map(|a| a.as_ref())
63    }
64
65    /// The tracker for a transcript, opened on first sight.
66    fn tracker_for(&mut self, path: &Path, harness: Harness) -> Option<&mut Box<dyn SessionTracker>> {
67        if !self.trackers.contains_key(path) {
68            let tracker = self.adapter(harness)?.open(path, SpanRetention::Recent);
69            self.trackers.insert(path.to_path_buf(), tracker);
70        }
71        self.trackers.get_mut(path)
72    }
73
74    pub fn collect(&mut self) -> Snapshot {
75        self.scanner.refresh();
76        self.rescan_fs_if_due();
77        let host = self.scanner.host();
78        let procs = self.scanner.processes();
79        let by_pid: HashMap<u32, &RawProc> = procs.iter().map(|p| (p.pid, p)).collect();
80        let (roots, orphans) = build_forest(&procs);
81
82        // Each adapter sees all of its processes before any is attributed.
83        for a in &mut self.adapters {
84            let mine: Vec<&ProcNode> = roots.iter().filter(|r| r.harness == Some(a.harness())).collect();
85            a.prepare(&mine);
86        }
87
88        let now = SystemTime::now();
89        let mut agents = Vec::new();
90        let mut attached: HashSet<PathBuf> = HashSet::new();
91
92        for root in roots {
93            let raw = by_pid.get(&root.pid).copied();
94            let harness = root.harness.unwrap_or(Harness::Unknown);
95            let proc_start = raw.map(|p| UNIX_EPOCH + Duration::from_secs(p.start_time)).unwrap_or(now);
96            let hints = self.adapter(harness).and_then(|a| a.hints(root.pid));
97            let cwd = root.cwd.clone().or_else(|| hints.as_ref().and_then(|h| h.cwd.clone()));
98
99            // One process can host several conversations. Claude Code runs one
100            // per process; the Codex app-server runs many.
101            let (paths, attribution) = match self.adapter(harness) {
102                Some(a) => {
103                    let ctx = AttributeContext {
104                        cwd: cwd.as_deref(),
105                        proc_start,
106                        now,
107                        attached: &attached,
108                        activity_timeout: self.opts.activity_timeout,
109                    };
110                    a.attribute(&root, raw, &ctx)
111                }
112                None => (Vec::new(), Attribution::None),
113            };
114
115            let (cpu, rss, count, mcp) = root.totals();
116            let hints = hints.unwrap_or_default();
117
118            // No transcript: the process still deserves a row.
119            if paths.is_empty() {
120                let summary = SessionSummary::default();
121                let state = live_state(&hints, summary.activity, None, cpu, &self.opts);
122                agents.push(Agent {
123                    id: format!("pid:{}", root.pid),
124                    name: hints.name.clone().unwrap_or_else(|| display_name(harness, cwd.as_deref())),
125                    harness,
126                    state,
127                    activity: summary.activity,
128                    pid: Some(root.pid),
129                    session_id: hints.session_id.clone(),
130                    session_path: None,
131                    cwd,
132                    model: None,
133                    harness_version: hints.version.clone(),
134                    usage: summary.usage,
135                    cost_usd: 0.0,
136                    cost_breakdown: Default::default(),
137                    price_source: None,
138                    unpriced_tokens: 0,
139                    turns: 0,
140                    subagent_turns: 0,
141                    tool_calls: 0,
142                    web_searches: 0,
143                    spans: Vec::new(),
144                    age_secs: root.age_secs,
145                    idle_secs: None,
146                    cpu_percent: cpu,
147                    rss_bytes: rss,
148                    process_count: count,
149                    mcp_count: mcp,
150                    tree: Some(root),
151                    attribution,
152                    shares_process: false,
153                    parse_warning: None,
154                });
155                continue;
156            }
157
158            for (i, path) in paths.iter().enumerate() {
159                // Only the first row carries the process, so that a machine's
160                // totals are not multiplied by the number of conversations.
161                let owns_process = i == 0;
162                let Some(tr) = self.tracker_for(path, harness) else { continue };
163                let _ = tr.refresh();
164                let mut summary = tr.summary().clone();
165                attached.insert(path.clone());
166
167                if summary.session_id.is_none() {
168                    summary.session_id = hints.session_id.clone();
169                }
170                if summary.harness_version.is_none() {
171                    summary.harness_version = hints.version.clone();
172                }
173
174                let idle_secs = summary.last_activity.and_then(|t| now.duration_since(t).ok()).map(|d| d.as_secs());
175                let state = live_state(&hints, summary.activity, idle_secs, cpu, &self.opts);
176                // A thread names itself after its own working directory, which
177                // is the only thing distinguishing two rows on one app-server.
178                let name = match (hints.name.clone(), paths.len()) {
179                    (Some(n), 1) => n,
180                    _ => display_name(harness, summary.cwd.as_deref().or(cwd.as_deref())),
181                };
182                let id = match summary.session_id.as_deref() {
183                    Some(sid) => format!("pid:{}:{}", root.pid, sid),
184                    None => format!("pid:{}:{}", root.pid, path.file_stem().map(|s| s.to_string_lossy().into_owned()).unwrap_or_default()),
185                };
186
187                agents.push(Agent {
188                    id,
189                    name,
190                    harness,
191                    state,
192                    activity: summary.activity,
193                    pid: Some(root.pid),
194                    session_id: summary.session_id.clone(),
195                    session_path: Some(path.clone()),
196                    cwd: summary.cwd.clone().or_else(|| cwd.clone()),
197                    model: summary.model.clone(),
198                    harness_version: summary.harness_version.clone(),
199                    usage: summary.usage,
200                    cost_usd: summary.cost_usd,
201                    cost_breakdown: summary.cost_breakdown,
202                    price_source: summary.model.as_deref().and_then(|m| crate::pricing::table().source_for(m)),
203                    unpriced_tokens: summary.unpriced_tokens,
204                    turns: summary.turns,
205                    subagent_turns: summary.subagent_turns,
206                    tool_calls: summary.tool_calls,
207                    web_searches: summary.web_searches,
208                    spans: summary.spans.to_vec(),
209                    age_secs: root.age_secs,
210                    idle_secs,
211                    cpu_percent: if owns_process { cpu } else { 0.0 },
212                    rss_bytes: if owns_process { rss } else { 0 },
213                    process_count: if owns_process { count } else { 0 },
214                    mcp_count: if owns_process { mcp } else { 0 },
215                    tree: if owns_process { Some(root.clone()) } else { None },
216                    attribution,
217                    shares_process: !owns_process,
218                    parse_warning: parse_warning(&summary, harness),
219                });
220            }
221        }
222
223        // Stopped agents: recently written transcripts nobody owns.
224        let stopped: Vec<(PathBuf, Harness)> =
225            self.adapters.iter().flat_map(|a| a.unowned(&attached).into_iter().map(move |p| (p, a.harness()))).collect();
226        for (p, harness) in stopped {
227            let Some(tr) = self.tracker_for(&p, harness) else { continue };
228            let _ = tr.refresh();
229            let s = tr.summary().clone();
230            if s.turns == 0 && s.usage.total() == 0 {
231                continue;
232            }
233            let idle_secs = s.last_activity.and_then(|t| now.duration_since(t).ok()).map(|d| d.as_secs());
234            let id = s.session_id.clone().unwrap_or_else(|| p.file_stem().map(|x| x.to_string_lossy().into_owned()).unwrap_or_default());
235            agents.push(Agent {
236                id: format!("session:{id}"),
237                name: display_name(harness, s.cwd.as_deref()),
238                harness,
239                state: AgentState::Stopped,
240                activity: s.activity,
241                pid: None,
242                session_id: Some(id),
243                session_path: Some(p),
244                cwd: s.cwd.clone(),
245                model: s.model.clone(),
246                harness_version: s.harness_version.clone(),
247                usage: s.usage,
248                cost_usd: s.cost_usd,
249                cost_breakdown: s.cost_breakdown,
250                price_source: s.model.as_deref().and_then(|m| crate::pricing::table().source_for(m)),
251                unpriced_tokens: s.unpriced_tokens,
252                turns: s.turns,
253                subagent_turns: s.subagent_turns,
254                tool_calls: s.tool_calls,
255                web_searches: s.web_searches,
256                spans: s.spans.to_vec(),
257                age_secs: idle_secs.unwrap_or(0),
258                idle_secs,
259                cpu_percent: 0.0,
260                rss_bytes: 0,
261                process_count: 0,
262                mcp_count: 0,
263                tree: None,
264                attribution: Attribution::TranscriptOnly,
265                shares_process: false,
266                parse_warning: parse_warning(&s, harness),
267            });
268        }
269
270        // Drop trackers for transcripts that fell out of the window.
271        let keep: HashSet<&PathBuf> = agents.iter().filter_map(|a| a.session_path.as_ref()).collect();
272        self.trackers.retain(|p, _| keep.contains(p));
273
274        let mut snap =
275            Snapshot { schema_version: SNAPSHOT_SCHEMA_VERSION, taken_at: now, host, agents, orphans, totals: Totals::default() };
276        snap.compute_totals();
277        snap
278    }
279}
280
281/// A transcript that parsed while its usage records did not is a format change,
282/// not a quiet session. Naming the harness version makes the report actionable:
283/// it is the first thing anyone will ask for.
284fn parse_warning(s: &SessionSummary, harness: Harness) -> Option<String> {
285    if !s.health.fields_unrecognised() {
286        return None;
287    }
288    let version = s.harness_version.as_deref().unwrap_or("unknown version");
289    Some(format!(
290        "usage fields not recognised in {} {}: tokens and cost are unreliable, agent-top may need updating",
291        harness.label(),
292        version
293    ))
294}
295
296fn live_state(hints: &RegistryHints, activity: Activity, idle_secs: Option<u64>, cpu: f32, opts: &CollectorOptions) -> AgentState {
297    // Statuses observed in the registry so far (Claude Code 2.1.259): "busy",
298    // "idle", "shell". Unknown values fall through to the transcript heuristic.
299    match hints.status.as_deref() {
300        Some("busy" | "running" | "working" | "shell" | "tool" | "thinking") => return AgentState::Running,
301        Some("idle" | "waiting" | "paused" | "permission" | "blocked") => return AgentState::Idle,
302        _ => {}
303    }
304    match activity {
305        Activity::Working => {
306            if idle_secs.map(|s| s > opts.activity_timeout.as_secs()).unwrap_or(false) {
307                AgentState::Idle
308            } else {
309                AgentState::Running
310            }
311        }
312        Activity::Waiting => AgentState::Idle,
313        Activity::Unknown => {
314            if cpu > 5.0 || idle_secs.map(|s| s < 10).unwrap_or(false) {
315                AgentState::Running
316            } else {
317                AgentState::Idle
318            }
319        }
320    }
321}
322
323fn display_name(harness: Harness, cwd: Option<&Path>) -> String {
324    match cwd.and_then(|c| c.file_name()).map(|f| f.to_string_lossy().into_owned()) {
325        Some(dir) => format!("{}:{}", harness.label(), dir),
326        None => harness.label().to_string(),
327    }
328}
329
330#[cfg(test)]
331mod tests {
332    use super::*;
333
334    #[test]
335    fn the_registry_status_beats_the_transcript_heuristic() {
336        let opts = CollectorOptions::default();
337        let busy = RegistryHints { status: Some("busy".into()), ..Default::default() };
338        assert_eq!(live_state(&busy, Activity::Waiting, Some(0), 0.0, &opts), AgentState::Running);
339        let idle = RegistryHints { status: Some("idle".into()), ..Default::default() };
340        assert_eq!(live_state(&idle, Activity::Working, Some(0), 90.0, &opts), AgentState::Idle);
341        // No registry, which is every harness but Claude Code: the transcript decides.
342        let none = RegistryHints::default();
343        assert_eq!(live_state(&none, Activity::Working, Some(1), 0.0, &opts), AgentState::Running);
344        assert_eq!(live_state(&none, Activity::Working, Some(opts.activity_timeout.as_secs() + 1), 0.0, &opts), AgentState::Idle);
345        assert_eq!(live_state(&none, Activity::Waiting, Some(1), 90.0, &opts), AgentState::Idle);
346        assert_eq!(live_state(&none, Activity::Unknown, Some(3), 0.0, &opts), AgentState::Running);
347        assert_eq!(live_state(&none, Activity::Unknown, Some(300), 0.0, &opts), AgentState::Idle);
348    }
349
350    #[test]
351    fn every_adapter_is_a_distinct_harness_and_recognises_its_own_fixture() {
352        let adapters = harness::adapters();
353        let mut seen = HashSet::new();
354        for a in &adapters {
355            assert!(seen.insert(a.harness()), "two adapters for {:?}", a.harness());
356        }
357        let fixtures = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
358        for (file, want) in
359            [("claude-2.1.226.jsonl", Harness::Claude), ("codex-0.130.jsonl", Harness::Codex), ("gemini-0.58.jsonl", Harness::Gemini)]
360        {
361            assert_eq!(harness::detect(&fixtures.join(file)), Some(want), "{file}");
362        }
363    }
364}
agent_top_core/collector.rs

agent_top_core/
collector.rs
