pma_core/
watch.rs

1//! `pma watch`: what running agents are doing, read from their logs. The
2//! runner observes; no agent reports. A `claude` log is `stream-json`, one
3//! record per event; any other worker's log is shown by its last line.
4
5use std::path::{Path, PathBuf};
6
7use crate::store::{Job, Run, RunState};
8
9/// What a run's log says so far.
10#[derive(Debug, Default, PartialEq)]
11pub struct Progress {
12    pub tool_calls: usize,
13    /// The last tool used, with its main argument.
14    pub last_tool: Option<String>,
15    /// The last thing the agent said, or a non-JSON log's last line.
16    pub last_text: Option<String>,
17}
18
19/// Reads a `stream-json` log, or the tail of any other.
20pub fn progress(log: &str) -> Progress {
21    let mut p = Progress::default();
22    let mut json = false;
23    for line in log.lines() {
24        let Ok(v) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
25            continue;
26        };
27        json = true;
28        if v["type"] != "assistant" {
29            continue;
30        }
31        for part in v["message"]["content"].as_array().into_iter().flatten() {
32            match part["type"].as_str() {
33                Some("tool_use") => {
34                    p.tool_calls += 1;
35                    let input = &part["input"];
36                    let arg = ["command", "file_path", "pattern", "path", "url"]
37                        .iter()
38                        .find_map(|k| input[*k].as_str());
39                    let name = part["name"].as_str().unwrap_or("?");
40                    p.last_tool = Some(match arg {
41                        Some(a) => format!("{name}: {}", one_line(a, 80)),
42                        None => name.to_string(),
43                    });
44                }
45                Some("text") => {
46                    if let Some(t) = part["text"].as_str().filter(|t| !t.trim().is_empty()) {
47                        p.last_text = Some(one_line(t, 100));
48                    }
49                }
50                _ => {}
51            }
52        }
53    }
54    if !json {
55        p.last_text = log
56            .lines()
57            .rev()
58            .find(|l| !l.trim().is_empty())
59            .map(|l| one_line(l, 100));
60    }
61    p
62}
63
64fn one_line(text: &str, max: usize) -> String {
65    let flat = text.split_whitespace().collect::<Vec<_>>().join(" ");
66    match flat.char_indices().nth(max) {
67        Some((i, _)) => format!("{}...", &flat[..i]),
68        None => flat,
69    }
70}
71
72/// The newest `agent-<n>.log` of a run.
73fn latest_log(state: &Path, run: i64) -> Option<PathBuf> {
74    let dir = state.join("runs").join(run.to_string());
75    (1..)
76        .map(|n| dir.join(format!("agent-{n}.log")))
77        .take_while(|p| p.exists())
78        .last()
79}
80
81pub fn age(seconds: i64) -> String {
82    match seconds {
83        s if s < 60 => format!("{s}s"),
84        s if s < 3600 => format!("{}m{:02}s", s / 60, s % 60),
85        s => format!("{}h{:02}m", s / 3600, (s % 3600) / 60),
86    }
87}
88
89/// A running agent: its run, how long it has run, how long its log has been
90/// quiet, and what the log says.
91pub struct Agent<'a> {
92    pub run: &'a Run,
93    pub running: String,
94    pub quiet: String,
95    pub progress: Progress,
96}
97
98/// The jobs not yet settled.
99pub fn open_jobs(jobs: &[Job]) -> Vec<&Job> {
100    jobs.iter()
101        .filter(|j| matches!(j.state.as_str(), "queued" | "running" | "awaiting-approval"))
102        .collect()
103}
104
105/// Each running run, with its progress read from its newest log.
106pub fn agents<'a>(state: &Path, runs: &'a [Run], now: i64) -> Vec<Agent<'a>> {
107    runs.iter()
108        .filter(|r| r.state == RunState::Running)
109        .map(|r| {
110            let log = latest_log(state, r.id);
111            let quiet = log
112                .as_ref()
113                .and_then(|l| std::fs::metadata(l).ok())
114                .and_then(|m| m.modified().ok())
115                .and_then(|m| m.elapsed().ok())
116                .map_or("?".into(), |d| age(d.as_secs() as i64));
117            Agent {
118                run: r,
119                running: r.dispatched_at.map_or("?".into(), |t| age(now - t)),
120                quiet,
121                progress: log
122                    .and_then(|l| std::fs::read_to_string(l).ok())
123                    .map(|text| progress(&text))
124                    .unwrap_or_default(),
125            }
126        })
127        .collect()
128}
129
130/// One screen: the jobs not yet settled, then each running run.
131pub fn screen(state: &Path, jobs: &[Job], runs: &[Run], now: i64) -> String {
132    let mut out = String::new();
133    let open = open_jobs(jobs);
134    for j in &open {
135        out.push_str(&format!(
136            "job #{} {}: pma {}\n",
137            j.id,
138            j.state,
139            j.args.join(" ")
140        ));
141    }
142    let running = agents(state, runs, now);
143    if running.is_empty() {
144        out.push_str("no agent is running\n");
145    }
146    for a in running {
147        let (r, since, quiet, p) = (a.run, a.running, a.quiet, a.progress);
148        out.push_str(&format!(
149            "#{} {} {}  {}  running {since}, last output {quiet} ago, {} tool calls\n",
150            r.id,
151            r.project,
152            r.node.as_deref().unwrap_or("dispatch"),
153            r.agent,
154            p.tool_calls
155        ));
156        if let Some(t) = p.last_tool {
157            out.push_str(&format!("    {t}\n"));
158        }
159        if let Some(t) = p.last_text {
160            out.push_str(&format!("    > {t}\n"));
161        }
162    }
163    out
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169
170    #[test]
171    fn a_stream_is_read_for_tools_and_text() {
172        let log = r#"{"type":"system","subtype":"init"}
173{"type":"assistant","message":{"content":[{"type":"text","text":"Reading the\n code."},{"type":"tool_use","name":"Read","input":{"file_path":"src/a.rs"}}]}}
174{"type":"user","message":{"content":[{"type":"tool_result","content":"..."}]}}
175{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Bash","input":{"command":"make test"}}]}}
176"#;
177        assert_eq!(
178            progress(log),
179            Progress {
180                tool_calls: 2,
181                last_tool: Some("Bash: make test".into()),
182                last_text: Some("Reading the code.".into()),
183            }
184        );
185    }
186
187    #[test]
188    fn a_plain_log_shows_its_last_line() {
189        let p = progress("building\n\ncompiling foo\n\n");
190        assert_eq!(p.tool_calls, 0);
191        assert_eq!(p.last_text.as_deref(), Some("compiling foo"));
192    }
193
194    #[test]
195    fn long_text_is_cut_on_a_character() {
196        assert_eq!(one_line("ééé", 2), "éé...");
197    }
198}
pma_core/watch.rs

pma_core/
watch.rs
