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::price_for;
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    summary: SessionSummary,
158    /// Dedupe state: the last API message id seen and what it contributed.
159    last_msg_id: Option<String>,
160    last_contrib: (TokenUsage, f64, u64),
161}
162
163impl ClaudeTranscript {
164    pub fn new(path: impl Into<PathBuf>) -> Self {
165        ClaudeTranscript {
166            reader: TailReader::new(path),
167            summary: SessionSummary { harness: Some(Harness::Claude), ..Default::default() },
168            last_msg_id: None,
169            last_contrib: (TokenUsage::default(), 0.0, 0),
170        }
171    }
172
173    pub fn set_registry_hints(&mut self, ps: &PidSession) {
174        self.summary.session_id.get_or_insert_with(|| ps.session_id.clone());
175        self.summary.cwd.get_or_insert_with(|| ps.cwd.clone());
176        if ps.version.is_some() {
177            self.summary.harness_version = ps.version.clone();
178        }
179        if self.summary.started_at.is_none() {
180            self.summary.started_at = ps.started();
181        }
182    }
183
184    fn ingest(&mut self, line: &str) {
185        let Ok(v) = serde_json::from_str::<Value>(line) else { return };
186        let kind = v.get("type").and_then(Value::as_str).unwrap_or("");
187        if let Some(ts) = v.get("timestamp").and_then(Value::as_str).and_then(parse_rfc3339_utc) {
188            if self.summary.started_at.is_none() {
189                self.summary.started_at = Some(ts);
190            }
191            self.summary.last_activity = Some(ts);
192        }
193        if self.summary.session_id.is_none() {
194            self.summary.session_id = v.get("sessionId").and_then(Value::as_str).map(str::to_string);
195        }
196        if self.summary.cwd.is_none() {
197            self.summary.cwd = v.get("cwd").and_then(Value::as_str).map(PathBuf::from);
198        }
199        if self.summary.harness_version.is_none() {
200            self.summary.harness_version = v.get("version").and_then(Value::as_str).map(str::to_string);
201        }
202        let sidechain = v.get("isSidechain").and_then(Value::as_bool).unwrap_or(false);
203        let is_meta = v.get("isMeta").and_then(Value::as_bool).unwrap_or(false);
204        let ts = v.get("timestamp").and_then(Value::as_str).and_then(parse_rfc3339_utc);
205        match kind {
206            "assistant" => self.ingest_assistant(&v, sidechain, ts),
207            "user" if !is_meta => {
208                // Either a prompt or a tool_result: in both cases the model owes a response.
209                self.summary.activity = Activity::Working;
210                if let Some(ts) = ts {
211                    self.close_spans(&v, ts);
212                }
213            }
214            _ => {}
215        }
216    }
217
218    /// A user line answering tool calls: every `tool_result` block closes a span.
219    fn close_spans(&mut self, v: &Value, ts: SystemTime) {
220        let Some(content) = v.pointer("/message/content").and_then(Value::as_array) else { return };
221        for b in content {
222            if b.get("type").and_then(Value::as_str) != Some("tool_result") {
223                continue;
224            }
225            let Some(id) = b.get("tool_use_id").and_then(Value::as_str) else { continue };
226            self.summary.spans.close(id, ts, b.get("is_error").and_then(Value::as_bool).unwrap_or(false));
227        }
228    }
229
230    fn ingest_assistant(&mut self, v: &Value, sidechain: bool, ts: Option<SystemTime>) {
231        let Some(msg) = v.get("message") else { return };
232        let id = msg.get("id").and_then(Value::as_str).map(str::to_string);
233        let model = msg.get("model").and_then(Value::as_str).unwrap_or("");
234        if !model.is_empty() && model != "<synthetic>" {
235            self.summary.model = Some(model.to_string());
236        }
237        if let Some(content) = msg.get("content").and_then(Value::as_array) {
238            let calls = content.iter().filter(|b| b.get("type").and_then(Value::as_str) == Some("tool_use"));
239            for b in calls {
240                self.summary.tool_calls += 1;
241                if let (Some(ts), Some(id)) = (ts, b.get("id").and_then(Value::as_str)) {
242                    let name = b.get("name").and_then(Value::as_str).unwrap_or("tool");
243                    self.summary.spans.open(id.to_string(), name.to_string(), ts, sidechain);
244                }
245            }
246        }
247        match msg.get("stop_reason").and_then(Value::as_str) {
248            Some("end_turn") | Some("stop_sequence") | Some("max_tokens") | Some("refusal") => {
249                self.summary.activity = Activity::Waiting;
250            }
251            _ => self.summary.activity = Activity::Working,
252        }
253
254        let usage = msg.get("usage").map(parse_usage).unwrap_or_default();
255        let price = price_for(model);
256        let cost = price.map(|p| p.cost(&usage)).unwrap_or(0.0);
257        let unpriced = if price.is_none() { usage.total() } else { 0 };
258
259        let same_message = id.is_some() && id == self.last_msg_id;
260        if same_message {
261            // Replace the previous contribution from this id with the latest one.
262            let (u, c, un) = self.last_contrib;
263            self.summary.usage.sub(&u);
264            self.summary.cost_usd -= c;
265            self.summary.unpriced_tokens = self.summary.unpriced_tokens.saturating_sub(un);
266        } else {
267            self.summary.turns += 1;
268            if sidechain {
269                self.summary.subagent_turns += 1;
270            }
271        }
272        self.summary.usage.add(&usage);
273        self.summary.cost_usd += cost;
274        self.summary.unpriced_tokens += unpriced;
275        self.last_msg_id = id;
276        self.last_contrib = (usage, cost, unpriced);
277    }
278}
279
280fn parse_usage(u: &Value) -> TokenUsage {
281    let g = |k: &str| u.get(k).and_then(Value::as_u64).unwrap_or(0);
282    let cache_write_total = g("cache_creation_input_tokens");
283    let (w1h, w5m) = match u.get("cache_creation") {
284        Some(cc) => (
285            cc.get("ephemeral_1h_input_tokens").and_then(Value::as_u64).unwrap_or(0),
286            cc.get("ephemeral_5m_input_tokens").and_then(Value::as_u64).unwrap_or(0),
287        ),
288        None => (0, 0),
289    };
290    // Older transcripts have only the total; treat it as 5-minute writes.
291    let (w1h, w5m) = if w1h + w5m == 0 { (0, cache_write_total) } else { (w1h, w5m) };
292    TokenUsage {
293        input: g("input_tokens"),
294        cache_write_5m: w5m,
295        cache_write_1h: w1h,
296        cache_read: g("cache_read_input_tokens"),
297        output: g("output_tokens"),
298    }
299}
300
301impl SessionTracker for ClaudeTranscript {
302    fn refresh(&mut self) -> anyhow::Result<bool> {
303        let (lines, more) = self.reader.read_new_lines(REFRESH_BUDGET_BYTES)?;
304        for l in &lines {
305            self.ingest(l);
306        }
307        Ok(more)
308    }
309
310    fn summary(&self) -> &SessionSummary {
311        &self.summary
312    }
313
314    fn path(&self) -> &Path {
315        self.reader.path()
316    }
317}
318
319#[cfg(test)]
320mod tests {
321    use super::*;
322    use std::io::Write;
323
324    #[test]
325    fn encodes_paths_like_claude_code() {
326        assert_eq!(
327            encode_project_path(Path::new("/Users/atlas/Documents/orbital/forge/agent-top")),
328            "-Users-atlas-Documents-orbital-forge-agent-top"
329        );
330        assert_eq!(encode_project_path(Path::new("/tmp/a.b_c")), "-tmp-a-b-c");
331    }
332
333    #[test]
334    fn dedupes_usage_by_message_id_and_tracks_state() {
335        let dir = std::env::temp_dir().join(format!("agent-top-claude-{}", std::process::id()));
336        std::fs::create_dir_all(&dir).unwrap();
337        let path = dir.join("s.jsonl");
338        let mut f = std::fs::File::create(&path).unwrap();
339        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}}"#;
340        writeln!(f, r#"{{"type":"user","timestamp":"2026-09-03T07:00:00.000Z","sessionId":"abc","cwd":"/tmp/p","message":{{"role":"user","content":"hi"}}}}"#).unwrap();
341        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();
342        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();
343        let mut t = ClaudeTranscript::new(&path);
344        t.refresh().unwrap();
345        let s = t.summary();
346        assert_eq!(s.turns, 1);
347        assert_eq!(s.tool_calls, 1);
348        assert_eq!(s.usage.total(), 1152);
349        assert_eq!(s.activity, Activity::Working);
350        assert_eq!(s.session_id.as_deref(), Some("abc"));
351        // sonnet-5: 2*2 + 100*4 + 1000*0.2 + 50*10 = 4 + 400 + 200 + 500 = 1104 micro-dollars
352        assert!((s.cost_usd - 0.001104).abs() < 1e-9);
353        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();
354        t.refresh().unwrap();
355        assert_eq!(t.summary().turns, 2);
356        assert_eq!(t.summary().activity, Activity::Waiting);
357        let _ = std::fs::remove_dir_all(&dir);
358    }
359
360    #[test]
361    fn builds_spans_from_tool_use_and_tool_result() {
362        let dir = std::env::temp_dir().join(format!("agent-top-claude-spans-{}", std::process::id()));
363        std::fs::create_dir_all(&dir).unwrap();
364        let path = dir.join("s.jsonl");
365        let mut f = std::fs::File::create(&path).unwrap();
366        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();
367        // Results arrive on one line, in the other order, one of them failed.
368        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();
369        // A subagent call that has not come back yet.
370        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();
371        let mut t = ClaudeTranscript::new(&path);
372        t.refresh().unwrap();
373        let spans = t.summary().spans.to_vec();
374        assert_eq!(spans.len(), 3);
375        assert_eq!(spans[0].name, "Bash");
376        assert_eq!(spans[0].duration_ms, Some(2_500));
377        assert!(!spans[0].error);
378        assert_eq!(spans[1].name, "Read");
379        assert_eq!(spans[1].duration_ms, Some(2_500));
380        assert!(spans[1].error);
381        assert!(spans[2].is_open());
382        assert!(spans[2].sidechain);
383        assert_eq!(t.summary().tool_calls, 3);
384        let _ = std::fs::remove_dir_all(&dir);
385    }
386}
agent_top_core/harness/claude.rs

agent_top_core/harness/
claude.rs
