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        let usage = msg.get("usage").map(parse_usage).unwrap_or_default();
264        let price = self.prices.lookup(model);
265        let cost = price.map(|p| p.cost(&usage)).unwrap_or(0.0);
266        let unpriced = if price.is_none() { usage.total() } else { 0 };
267
268        let same_message = id.is_some() && id == self.last_msg_id;
269        if same_message {
270            // Replace the previous contribution from this id with the latest one.
271            let (u, c, un) = self.last_contrib;
272            self.summary.usage.sub(&u);
273            self.summary.cost_usd -= c;
274            self.summary.unpriced_tokens = self.summary.unpriced_tokens.saturating_sub(un);
275        } else {
276            self.summary.turns += 1;
277            if sidechain {
278                self.summary.subagent_turns += 1;
279            }
280        }
281        self.summary.usage.add(&usage);
282        self.summary.cost_usd += cost;
283        self.summary.unpriced_tokens += unpriced;
284        self.last_msg_id = id;
285        self.last_contrib = (usage, cost, unpriced);
286    }
287}
288
289fn parse_usage(u: &Value) -> TokenUsage {
290    let g = |k: &str| u.get(k).and_then(Value::as_u64).unwrap_or(0);
291    let cache_write_total = g("cache_creation_input_tokens");
292    let (w1h, w5m) = match u.get("cache_creation") {
293        Some(cc) => (
294            cc.get("ephemeral_1h_input_tokens").and_then(Value::as_u64).unwrap_or(0),
295            cc.get("ephemeral_5m_input_tokens").and_then(Value::as_u64).unwrap_or(0),
296        ),
297        None => (0, 0),
298    };
299    // Older transcripts have only the total; treat it as 5-minute writes.
300    let (w1h, w5m) = if w1h + w5m == 0 { (0, cache_write_total) } else { (w1h, w5m) };
301    TokenUsage {
302        input: g("input_tokens"),
303        cache_write_5m: w5m,
304        cache_write_1h: w1h,
305        cache_read: g("cache_read_input_tokens"),
306        output: g("output_tokens"),
307    }
308}
309
310impl SessionTracker for ClaudeTranscript {
311    fn refresh(&mut self) -> anyhow::Result<bool> {
312        let (lines, more) = self.reader.read_new_lines(REFRESH_BUDGET_BYTES)?;
313        for l in &lines {
314            self.ingest(l);
315        }
316        Ok(more)
317    }
318
319    fn summary(&self) -> &SessionSummary {
320        &self.summary
321    }
322
323    fn path(&self) -> &Path {
324        self.reader.path()
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use super::*;
331    use std::io::Write;
332
333    #[test]
334    fn encodes_paths_like_claude_code() {
335        assert_eq!(
336            encode_project_path(Path::new("/Users/atlas/Documents/orbital/forge/agent-top")),
337            "-Users-atlas-Documents-orbital-forge-agent-top"
338        );
339        assert_eq!(encode_project_path(Path::new("/tmp/a.b_c")), "-tmp-a-b-c");
340    }
341
342    #[test]
343    fn dedupes_usage_by_message_id_and_tracks_state() {
344        let dir = std::env::temp_dir().join(format!("agent-top-claude-{}", std::process::id()));
345        std::fs::create_dir_all(&dir).unwrap();
346        let path = dir.join("s.jsonl");
347        let mut f = std::fs::File::create(&path).unwrap();
348        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}}"#;
349        writeln!(f, r#"{{"type":"user","timestamp":"2026-09-03T07:00:00.000Z","sessionId":"abc","cwd":"/tmp/p","message":{{"role":"user","content":"hi"}}}}"#).unwrap();
350        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();
351        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();
352        let mut t = ClaudeTranscript::new(&path);
353        t.refresh().unwrap();
354        let s = t.summary();
355        assert_eq!(s.turns, 1);
356        assert_eq!(s.tool_calls, 1);
357        assert_eq!(s.usage.total(), 1152);
358        assert_eq!(s.activity, Activity::Working);
359        assert_eq!(s.session_id.as_deref(), Some("abc"));
360        // sonnet-5: 2*2 + 100*4 + 1000*0.2 + 50*10 = 4 + 400 + 200 + 500 = 1104 micro-dollars
361        assert!((s.cost_usd - 0.001104).abs() < 1e-9);
362        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();
363        t.refresh().unwrap();
364        assert_eq!(t.summary().turns, 2);
365        assert_eq!(t.summary().activity, Activity::Waiting);
366        let _ = std::fs::remove_dir_all(&dir);
367    }
368
369    #[test]
370    fn builds_spans_from_tool_use_and_tool_result() {
371        let dir = std::env::temp_dir().join(format!("agent-top-claude-spans-{}", std::process::id()));
372        std::fs::create_dir_all(&dir).unwrap();
373        let path = dir.join("s.jsonl");
374        let mut f = std::fs::File::create(&path).unwrap();
375        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();
376        // Results arrive on one line, in the other order, one of them failed.
377        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();
378        // A subagent call that has not come back yet.
379        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();
380        let mut t = ClaudeTranscript::new(&path);
381        t.refresh().unwrap();
382        let spans = t.summary().spans.to_vec();
383        assert_eq!(spans.len(), 3);
384        assert_eq!(spans[0].name, "Bash");
385        assert_eq!(spans[0].duration_ms, Some(2_500));
386        assert!(!spans[0].error);
387        assert_eq!(spans[1].name, "Read");
388        assert_eq!(spans[1].duration_ms, Some(2_500));
389        assert!(spans[1].error);
390        assert!(spans[2].is_open());
391        assert!(spans[2].sidechain);
392        assert_eq!(t.summary().tool_calls, 3);
393        let _ = std::fs::remove_dir_all(&dir);
394    }
395}
agent_top_core/harness/claude.rs

agent_top_core/harness/
claude.rs
