| 1 | //! Opt-in product telemetry for Codewhale. |
| 2 | //! |
| 3 | //! The whole of what this crate may ever send is [`event`]. The whole of what |
| 4 | //! decides whether it may send anything is [`decision`]. Nothing else in the |
| 5 | //! tree is permitted to construct a payload or to reach the wire, and nothing in |
| 6 | //! here reads a prompt, a completion, a tool argument, a file, a path, a git |
| 7 | //! remote, a branch, a model id, a provider table name, an MCP server name, an |
| 8 | //! approval rule, an error body, a panic message, or a credential. |
| 9 | //! |
| 10 | //! # The shape of the guarantee |
| 11 | //! |
| 12 | //! Consent is a **value**, not a convention. [`decide`] is the only constructor |
| 13 | //! of [`TelemetryConsent`]; [`init`] takes one by value and there is no |
| 14 | //! bool-taking sibling. Six init sites cannot each drift from the predicate, |
| 15 | //! because they never see the predicate. |
| 16 | //! |
| 17 | //! Arming is a **`OnceLock`**, consulted by every write path including |
| 18 | //! [`record_blocking`]. This matters because the process panic hook is installed |
| 19 | //! before the command line is even parsed, long before any config resolution: it |
| 20 | //! cannot consult a resolved value, but it can consult a lock that is by |
| 21 | //! construction empty until resolution completes. A disabled user's panic |
| 22 | //! therefore writes nothing and creates no directory. |
| 23 | //! |
| 24 | //! Arming also **truncates** the buffer. No event recorded before consent can |
| 25 | //! ever be in the batch that follows it. |
| 26 | //! |
| 27 | //! # Failure posture |
| 28 | //! |
| 29 | //! Fail-open is absolute. Every fallible step ends in `.ok()?` or `let _ =`. |
| 30 | //! Nothing here returns an error to a caller, blocks a turn, blocks a tool, or |
| 31 | //! blocks process exit. Telemetry that costs a user their session is worse than |
| 32 | //! no telemetry. |
| 33 | |
| 34 | #![deny(missing_docs)] |
| 35 | |
| 36 | mod actor; |
| 37 | pub mod buffer; |
| 38 | pub mod client; |
| 39 | pub mod counters; |
| 40 | pub mod decision; |
| 41 | pub mod envelope; |
| 42 | pub mod event; |
| 43 | pub mod notice; |
| 44 | |
| 45 | #[cfg(test)] |
| 46 | mod tests; |
| 47 | |
| 48 | use std::sync::OnceLock; |
| 49 | use std::sync::atomic::{AtomicU8, Ordering}; |
| 50 | use std::time::Duration; |
| 51 | |
| 52 | pub use actor::{BATCH_MAX_BYTES, BATCH_MAX_EVENTS, FlushOutcome}; |
| 53 | pub use counters::{Counter, ErrorCounter, SessionCounters}; |
| 54 | pub use decision::{ |
| 55 | EndpointError, TELEMETRY_DIR, TelemetryConsent, TelemetryDecision, decide, decide_in_home, |
| 56 | re_decide, validate_endpoint, |
| 57 | }; |
| 58 | pub use envelope::reduce_panic_site; |
| 59 | pub use event::{ |
| 60 | Arch, Batch, ColdStartBucket, Counters, DurationBucket, Errors, Event, ExitClass, InstallKind, |
| 61 | Libc, Os, SCHEMA_VERSION, SessionSource, Surface, TurnWall, |
| 62 | }; |
| 63 | |
| 64 | /// How long the shutdown flush may hold the process. |
| 65 | /// |
| 66 | /// The terminal is still in alt-screen while this runs. The persistence actor's |
| 67 | /// unbounded `task.await` next door is not a pattern to copy: a hung TLS |
| 68 | /// handshake would hold a user's terminal past exit. |
| 69 | pub const SHUTDOWN_FLUSH_TIMEOUT: Duration = Duration::from_secs(3); |
| 70 | |
| 71 | /// Minimum gap between startup drains. |
| 72 | pub const STARTUP_DRAIN_INTERVAL_HOURS: i64 = 6; |
| 73 | |
| 74 | /// Everything a write path needs once the process is armed. |
| 75 | struct Armed { |
| 76 | handle: actor::Handle, |
| 77 | root: std::path::PathBuf, |
| 78 | exit_class: AtomicU8, |
| 79 | } |
| 80 | |
| 81 | /// The one gate. Unset means every write path is a hard no-op. |
| 82 | static ARMED: OnceLock<Armed> = OnceLock::new(); |
| 83 | |
| 84 | /// Arm telemetry for this process. |
| 85 | /// |
| 86 | /// Takes [`TelemetryConsent`] **by value**: there is no way to call this without |
| 87 | /// having gone through [`decide`], and no overload that accepts a `bool`. |
| 88 | /// |
| 89 | /// Idempotent — a second call is ignored, so a surface that dispatches twice |
| 90 | /// cannot start two writers against one buffer. |
| 91 | pub fn init(consent: TelemetryConsent) { |
| 92 | if ARMED.get().is_some() { |
| 93 | return; |
| 94 | } |
| 95 | let root = consent.root().to_path_buf(); |
| 96 | |
| 97 | // Clear the tombstone and drop anything buffered before consent. A stale |
| 98 | // buffer is not evidence of this user's answer. |
| 99 | if let Err(error) = buffer::arm(&root) { |
| 100 | tracing::debug!("telemetry could not prepare its buffer: {error}"); |
| 101 | return; |
| 102 | } |
| 103 | |
| 104 | let context = actor::Context { |
| 105 | root: root.clone(), |
| 106 | endpoint: consent.endpoint().map(str::to_string), |
| 107 | surface: consent.surface(), |
| 108 | config_path: consent.config_path().map(std::path::Path::to_path_buf), |
| 109 | app_version: env!("CARGO_PKG_VERSION").to_string(), |
| 110 | git_sha: envelope::release_build_sha(), |
| 111 | tty: envelope::current_tty(), |
| 112 | }; |
| 113 | |
| 114 | let _ = ARMED.set(Armed { |
| 115 | handle: actor::Handle::spawn(context), |
| 116 | root: root.clone(), |
| 117 | exit_class: AtomicU8::new(ExitClass::Clean.as_u8()), |
| 118 | }); |
| 119 | |
| 120 | record_install_or_upgrade(&root); |
| 121 | } |
| 122 | |
| 123 | /// Note that this binary's version differs from the one last seen on this |
| 124 | /// machine, at most once per version. |
| 125 | /// |
| 126 | /// The previous version comes from `$CODEWHALE_HOME/telemetry/state.json` and |
| 127 | /// from nowhere else. Session history and config mtimes would answer the same |
| 128 | /// question and carry a different privacy contract; reading them here would put |
| 129 | /// this crate one refactor away from the thread store. |
| 130 | /// |
| 131 | /// The state file is updated before the event is queued, so a process that dies |
| 132 | /// between the two reports nothing rather than reporting the same upgrade on |
| 133 | /// every launch. |
| 134 | fn record_install_or_upgrade(root: &std::path::Path) { |
| 135 | let current = env!("CARGO_PKG_VERSION"); |
| 136 | let mut state = envelope::read_state(root); |
| 137 | if state.last_version.as_deref() == Some(current) { |
| 138 | return; |
| 139 | } |
| 140 | let kind = match state.last_version.as_deref() { |
| 141 | None => InstallKind::Install, |
| 142 | Some(previous) if version_is_older(previous, current) => InstallKind::Upgrade, |
| 143 | Some(_) => InstallKind::Downgrade, |
| 144 | }; |
| 145 | let previous_version = state.last_version.clone(); |
| 146 | state.schema_version = SCHEMA_VERSION; |
| 147 | state.last_version = Some(current.to_string()); |
| 148 | if envelope::write_state(root, &state).is_err() { |
| 149 | // Nothing was recorded, so the next launch will try again. Emitting |
| 150 | // without the write would re-report the same upgrade forever. |
| 151 | return; |
| 152 | } |
| 153 | record(Event::InstallOrUpgrade { |
| 154 | kind, |
| 155 | previous_version, |
| 156 | }); |
| 157 | } |
| 158 | |
| 159 | /// Compare two dotted release numbers, ignoring any pre-release suffix. |
| 160 | /// |
| 161 | /// Deliberately not a semver dependency: the only question asked is which of |
| 162 | /// install / upgrade / downgrade to name, and a version this crate cannot parse |
| 163 | /// answers "not older", which reports a downgrade — the conservative direction, |
| 164 | /// since it never invents an upgrade that did not happen. |
| 165 | fn version_is_older(previous: &str, current: &str) -> bool { |
| 166 | fn parts(value: &str) -> Vec<u64> { |
| 167 | value |
| 168 | .split(['-', '+']) |
| 169 | .next() |
| 170 | .unwrap_or_default() |
| 171 | .split('.') |
| 172 | .map(|part| part.parse::<u64>().unwrap_or_default()) |
| 173 | .collect() |
| 174 | } |
| 175 | let (previous, current) = (parts(previous), parts(current)); |
| 176 | let width = previous.len().max(current.len()); |
| 177 | for index in 0..width { |
| 178 | let left = previous.get(index).copied().unwrap_or_default(); |
| 179 | let right = current.get(index).copied().unwrap_or_default(); |
| 180 | if left != right { |
| 181 | return left < right; |
| 182 | } |
| 183 | } |
| 184 | false |
| 185 | } |
| 186 | |
| 187 | /// Whether this process is armed. Every write path checks this first. |
| 188 | #[must_use] |
| 189 | pub fn is_armed() -> bool { |
| 190 | ARMED.get().is_some() |
| 191 | } |
| 192 | |
| 193 | /// This process's session accumulators. |
| 194 | /// |
| 195 | /// Deliberately **not** behind the arming gate. Every bump is a relaxed atomic |
| 196 | /// increment on a counter that never leaves this process unless [`init`] was |
| 197 | /// reached, so gating them would buy nothing and would put an `is_armed()` |
| 198 | /// branch on eleven hot call sites. The gate that matters is on the write |
| 199 | /// paths, and a snapshot of these numbers only ever reaches a payload through |
| 200 | /// one. |
| 201 | pub fn session_counters() -> &'static SessionCounters { |
| 202 | static COUNTERS: OnceLock<SessionCounters> = OnceLock::new(); |
| 203 | COUNTERS.get_or_init(SessionCounters::default) |
| 204 | } |
| 205 | |
| 206 | /// Queue an event for the writer thread. |
| 207 | /// |
| 208 | /// Non-blocking, and a no-op when unarmed. |
| 209 | pub fn record(event: Event) { |
| 210 | let Some(armed) = ARMED.get() else { |
| 211 | return; |
| 212 | }; |
| 213 | armed.handle.record(event); |
| 214 | } |
| 215 | |
| 216 | /// Write an event synchronously, without the writer thread and **without any |
| 217 | /// lock**. |
| 218 | /// |
| 219 | /// The synchronous escape hatch for the three paths where the async world is |
| 220 | /// gone or going: the panic hook, `record_caught_panic`, and the signal task |
| 221 | /// immediately before `std::process::exit`. One `O_APPEND` `write(2)` under |
| 222 | /// `PIPE_BUF`, a `sync_data`, and return — microseconds. |
| 223 | /// |
| 224 | /// Taking the compaction lock here would be a *blocking* acquisition on both of |
| 225 | /// those paths. `flock` is per-fd within a process, so an actor panic while |
| 226 | /// holding that lock would self-deadlock the hook, and a second Codewhale |
| 227 | /// process sharing `CODEWHALE_HOME` would hang Ctrl-C. |
| 228 | /// |
| 229 | /// A no-op when unarmed, which is what makes a disabled user's panic write |
| 230 | /// nothing and create no directory. |
| 231 | pub fn record_blocking(event: Event) { |
| 232 | let Some(armed) = ARMED.get() else { |
| 233 | return; |
| 234 | }; |
| 235 | let Ok(line) = serde_json::to_string(&event) else { |
| 236 | return; |
| 237 | }; |
| 238 | let path = buffer::buffer_path(&armed.root); |
| 239 | let _ = buffer::append(&armed.root, &path, &line); |
| 240 | } |
| 241 | |
| 242 | /// Record how this process is ending. |
| 243 | /// |
| 244 | /// Set by the panic hook, by the signal task before `std::process::exit`, and on |
| 245 | /// the clean path from the run's termination reason. **Never derived from an |
| 246 | /// exit code**: a cancelled turn and a SIGINT both exit 130, so a code-based |
| 247 | /// derivation would report every Esc as a signal. |
| 248 | pub fn set_exit_class(class: ExitClass) { |
| 249 | let Some(armed) = ARMED.get() else { |
| 250 | return; |
| 251 | }; |
| 252 | armed.exit_class.store(class.as_u8(), Ordering::Relaxed); |
| 253 | } |
| 254 | |
| 255 | /// The exit class recorded so far. `Clean` when unarmed or unset. |
| 256 | #[must_use] |
| 257 | pub fn exit_class() -> ExitClass { |
| 258 | ARMED.get().map_or(ExitClass::Clean, |armed| { |
| 259 | ExitClass::from_u8(armed.exit_class.load(Ordering::Relaxed)) |
| 260 | }) |
| 261 | } |
| 262 | |
| 263 | /// Flush whatever is buffered, waiting at most `deadline`. |
| 264 | /// |
| 265 | /// Blocking, so async callers must hand this to `spawn_blocking` and bound it — |
| 266 | /// [`SHUTDOWN_FLUSH_TIMEOUT`] is the teardown budget. Consent is re-resolved |
| 267 | /// from disk inside the writer thread before anything is sent. |
| 268 | /// |
| 269 | /// Returns [`FlushOutcome::Empty`] when unarmed. |
| 270 | pub fn flush_blocking(deadline: Duration) -> FlushOutcome { |
| 271 | ARMED |
| 272 | .get() |
| 273 | .map_or(FlushOutcome::Empty, |armed| armed.handle.flush(deadline)) |
| 274 | } |
| 275 | |
| 276 | /// Final flush, then stop the writer thread. |
| 277 | /// |
| 278 | /// Returns [`FlushOutcome::Empty`] when unarmed. |
| 279 | pub fn shutdown_blocking(deadline: Duration) -> FlushOutcome { |
| 280 | ARMED |
| 281 | .get() |
| 282 | .map_or(FlushOutcome::Empty, |armed| armed.handle.shutdown(deadline)) |
| 283 | } |
| 284 | |
| 285 | /// Whether a startup drain is due: a prior session crashed or was signalled and |
| 286 | /// left events behind, and enough time has passed since the last attempt. |
| 287 | /// |
| 288 | /// The check happens **before** the drain task is spawned, so "not due" means no |
| 289 | /// task at all rather than a task that returns early. |
| 290 | #[must_use] |
| 291 | pub fn startup_drain_due() -> bool { |
| 292 | let Some(armed) = ARMED.get() else { |
| 293 | return false; |
| 294 | }; |
| 295 | let path = buffer::buffer_path(&armed.root); |
| 296 | if buffer::read_lines(&path).is_empty() { |
| 297 | return false; |
| 298 | } |
| 299 | let state = envelope::read_state(&armed.root); |
| 300 | let Some(last) = state.last_flush else { |
| 301 | return true; |
| 302 | }; |
| 303 | let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(&last) else { |
| 304 | return true; |
| 305 | }; |
| 306 | chrono::Utc::now() |
| 307 | .signed_duration_since(parsed.with_timezone(&chrono::Utc)) |
| 308 | .num_hours() |
| 309 | >= STARTUP_DRAIN_INTERVAL_HOURS |
| 310 | } |
| 311 |