use crate::pricing::Provider;
use crate::session::{ActivityState, ContextUsage, Remote, Session, Surface};
use crate::util;
use serde_json::Value;
use std::collections::HashMap;
use std::process::Command;
use std::time::Duration;
pub const POLL: Duration = Duration::from_secs(15);
const CONNECT_TIMEOUT_SECS: u64 = 10;
const ALIVE_INTERVAL_SECS: u64 = 5;
const ALIVE_RETRIES: u64 = 2;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Host {
pub target: String,
pub command: String,
}
#[derive(Debug, Clone)]
pub enum Snapshot {
Rows(Vec<Session>),
Failed(String),
}
impl Host {
pub fn parse(spec: &str) -> Option<Host> {
let spec = spec.trim();
if spec.is_empty() {
return None;
}
match spec.rsplit_once(':') {
Some((target, command)) if !target.is_empty() && !command.is_empty() => Some(Host {
target: target.to_string(),
command: command.to_string(),
}),
_ => Some(Host {
target: spec.to_string(),
command: "cctop".to_string(),
}),
}
}
pub fn collect(flags: &[String]) -> Vec<Host> {
let env = std::env::var("CCTOP_HOSTS").unwrap_or_default();
let mut out: Vec<Host> = Vec::new();
for spec in flags.iter().map(String::as_str).chain(env.split(',')) {
if let Some(host) = Host::parse(spec)
&& !out.iter().any(|h| h.target == host.target)
{
out.push(host);
}
}
out
}
pub fn poll(&self) -> Snapshot {
match self.run(&["--json"]) {
Ok(json) => match parse(&self.target, &json) {
Ok(rows) => Snapshot::Rows(rows),
Err(why) => Snapshot::Failed(why),
},
Err(why) => Snapshot::Failed(why),
}
}
pub fn ssh_argv(&self, args: &[&str]) -> Vec<String> {
let mut argv: Vec<String> = [
"ssh",
"-T",
"-o",
"BatchMode=yes",
"-o",
&format!("ConnectTimeout={CONNECT_TIMEOUT_SECS}"),
"-o",
&format!("ServerAliveInterval={ALIVE_INTERVAL_SECS}"),
"-o",
&format!("ServerAliveCountMax={ALIVE_RETRIES}"),
&self.target,
"--",
&self.command,
]
.iter()
.map(|a| a.to_string())
.collect();
argv.extend(args.iter().map(|a| a.to_string()));
argv
}
pub fn shown_command(&self, args: &[&str]) -> String {
let mut words = vec!["ssh", self.target.as_str(), "--", self.command.as_str()];
words.extend(args);
words.join(" ")
}
pub fn probe(&self) -> Probe {
probe_from(self.run(&["--version"]))
}
pub fn update(&self) -> Result<String, UpdateFailure> {
update_outcome(self, self.run(&["--update"]))
}
pub fn sudo_update_command(&self) -> String {
format!("ssh -t {} sudo {} --update", self.target, self.command)
}
pub fn run(&self, args: &[&str]) -> Result<String, String> {
let argv = self.ssh_argv(args);
let out = Command::new(&argv[0])
.args(&argv[1..])
.output()
.map_err(|e| format!("could not run ssh: {e}"))?;
if !out.status.success() {
let why = String::from_utf8_lossy(&out.stderr);
let why = why.lines().next_back().unwrap_or("").trim();
return Err(match why.is_empty() {
true => format!("{} exited {}", self.command, out.status),
false => why.to_string(),
});
}
String::from_utf8(out.stdout).map_err(|_| "output was not UTF-8".to_string())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Probe {
Version(String),
Missing,
Unreadable(String),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Skew {
Older(String),
Newer(String),
Missing,
}
pub fn skew(probe: &Probe, local: &str) -> Option<Skew> {
match probe {
Probe::Missing => Some(Skew::Missing),
Probe::Unreadable(_) => None,
Probe::Version(v) if crate::update::is_newer(local, v) => Some(Skew::Older(v.clone())),
Probe::Version(v) if crate::update::is_newer(v, local) => Some(Skew::Newer(v.clone())),
Probe::Version(_) => None,
}
}
pub fn is_missing(why: &str) -> bool {
let lower = why.to_ascii_lowercase();
lower.contains("command not found") || lower.contains("no such file or directory")
}
fn probe_from(result: Result<String, String>) -> Probe {
match result {
Ok(out) => match parse_version(&out) {
Some(v) => Probe::Version(v),
None => Probe::Unreadable(format!("unexpected --version output: {}", out.trim())),
},
Err(why) if is_missing(&why) => Probe::Missing,
Err(why) => Probe::Unreadable(why),
}
}
pub fn parse_version(out: &str) -> Option<String> {
let line = out.lines().find(|l| !l.trim().is_empty())?;
let word = line.split_whitespace().last()?.trim_start_matches('v');
word.starts_with(|c: char| c.is_ascii_digit())
.then(|| word.to_string())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum UpdateFailure {
NeedsRoot(String),
Other(String),
}
fn update_outcome(host: &Host, result: Result<String, String>) -> Result<String, UpdateFailure> {
match result {
Ok(out) => {
let mut lines = out.lines().map(str::trim).filter(|l| !l.is_empty());
let said = lines
.clone()
.find(|l| l.starts_with("Updated ") || l.starts_with("Already on"))
.or_else(|| lines.next_back())
.unwrap_or("updated");
Ok(said.to_string())
}
Err(why) if why.contains("not writable by this user") || why.contains("os error 13") => {
Err(UpdateFailure::NeedsRoot(host.sudo_update_command()))
}
Err(why) => Err(UpdateFailure::Other(
why.trim_start_matches("Error: ").to_string(),
)),
}
}
#[derive(Debug, Default)]
pub struct Handshake {
probed: bool,
told_missing: bool,
}
#[derive(Debug, PartialEq, Eq)]
pub enum Ask {
Nothing,
Probe,
Missing,
}
impl Handshake {
pub fn after(&mut self, snapshot: &Snapshot) -> Ask {
match snapshot {
Snapshot::Rows(_) => {
self.told_missing = false;
match std::mem::replace(&mut self.probed, true) {
true => Ask::Nothing,
false => Ask::Probe,
}
}
Snapshot::Failed(why) => {
self.probed = false;
match is_missing(why) && !std::mem::replace(&mut self.told_missing, true) {
true => Ask::Missing,
false => Ask::Nothing,
}
}
}
}
}
pub fn parse(host: &str, json: &str) -> Result<Vec<Session>, String> {
let doc: Value = serde_json::from_str(json).map_err(|e| format!("unreadable output: {e}"))?;
let list = doc
.as_array()
.ok_or("output was not the array --json produces")?;
let mut rows: Vec<Session> = list.iter().filter_map(|v| row(host, v)).collect();
let labels: Vec<String> = rows.iter().map(|s| s.label_source.clone()).collect();
for (s, short) in rows.iter_mut().zip(util::abbreviate_paths(&labels)) {
s.abbrev_label = short;
}
Ok(rows)
}
fn row(host: &str, v: &Value) -> Option<Session> {
let provider = provider_of(text(v, "provider"))?;
let id = text(v, "session_id");
if id.is_empty() {
return None;
}
let mut s = Session::new(provider, id.to_string());
s.remote = Some(Remote {
host: host.to_string(),
branch: v.get("branch").and_then(Value::as_str).map(str::to_string),
..Default::default()
});
s.surface = match text(v, "surface") {
"editor" => Surface::Editor,
"desktop-code" => Surface::DesktopCode,
"desktop-cowork" => Surface::DesktopCowork,
_ => Surface::Cli,
};
s.started_at = text(v, "started_at").to_string();
s.last_active = text(v, "last_active").to_string();
s.label_source = text(v, "project").to_string();
s.model = text(v, "model").to_string();
s.harness = text(v, "harness").to_string();
s.title = v
.get("title")
.and_then(Value::as_str)
.filter(|t| !t.is_empty())
.map(str::to_string);
s.profile = v
.get("profile")
.and_then(Value::as_str)
.filter(|p| !p.is_empty())
.map(str::to_string);
s.permission = crate::hook::Permission::parse(text(v, "permission"));
s.activity_state = match text(v, "state") {
"waiting" => ActivityState::WaitingForInput,
"asking" => ActivityState::Asking,
"error" => ActivityState::ApiError,
_ => ActivityState::Working,
};
s.inferred_running = v.get("running").and_then(Value::as_bool).unwrap_or(false);
if let Some(p) = v.get("process") {
s.process = Some(crate::proc::ProcInfo {
pids: num(p, "pids") as usize,
cpu: num(p, "cpu") as f32,
memory: num(p, "memory") as u64,
command: text(p, "command").to_string(),
process_list: Vec::new(),
});
}
let tokens = v.get("tokens");
s.input_tokens = tokens.map(|t| num(t, "input") as u64).unwrap_or(0);
s.output_tokens = tokens.map(|t| num(t, "output") as u64).unwrap_or(0);
s.tokens_per_min = num(v, "tokens_per_min");
if let Some(a) = v.get("activity") {
s.tool_count = num(a, "tool_count") as u64;
s.tool_errors = num(a, "tool_errors") as u64;
s.compactions = num(a, "compactions") as u32;
}
if let Some(c) = v.get("cost") {
s.cost_available = c.get("available").and_then(Value::as_bool).unwrap_or(true);
let included = c.get("included").and_then(Value::as_bool).unwrap_or(false);
s.total_cost = match included {
true => None,
false => Some(
c.get("total")
.and_then(Value::as_str)
.and_then(|t| t.trim_start_matches('$').replace(',', "").parse().ok())
.unwrap_or(0.0),
),
};
s.cost_is_free = c.get("free").and_then(Value::as_bool).unwrap_or(false);
s.cost_hour = c
.get("last_hour")
.and_then(Value::as_f64)
.unwrap_or_else(|| num(c, "this_hour"));
s.cost_today = num(c, "today");
s.cost_per_min = num(c, "per_min");
s.costs_by_day = buckets(c.get("by_day"));
s.costs_by_hour = buckets(c.get("by_hour"));
}
if let Some(ctx) = v.get("context") {
let max = num(ctx, "max") as u64;
if max > 0 {
s.context = Some(ContextUsage {
used: num(ctx, "used") as u64,
max,
compacted: ctx
.get("compacted")
.and_then(Value::as_bool)
.unwrap_or(false),
});
}
}
s.conflict = match text(v.get("conflict").unwrap_or(&Value::Null), "level") {
"file" => Some(crate::collide::Overlap::File),
"directory" => Some(crate::collide::Overlap::Directory),
_ => None,
};
s.subagents = v
.get("subagents")
.and_then(Value::as_array)
.map(|list| {
list.iter()
.filter_map(|sa| serde_json::from_value(sa.clone()).ok())
.collect()
})
.unwrap_or_default();
s.subagents_cost = s.subagents.iter().map(|sa| sa.cost).sum();
Some(s)
}
fn buckets(v: Option<&Value>) -> HashMap<String, HashMap<String, f64>> {
let Some(map) = v.and_then(Value::as_object) else {
return HashMap::new();
};
map.iter()
.filter_map(|(k, v)| v.as_f64().map(|amount| (k.clone(), amount)))
.map(|(k, amount)| (k, HashMap::from([("remote".to_string(), amount)])))
.collect()
}
fn text<'a>(v: &'a Value, key: &str) -> &'a str {
v.get(key).and_then(Value::as_str).unwrap_or("")
}
fn num(v: &Value, key: &str) -> f64 {
v.get(key).and_then(Value::as_f64).unwrap_or(0.0)
}
fn provider_of(name: &str) -> Option<Provider> {
Some(match name {
"claude" => Provider::Claude,
"codex" => Provider::Codex,
"cursor" => Provider::Cursor,
"devin" => Provider::Devin,
"gemini" => Provider::Gemini,
"opencode" => Provider::OpenCode,
"pi" => Provider::Pi,
"windsurf" => Provider::Windsurf,
_ => return None,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_spec_may_name_the_binary_that_ssh_cannot_find() {
assert_eq!(
Host::parse("box"),
Some(Host {
target: "box".into(),
command: "cctop".into()
})
);
assert_eq!(
Host::parse("flo@box:/opt/bin/cctop"),
Some(Host {
target: "flo@box".into(),
command: "/opt/bin/cctop".into()
})
);
assert_eq!(Host::parse(" "), None);
}
#[test]
fn a_snapshot_round_trips_into_rows() {
let json = r#"[{
"provider": "claude",
"surface": "cli",
"state": "waiting",
"session_id": "abc123",
"started_at": "2026-08-11T09:00:00Z",
"last_active": "2026-08-11T10:00:00Z",
"project": "/srv/work/api",
"title": null,
"model": "claude-opus-5",
"harness": "ClaudeCode",
"branch": "feat/idle",
"permission": "edits",
"running": true,
"process": {"cpu": 12.5, "memory": 4096, "command": "claude", "pids": 3},
"cost": {
"available": true, "total": "$1.25", "included": false, "free": false,
"this_hour": 0.5, "today": 1.25, "per_min": 0.01,
"by_day": {"2026-08-11": 1.25},
"by_hour": {"2026-08-11T10": 0.5}
},
"tokens": {"input": 1000, "output": 200, "total": 1200},
"tokens_per_min": 42.0,
"activity": {"tool_count": 40, "tool_errors": 4, "compactions": 2},
"context": {"used": 100000, "max": 200000},
"conflict": {"level": "file", "peers": ["def"], "files": ["/srv/work/api/x.rs"]}
}]"#;
let rows = parse("box", json).expect("parses");
let s = &rows[0];
assert_eq!(s.remote.as_ref().map(|r| r.host.as_str()), Some("box"));
assert_eq!(
crate::ui::columns::branch_of(s).as_deref(),
Some("feat/idle")
);
assert!(s.is_running(), "a live remote row must read as live");
assert_eq!(s.activity_state, ActivityState::WaitingForInput);
assert_eq!(s.total_cost, Some(1.25));
assert_eq!(s.cost_today, 1.25);
assert_eq!(s.cost_hour, 0.5);
assert_eq!(s.tool_errors, 4);
assert_eq!(s.error_rate(), Some(0.1));
assert_eq!(s.compactions, 2);
assert_eq!(s.conflict, Some(crate::collide::Overlap::File));
assert_eq!(s.process.as_ref().map(|p| p.memory), Some(4096));
assert_eq!(s.costs_by_day["2026-08-11"].values().sum::<f64>(), 1.25);
}
#[test]
fn a_newer_peer_rolling_hour_wins_over_its_clock_hour() {
let json = r#"[{
"provider": "claude", "session_id": "abc123",
"cost": {"available": true, "total": "$1.25", "included": false,
"this_hour": 0.0, "last_hour": 0.75}
}]"#;
let rows = parse("box", json).expect("parses");
assert_eq!(rows[0].cost_hour, 0.75);
}
#[test]
fn an_older_remote_still_produces_rows() {
let json = r#"[
{"provider": "codex", "session_id": "x", "project": "/w"},
{"provider": "codex"},
{"provider": "some-future-agent", "session_id": "y"}
]"#;
let rows = parse("box", json).expect("parses");
assert_eq!(
rows.len(),
1,
"no id and no known provider are both dropped"
);
assert_eq!(rows[0].session_id, "x");
assert!(!rows[0].is_running());
assert!(crate::ui::columns::branch_of(&rows[0]).is_none());
}
#[test]
fn a_version_probe_reads_what_the_far_side_said() {
assert_eq!(
probe_from(Ok("cctop 0.17.4\n".into())),
Probe::Version("0.17.4".into())
);
for why in [
"bash: line 1: cctop: command not found",
"zsh:1: command not found: cctop",
"bash: /opt/cctop: No such file or directory",
] {
assert_eq!(probe_from(Err(why.into())), Probe::Missing, "{why}");
}
assert!(matches!(
probe_from(Err(
"ssh: connect to host box port 22: Connection refused".into()
)),
Probe::Unreadable(_)
));
assert!(matches!(
probe_from(Ok("welcome to box!\n".into())),
Probe::Unreadable(_)
));
}
#[test]
fn skew_is_a_version_comparison_not_a_string_one() {
let v = |s: &str| Probe::Version(s.into());
assert_eq!(
skew(&v("0.17.4"), "0.18.1"),
Some(Skew::Older("0.17.4".into()))
);
assert_eq!(
skew(&v("0.10.0"), "0.9.3"),
Some(Skew::Newer("0.10.0".into()))
);
assert_eq!(skew(&v("0.18.1"), "0.18.1"), None);
assert_eq!(skew(&Probe::Missing, "0.18.1"), Some(Skew::Missing));
assert_eq!(skew(&Probe::Unreadable("?".into()), "0.18.1"), None);
}
#[test]
fn the_version_is_asked_once_per_connection() {
let rows = || Snapshot::Rows(Vec::new());
let down = || Snapshot::Failed("Connection timed out".into());
let gone = || Snapshot::Failed("bash: cctop: command not found".into());
let mut h = Handshake::default();
assert_eq!(h.after(&rows()), Ask::Probe);
assert_eq!(h.after(&rows()), Ask::Nothing);
assert_eq!(h.after(&down()), Ask::Nothing);
assert_eq!(h.after(&rows()), Ask::Probe, "a reconnect asks again");
assert_eq!(h.after(&gone()), Ask::Missing);
assert_eq!(h.after(&gone()), Ask::Nothing);
assert_eq!(h.after(&rows()), Ask::Probe, "installed since");
}
#[test]
fn the_update_runs_the_command_it_shows() {
let host = Host::parse("flo@box:/usr/local/bin/cctop").expect("a host");
assert_eq!(
host.shown_command(&["--update"]),
"ssh flo@box -- /usr/local/bin/cctop --update"
);
let argv = host.ssh_argv(&["--update"]);
assert_eq!(argv[0], "ssh");
assert!(argv.contains(&"-T".to_string()), "no pty: {argv:?}");
assert!(argv.contains(&"BatchMode=yes".to_string()));
assert_eq!(
&argv[argv.len() - 4..],
["flo@box", "--", "/usr/local/bin/cctop", "--update"]
);
}
#[test]
fn a_root_owned_remote_is_told_to_run_sudo_by_hand() {
let host = Host::parse("box").expect("a host");
let sudo = || {
Err(UpdateFailure::NeedsRoot(
"ssh -t box sudo cctop --update".into(),
))
};
let root = "Error: /usr/local/bin is not writable by this user, so the new binary \
cannot replace the old one: re-run it as `sudo cctop --update`.";
assert_eq!(update_outcome(&host, Err(root.into())), sudo());
assert_eq!(
update_outcome(&host, Err("Permission denied (os error 13)".into())),
sudo()
);
let key = "flo@box: Permission denied (publickey).";
assert_eq!(
update_outcome(&host, Err(key.into())),
Err(UpdateFailure::Other(key.into()))
);
let out = "Current version 0.17.4; checking for updates…\nDownloading x.tar.gz…\n\
Updated 0.17.4 -> 0.18.1.\n\nWhat changed since 0.17.4:\n 0.18.0\n";
assert_eq!(
update_outcome(&host, Ok(out.into())),
Ok("Updated 0.17.4 -> 0.18.1.".into())
);
}
#[test]
fn output_that_is_not_a_snapshot_is_an_error_not_an_empty_machine() {
assert!(parse("box", "command not found: cctop").is_err());
assert!(parse("box", "{}").is_err());
assert!(parse("box", "[]").expect("valid, just empty").is_empty());
}
}