agent_top_core/harness/
claude.rs

1//! Claude Code: `~/.claude/sessions/<pid>.json` registry and
2//! `~/.claude/projects/<encoded-cwd>/<session>.jsonl` transcripts.
3//!
4//! Format notes (verified on Claude Code 2.1.259, 2026-09-03):
5//! * One API response is written as several lines, one per content block,
6//!   every line carrying the same `message.id` and the same `message.usage`.
7//!   Usage must be counted once per id.
8//! * `usage.cache_creation.ephemeral_1h_input_tokens` /
9//!   `ephemeral_5m_input_tokens` split cache writes by TTL, which have
10//!   different prices.
11//! * `isSidechain: true` marks subagent turns.
12//! * A `tool_use` block in an assistant message and the `tool_result` block
13//!   that answers it carry the same id in `id` / `tool_use_id`, and their
14//!   lines carry the timestamps that bracket the call. That pairing is the
15//!   trace: verified 240/240 on a real session.
16//! * The registry file has `status: "busy" | "idle"`, which is the harness's
17//!   own opinion of its state and beats any transcript heuristic.
18
19use super::{REFRESH_BUDGET_BYTES, SessionSummary, SessionTracker, parse_rfc3339_utc};
20use crate::jsonl::TailReader;
21use crate::model::{Activity, Harness, TokenUsage};
22use crate::pricing::{self, Table};
23use serde::Deserialize;
24use serde_json::Value;
25use std::path::{Path, PathBuf};
26use std::time::{Duration, SystemTime, UNIX_EPOCH};
27
28pub fn home() -> Option<PathBuf> {
29    std::env::var_os("HOME").map(PathBuf::from)
30}
31
32pub fn claude_dir() -> Option<PathBuf> {
33    if let Some(d) = std::env::var_os("CLAUDE_CONFIG_DIR") {
34        return Some(PathBuf::from(d));
35    }
36    home().map(|h| h.join(".claude"))
37}
38
39pub fn sessions_dir() -> Option<PathBuf> {
40    claude_dir().map(|d| d.join("sessions"))
41}
42
43pub fn projects_dir() -> Option<PathBuf> {
44    claude_dir().map(|d| d.join("projects"))
45}
46
47/// Claude Code's project directory name: every character that is not
48/// ASCII alphanumeric becomes `-`, so `/Users/a/x.y` is `-Users-a-x-y`.
49pub fn encode_project_path(p: &Path) -> String {
50    p.to_string_lossy().chars().map(|c| if c.is_ascii_alphanumeric() { c } else { '-' }).collect()
51}
52
53pub fn transcript_path(cwd: &Path, session_id: &str) -> Option<PathBuf> {
54    projects_dir().map(|d| d.join(encode_project_path(cwd)).join(format!("{session_id}.jsonl")))
55}
56
57/// One `~/.claude/sessions/<pid>.json`.
58#[derive(Debug, Clone, Deserialize)]
59#[serde(rename_all = "camelCase")]
60pub struct PidSession {
61    pub pid: u32,
62    pub session_id: String,
63    pub cwd: PathBuf,
64    #[serde(default)]
65    pub name: Option<String>,
66    #[serde(default)]
67    pub status: Option<String>,
68    #[serde(default)]
69    pub version: Option<String>,
70    #[serde(default)]
71    pub kind: Option<String>,
72    #[serde(default)]
73    pub entrypoint: Option<String>,
74    #[serde(default)]
75    pub started_at: Option<u64>,
76    #[serde(default)]
77    pub updated_at: Option<u64>,
78}
79
80impl PidSession {
81    pub fn started(&self) -> Option<SystemTime> {
82        self.started_at.map(|ms| UNIX_EPOCH + Duration::from_millis(ms))
83    }
84}
85
86/// Read every registry file. Stale files for dead pids are returned too; the
87/// caller reconciles against the process table.
88pub fn read_pid_sessions() -> Vec<PidSession> {
89    let Some(dir) = sessions_dir() else { return Vec::new() };
90    let Ok(rd) = std::fs::read_dir(&dir) else { return Vec::new() };
91    let mut out = Vec::new();
92    for e in rd.flatten() {
93        let p = e.path();
94        if p.extension().and_then(|x| x.to_str()) != Some("json") {
95            continue;
96        }
97        if let Ok(s) = std::fs::read_to_string(&p)
98            && let Ok(ps) = serde_json::from_str::<PidSession>(&s)
99        {
100            out.push(ps);
101        }
102    }
103    out
104}
105
106/// Transcripts modified after `since`, across all projects.
107pub fn recent_transcripts(since: SystemTime) -> Vec<PathBuf> {
108    let Some(dir) = projects_dir() else { return Vec::new() };
109    let Ok(projects) = std::fs::read_dir(&dir) else { return Vec::new() };
110    let mut out = Vec::new();
111    for proj in projects.flatten() {
112        let Ok(files) = std::fs::read_dir(proj.path()) else { continue };
113        for f in files.flatten() {
114            let p = f.path();
115            if p.extension().and_then(|x| x.to_str()) != Some("jsonl") {
116                continue;
117            }
118            if let Ok(md) = f.metadata()
119                && md.modified().map(|m| m >= since).unwrap_or(false)
120            {
121                out.push(p);
122            }
123        }
124    }
125    out
126}
127
128/// Fallback attribution when the registry has no entry: the transcript in
129/// the cwd's project directory created closest after the process start.
130pub fn guess_transcript(cwd: &Path, proc_start: SystemTime) -> Option<PathBuf> {
131    let dir = projects_dir()?.join(encode_project_path(cwd));
132    let rd = std::fs::read_dir(&dir).ok()?;
133    let slack = Duration::from_secs(15);
134    let mut best: Option<(Duration, PathBuf)> = None;
135    for f in rd.flatten() {
136        let p = f.path();
137        if p.extension().and_then(|x| x.to_str()) != Some("jsonl") {
138            continue;
139        }
140        // A file we cannot stat is skipped, not fatal: one unreadable
141        // transcript must not abandon attribution for the whole directory.
142        let Ok(md) = f.metadata() else { continue };
143        let Ok(created) = md.created().or_else(|_| md.modified()) else { continue };
144        if created + slack < proc_start {
145            continue;
146        }
147        let gap = created.duration_since(proc_start).unwrap_or(Duration::ZERO);
148        if best.as_ref().map(|(g, _)| gap < *g).unwrap_or(true) {
149            best = Some((gap, p));
150        }
151    }
152    best.map(|(_, p)| p)
153}
154
155pub struct ClaudeTranscript {
156    reader: TailReader,
157    prices: &'static Table,
158    summary: SessionSummary,
159    /// Dedupe state: the last API message id seen and what it contributed.
160    last_msg_id: Option<String>,
161    last_contrib: (TokenUsage, f64, u64),
162}
163
164impl ClaudeTranscript {
165    pub fn new(path: impl Into<PathBuf>) -> Self {
166        ClaudeTranscript {
167            reader: TailReader::new(path),
168            prices: pricing::table(),
169            summary: SessionSummary { harness: Some(Harness::Claude), ..Default::default() },
170            last_msg_id: None,
171            last_contrib: (TokenUsage::default(), 0.0, 0),
172        }
173    }
174
175    /// Price with this table instead of the process-wide one. Lets a test
176    /// assert a cost without the developer's own price file changing it.
177    pub fn with_prices(mut self, prices: &'static Table) -> Self {
178        self.prices = prices;
179        self
180    }
181
182    pub fn set_registry_hints(&mut self, ps: &PidSession) {
183        self.summary.session_id.get_or_insert_with(|| ps.session_id.clone());
184        self.summary.cwd.get_or_insert_with(|| ps.cwd.clone());
185        if ps.version.is_some() {
186            self.summary.harness_version = ps.version.clone();
187        }
188        if self.summary.started_at.is_none() {
189            self.summary.started_at = ps.started();
190        }
191    }
192
193    fn ingest(&mut self, line: &str) {
194        let Ok(v) = serde_json::from_str::<Value>(line) else { return };
195        let kind = v.get("type").and_then(Value::as_str).unwrap_or("");
196        if let Some(ts) = v.get("timestamp").and_then(Value::as_str).and_then(parse_rfc3339_utc) {
197            if self.summary.started_at.is_none() {
198                self.summary.started_at = Some(ts);
199            }
200            self.summary.last_activity = Some(ts);
201        }
202        if self.summary.session_id.is_none() {
203            self.summary.session_id = v.get("sessionId").and_then(Value::as_str).map(str::to_string);
204        }
205        if self.summary.cwd.is_none() {
206            self.summary.cwd = v.get("cwd").and_then(Value::as_str).map(PathBuf::from);
207        }
208        if self.summary.harness_version.is_none() {
209            self.summary.harness_version = v.get("version").and_then(Value::as_str).map(str::to_string);
210        }
211        let sidechain = v.get("isSidechain").and_then(Value::as_bool).unwrap_or(false);
212        let is_meta = v.get("isMeta").and_then(Value::as_bool).unwrap_or(false);
213        let ts = v.get("timestamp").and_then(Value::as_str).and_then(parse_rfc3339_utc);
214        match kind {
215            "assistant" => self.ingest_assistant(&v, sidechain, ts),
216            "user" if !is_meta => {
217                // Either a prompt or a tool_result: in both cases the model owes a response.
218                self.summary.activity = Activity::Working;
219                if let Some(ts) = ts {
220                    self.close_spans(&v, ts);
221                }
222            }
223            _ => {}
224        }
225    }
226
227    /// A user line answering tool calls: every `tool_result` block closes a span.
228    fn close_spans(&mut self, v: &Value, ts: SystemTime) {
229        let Some(content) = v.pointer("/message/content").and_then(Value::as_array) else { return };
230        for b in content {
231            if b.get("type").and_then(Value::as_str) != Some("tool_result") {
232                continue;
233            }
234            let Some(id) = b.get("tool_use_id").and_then(Value::as_str) else { continue };
235            self.summary.spans.close(id, ts, b.get("is_error").and_then(Value::as_bool).unwrap_or(false));
236        }
237    }
238
239    fn ingest_assistant(&mut self, v: &Value, sidechain: bool, ts: Option<SystemTime>) {
240        let Some(msg) = v.get("message") else { return };
241        let id = msg.get("id").and_then(Value::as_str).map(str::to_string);
242        let model = msg.get("model").and_then(Value::as_str).unwrap_or("");
243        if !model.is_empty() && model != "<synthetic>" {
244            self.summary.model = Some(model.to_string());
245        }
246        if let Some(content) = msg.get("content").and_then(Value::as_array) {
247            let calls = content.iter().filter(|b| b.get("type").and_then(Value::as_str) == Some("tool_use"));
248            for b in calls {
249                self.summary.tool_calls += 1;
250                if let (Some(ts), Some(id)) = (ts, b.get("id").and_then(Value::as_str)) {
251                    let name = b.get("name").and_then(Value::as_str).unwrap_or("tool");
252                    self.summary.spans.open(id.to_string(), name.to_string(), ts, sidechain);
253                }
254            }
255        }
256        match msg.get("stop_reason").and_then(Value::as_str) {
257            Some("end_turn") | Some("stop_sequence") | Some("max_tokens") | Some("refusal") => {
258                self.summary.activity = Activity::Waiting;
259            }
260            _ => self.summary.activity = Activity::Working,
261        }
262
263        // Health is judged on the record being present but unreadable, which is
264        // what a renamed field looks like from in here.
265        if !same_message_id(id.as_deref(), self.last_msg_id.as_deref()) {
266            self.summary.health.billable_messages += 1;
267        }
268        let usage = match msg.get("usage") {
269            Some(u) => {
270                let parsed = parse_usage(u);
271                self.summary.health.usage_records += 1;
272                if parsed.total() == 0 {
273                    self.summary.health.empty_usage_records += 1;
274                }
275                parsed
276            }
277            None => TokenUsage::default(),
278        };
279        let price = self.prices.lookup(model);
280        let cost = price.map(|p| p.cost(&usage)).unwrap_or(0.0);
281        let unpriced = if price.is_none() { usage.total() } else { 0 };
282
283        let same_message = id.is_some() && id == self.last_msg_id;
284        if same_message {
285            // Replace the previous contribution from this id with the latest one.
286            let (u, c, un) = self.last_contrib;
287            self.summary.usage.sub(&u);
288            self.summary.cost_usd -= c;
289            self.summary.unpriced_tokens = self.summary.unpriced_tokens.saturating_sub(un);
290        } else {
291            self.summary.turns += 1;
292            if sidechain {
293                self.summary.subagent_turns += 1;
294            }
295        }
296        self.summary.usage.add(&usage);
297        self.summary.cost_usd += cost;
298        self.summary.unpriced_tokens += unpriced;
299        self.last_msg_id = id;
300        self.last_contrib = (usage, cost, unpriced);
301    }
302}
303
304fn same_message_id(a: Option<&str>, b: Option<&str>) -> bool {
305    matches!((a, b), (Some(x), Some(y)) if x == y)
306}
307
308fn parse_usage(u: &Value) -> TokenUsage {
309    let g = |k: &str| u.get(k).and_then(Value::as_u64).unwrap_or(0);
310    let cache_write_total = g("cache_creation_input_tokens");
311    let (w1h, w5m) = match u.get("cache_creation") {
312        Some(cc) => (
313            cc.get("ephemeral_1h_input_tokens").and_then(Value::as_u64).unwrap_or(0),
314            cc.get("ephemeral_5m_input_tokens").and_then(Value::as_u64).unwrap_or(0),
315        ),
316        None => (0, 0),
317    };
318    // Older transcripts have only the total; treat it as 5-minute writes.
319    let (w1h, w5m) = if w1h + w5m == 0 { (0, cache_write_total) } else { (w1h, w5m) };
320    TokenUsage {
321        input: g("input_tokens"),
322        cache_write_5m: w5m,
323        cache_write_1h: w1h,
324        cache_read: g("cache_read_input_tokens"),
325        output: g("output_tokens"),
326    }
327}
328
329impl SessionTracker for ClaudeTranscript {
330    fn refresh(&mut self) -> anyhow::Result<bool> {
331        let (lines, more) = self.reader.read_new_lines(REFRESH_BUDGET_BYTES)?;
332        for l in &lines {
333            self.ingest(l);
334        }
335        Ok(more)
336    }
337
338    fn summary(&self) -> &SessionSummary {
339        &self.summary
340    }
341
342    fn path(&self) -> &Path {
343        self.reader.path()
344    }
345}
346
347#[cfg(test)]
348mod tests {
349    use super::*;
350    use std::io::Write;
351
352    #[test]
353    fn encodes_paths_like_claude_code() {
354        assert_eq!(
355            encode_project_path(Path::new("/Users/atlas/Documents/orbital/forge/agent-top")),
356            "-Users-atlas-Documents-orbital-forge-agent-top"
357        );
358        assert_eq!(encode_project_path(Path::new("/tmp/a.b_c")), "-tmp-a-b-c");
359    }
360
361    #[test]
362    fn dedupes_usage_by_message_id_and_tracks_state() {
363        let dir = std::env::temp_dir().join(format!("agent-top-claude-{}", std::process::id()));
364        std::fs::create_dir_all(&dir).unwrap();
365        let path = dir.join("s.jsonl");
366        let mut f = std::fs::File::create(&path).unwrap();
367        let usage = r#"{"input_tokens":2,"cache_creation_input_tokens":100,"cache_read_input_tokens":1000,"output_tokens":50,"cache_creation":{"ephemeral_1h_input_tokens":100,"ephemeral_5m_input_tokens":0}}"#;
368        writeln!(f, r#"{{"type":"user","timestamp":"2026-09-03T07:00:00.000Z","sessionId":"abc","cwd":"/tmp/p","message":{{"role":"user","content":"hi"}}}}"#).unwrap();
369        writeln!(f, r#"{{"type":"assistant","timestamp":"2026-09-03T07:00:01.000Z","message":{{"id":"msg_1","model":"claude-sonnet-5","stop_reason":"tool_use","content":[{{"type":"text","text":"x"}}],"usage":{usage}}}}}"#).unwrap();
370        writeln!(f, r#"{{"type":"assistant","timestamp":"2026-09-03T07:00:02.000Z","message":{{"id":"msg_1","model":"claude-sonnet-5","stop_reason":"tool_use","content":[{{"type":"tool_use","name":"Bash"}}],"usage":{usage}}}}}"#).unwrap();
371        let mut t = ClaudeTranscript::new(&path);
372        t.refresh().unwrap();
373        let s = t.summary();
374        assert_eq!(s.turns, 1);
375        assert_eq!(s.tool_calls, 1);
376        assert_eq!(s.usage.total(), 1152);
377        assert_eq!(s.activity, Activity::Working);
378        assert_eq!(s.session_id.as_deref(), Some("abc"));
379        // sonnet-5: 2*2 + 100*4 + 1000*0.2 + 50*10 = 4 + 400 + 200 + 500 = 1104 micro-dollars
380        assert!((s.cost_usd - 0.001104).abs() < 1e-9);
381        writeln!(f, r#"{{"type":"assistant","timestamp":"2026-09-03T07:00:03.000Z","message":{{"id":"msg_2","model":"claude-sonnet-5","stop_reason":"end_turn","content":[],"usage":{{"input_tokens":1,"output_tokens":1}}}}}}"#).unwrap();
382        t.refresh().unwrap();
383        assert_eq!(t.summary().turns, 2);
384        assert_eq!(t.summary().activity, Activity::Waiting);
385        let _ = std::fs::remove_dir_all(&dir);
386    }
387
388    #[test]
389    fn builds_spans_from_tool_use_and_tool_result() {
390        let dir = std::env::temp_dir().join(format!("agent-top-claude-spans-{}", std::process::id()));
391        std::fs::create_dir_all(&dir).unwrap();
392        let path = dir.join("s.jsonl");
393        let mut f = std::fs::File::create(&path).unwrap();
394        writeln!(f, r#"{{"type":"assistant","timestamp":"2026-09-03T07:00:00.000Z","message":{{"id":"m1","model":"claude-sonnet-5","stop_reason":"tool_use","content":[{{"type":"tool_use","id":"toolu_a","name":"Bash"}},{{"type":"tool_use","id":"toolu_b","name":"Read"}}],"usage":{{"input_tokens":1}}}}}}"#).unwrap();
395        // Results arrive on one line, in the other order, one of them failed.
396        writeln!(f, r#"{{"type":"user","timestamp":"2026-09-03T07:00:02.500Z","message":{{"role":"user","content":[{{"type":"tool_result","tool_use_id":"toolu_b","is_error":true}},{{"type":"tool_result","tool_use_id":"toolu_a","is_error":false}}]}},"toolUseResult":{{}}}}"#).unwrap();
397        // A subagent call that has not come back yet.
398        writeln!(f, r#"{{"type":"assistant","timestamp":"2026-09-03T07:00:03.000Z","isSidechain":true,"message":{{"id":"m2","model":"claude-sonnet-5","stop_reason":"tool_use","content":[{{"type":"tool_use","id":"toolu_c","name":"Grep"}}],"usage":{{"input_tokens":1}}}}}}"#).unwrap();
399        let mut t = ClaudeTranscript::new(&path);
400        t.refresh().unwrap();
401        let spans = t.summary().spans.to_vec();
402        assert_eq!(spans.len(), 3);
403        assert_eq!(spans[0].name, "Bash");
404        assert_eq!(spans[0].duration_ms, Some(2_500));
405        assert!(!spans[0].error);
406        assert_eq!(spans[1].name, "Read");
407        assert_eq!(spans[1].duration_ms, Some(2_500));
408        assert!(spans[1].error);
409        assert!(spans[2].is_open());
410        assert!(spans[2].sidechain);
411        assert_eq!(t.summary().tool_calls, 3);
412        let _ = std::fs::remove_dir_all(&dir);
413    }
414}
agent_top_core/harness/claude.rs

agent_top_core/harness/
claude.rs
