asimov_telemetry/
client.rs

1// This is free and unencumbered software released into the public domain.
2
3use crate::{Event, Outcome};
4use alloc::{
5    collections::BTreeMap,
6    format,
7    string::{String, ToString},
8    vec::Vec,
9};
10use asimov_http::proxy::{OriginProxy, ProxyOptions};
11use core::time::Duration;
12use reqwest::header::{HeaderMap, HeaderValue};
13use serde::Serialize;
14use serde_json::{Value, json};
15use std::{
16    fs::{self, File},
17    io::{self, BufRead, BufReader, Write},
18    path::Path,
19    sync::{Mutex, mpsc},
20    time::{SystemTime, UNIX_EPOCH},
21};
22
23const ENDPOINT: &str = "https://api.statsig.com/v1/log_event";
24const FLUSH_INTERVAL: Duration = Duration::from_secs(5 * 60);
25const STALE_AFTER: Duration = Duration::from_secs(60 * 60);
26const STALE_SHUTDOWN_WAIT: Duration = Duration::from_millis(500);
27const MAX_EVENT_AGE: Duration = Duration::from_secs(30 * 24 * 60 * 60);
28const BATCH_SIZE: usize = 500;
29
30pub struct Telemetry {
31    events: Mutex<File>,
32    user: User,
33    invocation_id: String,
34    caller: Caller,
35    stale_flush: Mutex<Option<mpsc::Receiver<()>>>,
36}
37
38impl Telemetry {
39    /// Events are appended to a per-invocation log in `directory`, which stays
40    /// locked while the process lives. Logs of exited processes are sent at most
41    /// once every five minutes.
42    ///
43    /// Uploads use shared environment/native proxy discovery from `asimov-http`.
44    /// Redirects are not followed; failed uploads remain queued for a later flush.
45    pub fn new(client_key: &str, app_version: &str, directory: &Path) -> Option<Self> {
46        if is_disabled(directory) {
47            return None;
48        }
49        Self::start(client_key, app_version, directory, ENDPOINT)
50    }
51
52    fn start(
53        client_key: &str,
54        app_version: &str,
55        directory: &Path,
56        endpoint: &str,
57    ) -> Option<Self> {
58        if !client_key.starts_with("client-") || client_key.len() <= "client-".len() {
59            return None;
60        }
61        let mut key = HeaderValue::from_str(client_key).ok()?;
62        key.set_sensitive(true);
63        fs::create_dir_all(directory).ok()?;
64        let id_path = directory.join("id");
65        let new_install = !id_path.exists();
66        if new_install {
67            let id = bs58::encode(dogma_uuid::Uuid::new_v4().as_bytes()).into_string();
68            fs::write(&id_path, id).ok()?;
69        }
70        let user_id = fs::read_to_string(id_path).ok()?;
71
72        let invocation_id = dogma_uuid::Uuid::now_v7().to_string();
73        let events =
74            File::create_new(directory.join(format!("events-{invocation_id}.jsonl"))).ok()?;
75        events.try_lock().ok()?;
76
77        let user = User {
78            user_id,
79            custom_ids: fs::read_to_string(directory.join("custom-ids.json"))
80                .ok()
81                .and_then(|ids| serde_json::from_str(&ids).ok())
82                .unwrap_or_default(),
83            app_version: app_version.to_string(),
84            custom: BTreeMap::from([
85                ("target_os", std::env::consts::OS),
86                ("target_arch", std::env::consts::ARCH),
87            ]),
88        };
89        let caller = Caller::detect();
90        if new_install {
91            let event = wire_event(
92                Event::Installed,
93                timestamp(),
94                &user,
95                &invocation_id,
96                &caller,
97            );
98            // On its own thread, as the caller may be inside an async runtime.
99            let sent = std::thread::scope(|scope| {
100                scope
101                    .spawn(|| send(key.clone(), endpoint, core::slice::from_ref(&event)))
102                    .join()
103            });
104            if !matches!(sent, Ok(Some(()))) {
105                let _ = writeln!(&events, "{event}");
106            }
107        }
108
109        let age = |name: &str| {
110            fs::metadata(directory.join(name))
111                .and_then(|metadata| metadata.modified())
112                .map(|time| SystemTime::now().duration_since(time).unwrap_or_default())
113                .unwrap_or(Duration::MAX)
114        };
115        let flush = (age("attempted") >= FLUSH_INTERVAL)
116            .then(|| {
117                let (sender, receiver) = mpsc::channel();
118                let directory = directory.to_path_buf();
119                let endpoint = endpoint.to_string();
120                std::thread::Builder::new()
121                    .name("asimov-telemetry".into())
122                    .spawn(move || {
123                        flush(&directory, key, &endpoint);
124                        let _ = sender.send(());
125                    })
126                    .ok()
127                    .map(|_| receiver)
128            })
129            .flatten();
130
131        Some(Self {
132            events: Mutex::new(events),
133            user,
134            invocation_id,
135            caller,
136            stale_flush: Mutex::new(flush.filter(|_| age("flushed") >= STALE_AFTER)),
137        })
138    }
139
140    pub fn log(&self, event: Event) {
141        let event = wire_event(
142            event,
143            timestamp(),
144            &self.user,
145            &self.invocation_id,
146            &self.caller,
147        );
148        if let Ok(mut events) = self.events.lock() {
149            let _ = writeln!(events, "{event}");
150        }
151    }
152
153    /// Only waits for an in-flight flush when the last successful one is stale.
154    pub fn shutdown(&self) {
155        let flush = self
156            .stale_flush
157            .lock()
158            .ok()
159            .and_then(|mut flush| flush.take());
160        if let Some(flush) = flush {
161            let _ = flush.recv_timeout(STALE_SHUTDOWN_WAIT);
162        }
163    }
164}
165
166/// Whether `ASIMOV_TELEMETRY` is `0`, `false`, `no` or `off`, `DO_NOT_TRACK` is `1`, or
167/// [`disable`] was called for `directory`.
168pub fn is_disabled(directory: &Path) -> bool {
169    std::env::var("ASIMOV_TELEMETRY").is_ok_and(|value| {
170        matches!(
171            value.trim().to_ascii_lowercase().as_str(),
172            "0" | "false" | "no" | "off"
173        )
174    }) || std::env::var("DO_NOT_TRACK").is_ok_and(|value| value.trim() == "1")
175        || directory.join("disabled").exists()
176}
177
178pub fn enable(directory: &Path) -> io::Result<()> {
179    match fs::remove_file(directory.join("disabled")) {
180        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
181        result => result,
182    }
183}
184
185/// Also discards any unsent events.
186pub fn disable(directory: &Path) -> io::Result<()> {
187    fs::create_dir_all(directory)?;
188    File::create(directory.join("disabled"))?;
189    for entry in fs::read_dir(directory)?.flatten() {
190        if entry.file_name().to_string_lossy().starts_with("events-") {
191            fs::remove_file(entry.path())?;
192        }
193    }
194    Ok(())
195}
196
197fn flush(directory: &Path, key: HeaderValue, endpoint: &str) -> Option<()> {
198    let touch = |name: &str| {
199        File::options()
200            .create(true)
201            .truncate(false)
202            .write(true)
203            .open(directory.join(name))
204            .and_then(|file| file.set_modified(SystemTime::now()))
205            .ok()
206    };
207    let lock = File::create(directory.join("flush.lock")).ok()?;
208    lock.try_lock().ok()?;
209    touch("attempted")?;
210
211    let mut logs = Vec::new();
212    for path in fs::read_dir(directory)
213        .ok()?
214        .flatten()
215        .map(|entry| entry.path())
216    {
217        let is_log = path
218            .file_name()
219            .and_then(|name| name.to_str())
220            .is_some_and(|name| name.starts_with("events-") && name.ends_with(".jsonl"));
221        let Some(file) = is_log.then(|| File::open(&path).ok()).flatten() else {
222            continue;
223        };
224        if file.try_lock().is_err() {
225            continue;
226        }
227        let expired = file
228            .metadata()
229            .and_then(|metadata| metadata.modified())
230            .is_ok_and(|time| time.elapsed().is_ok_and(|age| age > MAX_EVENT_AGE));
231        if expired {
232            drop(file);
233            let _ = fs::remove_file(&path);
234            continue;
235        }
236        logs.push((path, file));
237    }
238
239    let events: Vec<Value> = logs
240        .iter()
241        .flat_map(|(_, file)| BufReader::new(file).lines().map_while(Result::ok))
242        .filter_map(|line| serde_json::from_str(&line).ok())
243        .collect();
244    if !events.is_empty() {
245        send(key, endpoint, &events)?;
246    }
247
248    for (path, file) in logs {
249        drop(file);
250        let _ = fs::remove_file(path);
251    }
252    touch("flushed")
253}
254
255fn send(key: HeaderValue, endpoint: &str, events: &[Value]) -> Option<()> {
256    let endpoint = reqwest::Url::parse(endpoint).ok()?;
257    let proxy = OriginProxy::resolve(&endpoint, &ProxyOptions::default()).ok()?;
258    let mut headers = HeaderMap::new();
259    headers.insert("statsig-api-key", key);
260    headers.insert("statsig-sdk-type", HeaderValue::from_static("asimov.rs"));
261    headers.insert(
262        "statsig-sdk-version",
263        HeaderValue::from_static(env!("CARGO_PKG_VERSION")),
264    );
265    let builder = reqwest::Client::builder()
266        .default_headers(headers)
267        .timeout(Duration::from_secs(10));
268    let client = proxy.configure_reqwest(builder).ok()?.build().ok()?;
269    let runtime = tokio::runtime::Builder::new_current_thread()
270        .enable_all()
271        .build()
272        .ok()?;
273    runtime.block_on(async {
274        for batch in events.chunks(BATCH_SIZE) {
275            // Every batch uses the exact URL whose proxy policy was resolved.
276            let response = client
277                .post(endpoint.clone())
278                .header("statsig-client-time", timestamp().to_string())
279                .json(&json!({"events": batch}))
280                .send()
281                .await
282                .ok()?;
283            if !response.status().is_success() {
284                return None;
285            }
286        }
287        Some(())
288    })
289}
290
291fn timestamp() -> u128 {
292    SystemTime::now()
293        .duration_since(UNIX_EPOCH)
294        .unwrap_or_default()
295        .as_millis()
296}
297
298struct Caller {
299    kind: &'static str,
300    agent: &'static str,
301    source: &'static str,
302}
303
304impl Caller {
305    fn detect() -> Self {
306        let Some(agent) = agent_detector::detect() else {
307            return Self {
308                kind: "unknown",
309                agent: "unknown",
310                source: "none",
311            };
312        };
313        Self {
314            kind: "agent",
315            agent: Self::agent_name(&agent.name),
316            source: match agent.source {
317                agent_detector::DetectionSource::ParentProcess => "process",
318                agent_detector::DetectionSource::StandardEnvVar
319                | agent_detector::DetectionSource::ToolEnvVar => "environment",
320                _ => "unknown",
321            },
322        }
323    }
324
325    fn agent_name(name: &str) -> &'static str {
326        // `AI_AGENT` may carry a suffix, e.g. Claude Code sets `claude-code_2-1-286_agent`.
327        [
328            "pi",
329            "hermes",
330            "claude-code",
331            "codex",
332            "openclaw",
333            "opencode",
334            "cursor",
335            "cursor-cli",
336            "gemini",
337            "cowork",
338            "aider",
339            "github-copilot",
340            "goose",
341            "cline",
342            "oh-my-pi",
343        ]
344        .into_iter()
345        .find(|known| {
346            name.strip_prefix(known).is_some_and(|suffix| {
347                !suffix.starts_with(|c: char| c.is_ascii_alphanumeric() || c == '-')
348            })
349        })
350        .unwrap_or("other")
351    }
352}
353
354#[derive(Serialize)]
355#[serde(rename_all = "camelCase")]
356struct User {
357    #[serde(rename = "userID")]
358    user_id: String,
359    #[serde(rename = "customIDs", skip_serializing_if = "BTreeMap::is_empty")]
360    custom_ids: BTreeMap<String, String>,
361    app_version: String,
362    custom: BTreeMap<&'static str, &'static str>,
363}
364
365fn wire_event(
366    event: Event,
367    time: u128,
368    user: &User,
369    invocation_id: &str,
370    caller: &Caller,
371) -> Value {
372    let mut metadata = BTreeMap::from([
373        ("schema_version", "1".to_string()),
374        ("invocation_id", invocation_id.to_string()),
375        ("caller_kind", caller.kind.to_string()),
376        ("agent_name", caller.agent.to_string()),
377        ("detection_source", caller.source.to_string()),
378    ]);
379    let name = match event {
380        Event::Installed => "cli_installed",
381        Event::CommandStarted { command, modules } => {
382            metadata.insert("command", command);
383            if let [module] = modules.as_slice() {
384                metadata.insert("module", module.clone());
385            }
386            metadata.insert("modules", json!(modules).to_string());
387            "cli_command_started"
388        },
389        Event::CommandFinished {
390            command,
391            modules,
392            exit_code,
393            duration,
394        } => {
395            metadata.insert("command", command);
396            if let [module] = modules.as_slice() {
397                metadata.insert("module", module.clone());
398            }
399            metadata.insert("modules", json!(modules).to_string());
400            metadata.insert("exit_code", exit_code.to_string());
401            metadata.insert(
402                "outcome",
403                if exit_code == 0 { "success" } else { "failure" }.to_string(),
404            );
405            metadata.insert("duration_ms", duration.as_millis().to_string());
406            "cli_command_finished"
407        },
408        Event::ModuleOperationFinished {
409            operation,
410            module,
411            outcome,
412            duration,
413        } => {
414            metadata.insert(
415                "operation",
416                match operation {
417                    crate::Operation::Fetch => "fetch",
418                    crate::Operation::Read => "read",
419                    crate::Operation::List => "list",
420                }
421                .to_string(),
422            );
423            metadata.insert("module", module);
424            metadata.insert(
425                "outcome",
426                match outcome {
427                    Outcome::Success => "success",
428                    Outcome::Failure => "failure",
429                }
430                .to_string(),
431            );
432            metadata.insert("duration_ms", duration.as_millis().to_string());
433            "cli_module_operation_finished"
434        },
435    };
436    json!({"eventName": name, "user": user, "time": time, "metadata": metadata})
437}
438
439#[cfg(test)]
440mod tests;
asimov_telemetry/client.rs

asimov_telemetry/
client.rs
