diff --git a/scripts/prepare-berdctl-sidecar.sh b/scripts/prepare-berdctl-sidecar.sh index 0ed182741..d6af0ea9e 100755 --- a/scripts/prepare-berdctl-sidecar.sh +++ b/scripts/prepare-berdctl-sidecar.sh @@ -51,6 +51,10 @@ BUNDLE_BERD_CALL=0 if [[ "$TRIPLE" == *apple-darwin && "${BERD_CALL_BUNDLE:-1}" == "1" ]]; then BUNDLE_BERD_CALL=1 CARGO_ARGS+=(-p berd-call) + # The CLI's twin of the app's block-telemetry-enforced, from the same gate. + if [[ "${VITE_TELEMETRY_ENFORCED:-0}" == "1" ]]; then + CARGO_ARGS+=(--features berd-call/telemetry-enforced) + fi fi (cd src-tauri && cargo "${CARGO_ARGS[@]}") diff --git a/scripts/release/build-macos.sh b/scripts/release/build-macos.sh index 5ce772191..68fc96194 100755 --- a/scripts/release/build-macos.sh +++ b/scripts/release/build-macos.sh @@ -463,7 +463,19 @@ echo "+++ :hammer: pnpm tauri build (unsigned)" GOOSE_BUILD_PROFILE=release ./scripts/prepare-goose-sidecar.sh # ACP bridges are installed into the managed Node runtime on demand; they are # no longer staged as build resources. -VITE_FEEDBACK="$VITE_FEEDBACK_VALUE" ./scripts/prepare-berdctl-sidecar.sh "$TARGET_TRIPLE" +# berd-call is compiled here rather than under the VITE env below, so it gets +# the app's telemetry policy explicitly. An empty environment, the CLI's twin of +# the renderer's VITE_TELEMETRY=0, compiles its telemetry out. +BERD_CALL_TELEMETRY_ENVIRONMENT="$VITE_ENVIRONMENT_VALUE" +for pair in ${VITE_EXTRA_ENV[@]+"${VITE_EXTRA_ENV[@]}"}; do + if [[ "$pair" == "VITE_TELEMETRY=0" ]]; then BERD_CALL_TELEMETRY_ENVIRONMENT=""; fi +done +BERD_CALL_TELEMETRY_CHANNEL="$(jq -r '.telemetry.channel // "public"' distro/distro.json 2>/dev/null || echo public)" +BERD_CALL_VERSION="$RELEASE_VERSION" \ + BERD_CALL_TELEMETRY_ENVIRONMENT="$BERD_CALL_TELEMETRY_ENVIRONMENT" \ + BERD_CALL_TELEMETRY_CHANNEL="$BERD_CALL_TELEMETRY_CHANNEL" \ + VITE_TELEMETRY_ENFORCED="$VITE_TELEMETRY_ENFORCED_VALUE" \ + VITE_FEEDBACK="$VITE_FEEDBACK_VALUE" ./scripts/prepare-berdctl-sidecar.sh "$TARGET_TRIPLE" if [[ "$VITE_AGENT_TOOLS_VALUE" == "1" ]]; then ./scripts/prepare-bb-cli-resource.sh "$TARGET_TRIPLE" tmp="$(mktemp)" diff --git a/scripts/release/tests/release-scripts.test.mjs b/scripts/release/tests/release-scripts.test.mjs index 919540594..17895d804 100644 --- a/scripts/release/tests/release-scripts.test.mjs +++ b/scripts/release/tests/release-scripts.test.mjs @@ -590,6 +590,15 @@ describe("build-macos Block-service feature seam", () => { ); expect(config.bundle.externalBin).toContain("binaries/berd-call"); expect(stage).toContain("CARGO_ARGS+=(-p berd-call)"); + expect(stage).toContain( + "CARGO_ARGS+=(--features berd-call/telemetry-enforced)", + ); + expect(release).toContain( + 'VITE_TELEMETRY_ENFORCED="$VITE_TELEMETRY_ENFORCED_VALUE" \\\n VITE_FEEDBACK="$VITE_FEEDBACK_VALUE" ./scripts/prepare-berdctl-sidecar.sh', + ); + expect(release).toContain( + 'if [[ "$pair" == "VITE_TELEMETRY=0" ]]; then BERD_CALL_TELEMETRY_ENVIRONMENT=""; fi', + ); expect(stage).toContain("stage_cli berd-call"); expect(stage).toContain('out="$OUT_DIR/$name-$TRIPLE"'); expect(release).toContain( diff --git a/src-tauri/crates/berd-call/Cargo.toml b/src-tauri/crates/berd-call/Cargo.toml index d02f9ca5f..fc17031dc 100644 --- a/src-tauri/crates/berd-call/Cargo.toml +++ b/src-tauri/crates/berd-call/Cargo.toml @@ -16,6 +16,9 @@ path = "src/main.rs" default = ["static"] static = ["sherpa-onnx/static"] shared = ["sherpa-onnx/shared"] +# Internal builds report usage without a consent setting, like the app's +# block-telemetry-enforced. +telemetry-enforced = [] [dependencies] audioadapter-buffers = "3.0" diff --git a/src-tauri/crates/berd-call/src/cli_help.rs b/src-tauri/crates/berd-call/src/cli_help.rs index a46e22074..7353214fe 100644 --- a/src-tauri/crates/berd-call/src/cli_help.rs +++ b/src-tauri/crates/berd-call/src/cli_help.rs @@ -25,6 +25,7 @@ Diagnostics: benchmark stt Benchmark an STT backend Other: + telemetry Show or change usage telemetry consent help [command] Show help for a command version Print the installed version @@ -108,6 +109,17 @@ const STOP_HELP: &str = r#"Stop the active local voice call. Usage: berd-call stop [--port PORT]"#; +const TELEMETRY_HELP: &str = r#"Show or change usage telemetry consent. + +Usage: + berd-call telemetry [status | enable | disable] + +Calls report anonymous usage (start, end, duration, turn counts, end reason, +audio backends, speech rate and voice, with the app version and a random +installation id) only with consent. Consent is shared with Berd: enabling or disabling +here changes Berd's telemetry setting too. Internal builds always report. +Results are JSON with enabled and enforced."#; + const SESSION_HELP: &str = r#"Run the host-facing framed voice-session protocol. Usage: @@ -213,6 +225,7 @@ fn help_for(topic: &[&str]) -> Option<&'static str> { ["catch-up"] | ["wait-for-input"] => Some(POLL_INPUT_HELP), ["settings"] => Some(SETTINGS_HELP), ["stop"] => Some(STOP_HELP), + ["telemetry"] | ["telemetry", "status" | "enable" | "disable"] => Some(TELEMETRY_HELP), ["session"] => Some(SESSION_HELP), ["synthesize"] => Some(SYNTHESIZE_HELP), ["voices"] | ["voices", "list" | "download" | "preview"] => Some(VOICES_HELP), diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index 9699a94f5..805697456 100644 --- a/src-tauri/crates/berd-call/src/host_session.rs +++ b/src-tauri/crates/berd-call/src/host_session.rs @@ -71,7 +71,7 @@ pub(crate) fn run(options: StartOptions) -> Result<(), String> { .unwrap_or_else(default_input_during_tts_policy), restarted: None, }; - loop { + let outcome = loop { match run_session( &server, session, @@ -80,32 +80,28 @@ pub(crate) fn run(options: StartOptions) -> Result<(), String> { &stop_requested, &status_sounds, ) { - Ok(None) => { - transcript.finish(CLEAN_STOP); - return Ok(()); - } + Ok(None) => break Ok(()), // A stop recorded while the old session was shutting down wins // over the restart it would otherwise hand off to. Ok(Some(next)) if stop_requested.load(Ordering::SeqCst) => { if let Some(response) = next.restarted { let _ = response.send(Err("voice call stopped".into())); } - transcript.finish(CLEAN_STOP); - return Ok(()); + break Ok(()); } Ok(Some(next)) => { session = next; } - Err(_) if stop_requested.load(Ordering::SeqCst) => { - transcript.finish(CLEAN_STOP); - return Ok(()); - } - Err(message) => { - transcript.finish(&format!("Berd Call failed: {message}")); - return Err(message); - } + Err(_) if stop_requested.load(Ordering::SeqCst) => break Ok(()), + Err(message) => break Err(message), } + }; + match &outcome { + Ok(()) => transcript.finish(CLEAN_STOP), + Err(message) => transcript.finish(&format!("Berd Call failed: {message}")), } + crate::telemetry::end(outcome.is_err()); + outcome } /// Where live transcript records go: nowhere, stdout TSV, or a Codex task. @@ -268,6 +264,13 @@ fn run_session( )?; let _ = response.send(control.status()); } + if let Ok(tts) = actor + .session + .lock() + .map(|session| session.tts.settings.clone()) + { + crate::telemetry::start(&control.session_arguments, &tts); + } let control: Arc = control; while running.load(Ordering::SeqCst) { server.poll(Arc::clone(&control))?; @@ -758,6 +761,7 @@ impl HostControl for SessionControl { fn stop(&self) -> Result { // Recorded before enqueueing, so a stop that races a restart handoff // still ends the call even if this session never reads the command. + crate::telemetry::user_stopped(); self.stop_requested.store(true, Ordering::SeqCst); let (tx, rx) = mpsc::sync_channel(1); if self @@ -1370,6 +1374,7 @@ impl SessionActor { self.stream_user_interim(token, &text)? } SessionMessage::LiveEvent { token, text } => { + crate::telemetry::user_utterance(); self.transcript.record(Some(token), "user", &text)? } SessionMessage::Pending { id, utterances } if self.pending_prepare_id() == Some(id) => { @@ -1400,6 +1405,7 @@ impl SessionActor { } } SessionMessage::SpeechCompleted { id, .. } if self.pending_prepare_id() == Some(id) => { + crate::telemetry::assistant_response(); self.respond_speak(Ok(json!({"spoke":true,"status":"completed"}))); } SessionMessage::SpeechInterrupted { diff --git a/src-tauri/crates/berd-call/src/lib.rs b/src-tauri/crates/berd-call/src/lib.rs index 4198042da..b3c21877c 100644 --- a/src-tauri/crates/berd-call/src/lib.rs +++ b/src-tauri/crates/berd-call/src/lib.rs @@ -32,6 +32,7 @@ pub mod session; pub mod siri; mod status_sounds; mod synthesis; +pub mod telemetry_store; mod tts; #[cfg(target_os = "macos")] pub mod update_guard; diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index fc2a0ce8a..06f811c17 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -56,6 +56,7 @@ mod session_audio; mod session_framing; #[cfg(target_os = "macos")] mod system_input_mute; +mod telemetry; use session_audio::{ AudioHostAck, AudioOutputControlRequest, AudioPipeTransport, RemotePcmAudioOutput, @@ -552,6 +553,12 @@ fn main() { print_pretty_json(&response) .unwrap_or_else(|error| operational_error(operation, error)); } + Some("telemetry") => { + let response = telemetry::command(&args[2..]) + .unwrap_or_else(|error| operational_error("telemetry", error)); + print_pretty_json(&response) + .unwrap_or_else(|error| operational_error("telemetry", error)); + } Some("settings") => { let (port, request) = parse_or_exit(parse_host_settings_args(&args), &args); let response = host_control::request(port, request) diff --git a/src-tauri/crates/berd-call/src/telemetry.rs b/src-tauri/crates/berd-call/src/telemetry.rs new file mode 100644 index 000000000..6013d3faa --- /dev/null +++ b/src-tauri/crates/berd-call/src/telemetry.rs @@ -0,0 +1,737 @@ +//! Usage telemetry for standalone calls: one start and one end event per call, +//! in Berd's voice-conversation shape under berd-call's own service and scope, +//! so CLI calls stay distinguishable from in-app calls. Consent is Berd's +//! shared preference (`berd_call::telemetry_store`), forced on only in builds +//! with `telemetry-enforced`. A build without a compiled-in environment never +//! touches the network. Sending is best effort and never fails a call; the +//! only cost to a call is the bounded flush when it ends. +#![cfg_attr(not(target_os = "macos"), allow(dead_code))] + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{mpsc, Mutex}; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use berd_call::telemetry_store as store; +use berd_call::TtsSettings; +use serde_json::{json, Value}; + +const SCHEMA_VERSION: &str = "berd-call-otlp-logs-v1"; +const ENFORCED: bool = cfg!(feature = "telemetry-enforced"); +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +const FLUSH_TIMEOUT: Duration = Duration::from_secs(5); + +/// What a release build reports as, compiled in by `scripts/release/build-macos.sh`. +#[derive(Clone, Copy)] +struct Build { + origin: &'static str, + environment: &'static str, + channel: &'static str, + version: &'static str, +} + +fn build() -> Option { + let (origin, environment) = match option_env!("BERD_CALL_TELEMETRY_ENVIRONMENT") { + Some("production") => ("https://otel.berd.xyz", "production"), + Some("staging") => ("https://otel.test.blockstaging.build", "staging"), + _ => return None, + }; + Some(Build { + origin, + environment, + channel: match option_env!("BERD_CALL_TELEMETRY_CHANNEL") { + Some("internal") => "internal", + _ => "public", + }, + version: option_env!("BERD_CALL_VERSION").unwrap_or(env!("CARGO_PKG_VERSION")), + }) +} + +/// Berd.app's data dir (Tauri's app-data dir for `xyz.block.berd`), which +/// holds the shared consent and installation id. +fn berd_dir() -> Option { + let home = std::env::var_os("HOME").map(PathBuf::from); + #[cfg(target_os = "macos")] + let root = home?.join("Library/Application Support"); + #[cfg(not(target_os = "macos"))] + let root = std::env::var_os("XDG_DATA_HOME") + .map(PathBuf::from) + .or_else(|| home.map(|home| home.join(".local/share")))?; + Some(root.join("xyz.block.berd")) +} + +fn consented(dir: &Path) -> bool { + ENFORCED || store::load_telemetry_consent(dir).unwrap_or(false) +} + +fn ensure_consent(dir: &Path) -> Result<(), String> { + consented(dir) + .then_some(()) + .ok_or_else(|| "telemetry is disabled".to_string()) +} + +/// `berd-call telemetry [status|enable|disable]`. +pub(crate) fn command(args: &[String]) -> Result { + let dir = berd_dir().ok_or("cannot locate Berd's settings: HOME is unset")?; + command_in(&dir, &args.iter().map(String::as_str).collect::>()) +} + +fn command_in(dir: &Path, args: &[&str]) -> Result { + match args { + [] | ["status"] => {} + ["enable" | "disable"] if ENFORCED => { + return Err("telemetry is always enabled in this build".into()) + } + [action @ ("enable" | "disable")] => { + store::persist_telemetry_consent(dir, *action == "enable")? + } + _ => return Err("usage: berd-call telemetry [status|enable|disable]".into()), + } + Ok(json!({"enabled": consented(dir), "enforced": ENFORCED})) +} + +/// The call's audio setup, as Berd's voice events report it. +#[derive(Clone, Debug, PartialEq)] +struct Context { + input: &'static str, + output: &'static str, + rate: Option, + voice: Option, +} + +impl Context { + /// Reports the selected voice only from its backend's own catalog, and any + /// other voice as "custom", so free-form names never leave the machine. + fn new( + session_arguments: &[String], + tts: &TtsSettings, + siri_voices: impl FnOnce() -> Vec, + ) -> Option { + let input = session_arguments + .windows(2) + .rfind(|pair| pair[0] == "--stt-backend") + .map_or("macos", |pair| pair[1].as_str()); + let (output, known) = match tts { + TtsSettings::OpenAi { voice, .. } => ( + "openai", + berd_call::openai::OPENAI_VOICE_IDS.contains(&voice.as_str()), + ), + TtsSettings::Siri { voice, .. } => ("siri", siri_voices().contains(voice)), + TtsSettings::Pocket { voice, .. } => ( + "pocket", + berd_call::pocket_assets::voices() + .iter() + .any(|known| known.id == voice), + ), + }; + let (rate, voice) = (tts.rate(), tts.voice()); + Some(Self { + input: ["macos", "parakeet", "openai"] + .into_iter() + .find(|known| *known == input)?, + output, + rate: (0.25..=4.0).contains(&rate).then(|| rate.to_string()), + voice: (!voice.is_empty()).then(|| if known { voice } else { "custom" }.to_string()), + }) + } + + fn attributes(&self) -> Vec<(&'static str, String)> { + let mut attributes = vec![ + ("input_backend", self.input.to_string()), + ("output_backend", self.output.to_string()), + ("voice_mode", "chained".to_string()), + ]; + attributes.extend(self.rate.clone().map(|rate| ("tts_rate", rate))); + attributes.extend(self.voice.clone().map(|voice| ("tts_voice", voice))); + attributes + } +} + +struct Call { + context: Context, + started: Instant, + records: mpsc::Sender, + sent: mpsc::Receiver<()>, +} + +static CALL: Mutex> = Mutex::new(None); +static USER_UTTERANCES: AtomicU64 = AtomicU64::new(0); +static ASSISTANT_RESPONSES: AtomicU64 = AtomicU64::new(0); +static USER_STOPPED: AtomicBool = AtomicBool::new(false); + +pub(crate) fn user_utterance() { + USER_UTTERANCES.fetch_add(1, Ordering::Relaxed); +} + +pub(crate) fn assistant_response() { + ASSISTANT_RESPONSES.fetch_add(1, Ordering::Relaxed); +} + +pub(crate) fn user_stopped() { + USER_STOPPED.store(true, Ordering::Relaxed); +} + +/// Reports the call's start once, when its first session is ready. +pub(crate) fn start(session_arguments: &[String], tts: &TtsSettings) { + let (Some(build), Some(dir)) = (build(), berd_dir()) else { + return; + }; + let Ok(mut call) = CALL.lock() else { return }; + if call.is_some() || !consented(&dir) { + return; + } + let Some(context) = Context::new(session_arguments, tts, || { + berd_call::siri::load_voice_catalog(None) + .map(|catalog| catalog.voices.into_iter().map(|voice| voice.name).collect()) + .unwrap_or_default() + }) else { + return; + }; + let (records, queued) = mpsc::channel(); + let (done, sent) = mpsc::channel(); + let worker = thread::Builder::new() + .name("berd-call-telemetry".into()) + .spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build(); + if let (Ok(runtime), Ok(client)) = (runtime, client()) { + for record in queued { + let _ = runtime.block_on(export(&client, build, &dir, record)); + } + } + let _ = done.send(()); + }); + if worker.is_err() { + return; + } + // Consent granted mid-call starts reporting at a later session, so the + // counts begin with the started event rather than with the process. + USER_UTTERANCES.store(0, Ordering::Relaxed); + ASSISTANT_RESPONSES.store(0, Ordering::Relaxed); + let _ = records.send(record( + "berd_voice_conversation_started", + SystemTime::now(), + &context.attributes(), + )); + *call = Some(Call { + context, + started: Instant::now(), + records, + sent, + }); +} + +/// Reports the call's end and waits briefly for queued events to send. +pub(crate) fn end(failed: bool) { + let Some(call) = CALL.lock().ok().and_then(|mut call| call.take()) else { + return; + }; + let reason = if failed { + "error" + } else if USER_STOPPED.load(Ordering::Relaxed) { + "user" + } else { + "clean-shutdown" + }; + let _ = call.records.send(record( + "berd_voice_conversation_ended", + SystemTime::now(), + &ended_attributes( + &call.context, + call.started.elapsed(), + USER_UTTERANCES.load(Ordering::Relaxed), + ASSISTANT_RESPONSES.load(Ordering::Relaxed), + reason, + ), + )); + drop(call.records); + let _ = call.sent.recv_timeout(FLUSH_TIMEOUT); +} + +fn ended_attributes( + context: &Context, + duration: Duration, + user_utterances: u64, + assistant_responses: u64, + reason: &'static str, +) -> Vec<(&'static str, String)> { + let mut attributes = context.attributes(); + attributes.extend([ + ("duration_ms", duration.as_millis().to_string()), + ("user_utterance_count", user_utterances.to_string()), + ("assistant_response_count", assistant_responses.to_string()), + ("end_reason", reason.to_string()), + ]); + attributes +} + +fn attributes<'a>(pairs: impl IntoIterator) -> Value { + pairs + .into_iter() + .map(|(key, value)| json!({"key": key, "value": {"stringValue": value}})) + .collect() +} + +fn record(event: &str, at: SystemTime, event_attributes: &[(&'static str, String)]) -> Value { + let nanos = at + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + .to_string(); + json!({ + "timeUnixNano": nanos, + "observedTimeUnixNano": nanos, + "body": {}, + "eventName": event, + "attributes": attributes(event_attributes.iter().map(|(key, value)| (*key, value.as_str()))), + "droppedAttributesCount": 0, + }) +} + +fn payload(build: Build, installation_id: &str, record: Value) -> Value { + json!({"resourceLogs": [{ + "resource": { + "attributes": attributes([ + ("service.name", "berd-call"), + ("service.version", build.version), + ("deployment.environment", build.environment), + ("distribution.channel", build.channel), + ("installation.id", installation_id), + ]), + "droppedAttributesCount": 0, + }, + "scopeLogs": [{"scope": {"name": "berd-call.telemetry"}, "logRecords": [record]}], + }]}) +} + +fn client() -> reqwest::Result { + reqwest::Client::builder() + .connect_timeout(CONNECT_TIMEOUT) + .timeout(REQUEST_TIMEOUT) + .redirect(reqwest::redirect::Policy::none()) + .build() +} + +#[derive(serde::Deserialize)] +struct Bootstrap { + token: String, +} + +/// Bootstraps an upload token and posts one record, re-reading consent before +/// each request so a revocation (from the CLI or Berd.app) stops the next one. +async fn export( + client: &reqwest::Client, + build: Build, + dir: &Path, + record: Value, +) -> Result<(), String> { + ensure_consent(dir)?; + let installation_id = store::load_or_create_installation_id(dir)?; + let failed = |error: reqwest::Error| error.to_string(); + let bootstrap: Bootstrap = client + .post(format!("{}/v1/bootstrap", build.origin)) + .header(reqwest::header::ACCEPT, "application/json") + .json(&json!({"installationId": installation_id})) + .send() + .await + .and_then(reqwest::Response::error_for_status) + .map_err(failed)? + .json() + .await + .map_err(failed)?; + ensure_consent(dir)?; + client + .post(format!("{}/v1/logs", build.origin)) + .header(reqwest::header::ACCEPT, "application/json") + .header("x-berd-schema-version", SCHEMA_VERSION) + .bearer_auth(bootstrap.token) + .json(&payload(build, &installation_id, record)) + .send() + .await + .and_then(reqwest::Response::error_for_status) + .map_err(failed)?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::{BufRead, BufReader, Read, Write}; + use std::net::TcpListener; + use std::sync::Arc; + + const ID: &str = "11111111-2222-4333-8444-555555555555"; + + fn production() -> Build { + Build { + origin: "https://otel.berd.xyz", + environment: "production", + channel: "internal", + version: "0.5.42", + } + } + + fn pocket(rate: f32) -> TtsSettings { + TtsSettings::Pocket { + model: "pocket".into(), + voice: "anna".into(), + rate, + } + } + + fn context(values: &[&str], tts: &TtsSettings) -> Option { + Context::new(&arguments(values), tts, || vec!["Samantha".into()]) + } + + fn arguments(values: &[&str]) -> Vec { + values.iter().map(|value| value.to_string()).collect() + } + + fn at(seconds: u64) -> SystemTime { + UNIX_EPOCH + Duration::from_secs(seconds) + } + + #[test] + fn context_reports_backends_and_a_valid_rate() { + let parakeet = context( + &[ + "session", + "--stt-backend", + "parakeet", + "--tts-backend", + "pocket", + ], + &pocket(1.25), + ) + .unwrap(); + assert_eq!( + parakeet.attributes(), + [ + ("input_backend", "parakeet".to_string()), + ("output_backend", "pocket".to_string()), + ("voice_mode", "chained".to_string()), + ("tts_rate", "1.25".to_string()), + ("tts_voice", "anna".to_string()), + ] + ); + let siri = |voice: &str| TtsSettings::Siri { + voice: voice.into(), + language: "en-US".into(), + rate: 1.0, + }; + let default = context(&["session"], &siri("Samantha")).unwrap(); + assert_eq!((default.input, default.output), ("macos", "siri")); + assert_eq!(default.rate.as_deref(), Some("1")); + assert_eq!(default.voice.as_deref(), Some("Samantha")); + assert_eq!( + context(&["session"], &siri("Fred's clone")) + .unwrap() + .voice + .as_deref(), + Some("custom") + ); + assert_eq!(context(&["session"], &siri("")).unwrap().voice, None); + let openai = |voice: &str| TtsSettings::OpenAi { + model: "gpt-4o-mini-tts".into(), + voice: voice.into(), + rate: 1.0, + }; + assert_eq!( + context(&["session"], &openai("marin")) + .unwrap() + .voice + .as_deref(), + Some("marin") + ); + assert_eq!( + context(&["session"], &openai("my-voice")) + .unwrap() + .voice + .as_deref(), + Some("custom") + ); + assert_eq!(context(&["session"], &pocket(5.0)).unwrap().rate, None); + assert_eq!( + context(&["session", "--stt-backend", "whisper"], &siri("Samantha")), + None + ); + } + + #[test] + fn payloads_match_the_gateway_contract() { + let context = context(&["session", "--stt-backend", "parakeet"], &pocket(1.25)).unwrap(); + let started = payload( + production(), + ID, + record( + "berd_voice_conversation_started", + at(1_762_264_800), + &context.attributes(), + ), + ); + let ended = payload( + production(), + ID, + record( + "berd_voice_conversation_ended", + at(1_762_265_100), + &ended_attributes(&context, Duration::from_millis(300_250), 4, 3, "user"), + ), + ); + let fixture = |event: &str, nanos: &str, extra: Value| { + let mut attributes = json!([ + {"key": "input_backend", "value": {"stringValue": "parakeet"}}, + {"key": "output_backend", "value": {"stringValue": "pocket"}}, + {"key": "voice_mode", "value": {"stringValue": "chained"}}, + {"key": "tts_rate", "value": {"stringValue": "1.25"}}, + {"key": "tts_voice", "value": {"stringValue": "anna"}}, + ]); + attributes + .as_array_mut() + .unwrap() + .extend(extra.as_array().unwrap().clone()); + json!({"resourceLogs": [{ + "resource": { + "attributes": [ + {"key": "service.name", "value": {"stringValue": "berd-call"}}, + {"key": "service.version", "value": {"stringValue": "0.5.42"}}, + {"key": "deployment.environment", "value": {"stringValue": "production"}}, + {"key": "distribution.channel", "value": {"stringValue": "internal"}}, + {"key": "installation.id", "value": {"stringValue": ID}}, + ], + "droppedAttributesCount": 0, + }, + "scopeLogs": [{ + "scope": {"name": "berd-call.telemetry"}, + "logRecords": [{ + "timeUnixNano": nanos, + "observedTimeUnixNano": nanos, + "body": {}, + "eventName": event, + "attributes": attributes, + "droppedAttributesCount": 0, + }], + }], + }]}) + }; + assert_eq!( + started, + fixture( + "berd_voice_conversation_started", + "1762264800000000000", + json!([]) + ) + ); + assert_eq!( + ended, + fixture( + "berd_voice_conversation_ended", + "1762265100000000000", + json!([ + {"key": "duration_ms", "value": {"stringValue": "300250"}}, + {"key": "user_utterance_count", "value": {"stringValue": "4"}}, + {"key": "assistant_response_count", "value": {"stringValue": "3"}}, + {"key": "end_reason", "value": {"stringValue": "user"}}, + ]) + ) + ); + if let Some(path) = std::env::var_os("BERD_CALL_TELEMETRY_FIXTURES") { + let fixtures = json!({"started": started, "ended": ended}); + std::fs::write(path, serde_json::to_string_pretty(&fixtures).unwrap()).unwrap(); + } + } + + #[cfg(not(feature = "telemetry-enforced"))] + #[test] + fn the_command_shares_berds_preference() { + let dir = tempfile::tempdir().unwrap(); + let status = |args: &[&str]| command_in(dir.path(), args).unwrap(); + assert_eq!(status(&[]), json!({"enabled": false, "enforced": false})); + assert_eq!( + status(&["enable"]), + json!({"enabled": true, "enforced": false}) + ); + assert_eq!(store::load_telemetry_consent(dir.path()), Ok(true)); + assert_eq!( + status(&["disable"]), + json!({"enabled": false, "enforced": false}) + ); + assert_eq!( + status(&["status"]), + json!({"enabled": false, "enforced": false}) + ); + assert!(command_in(dir.path(), &["maybe"]).is_err()); + assert!(command_in(dir.path(), &["enable", "now"]).is_err()); + } + + #[cfg(feature = "telemetry-enforced")] + #[test] + fn enforced_builds_are_always_enabled() { + let dir = tempfile::tempdir().unwrap(); + assert_eq!( + command_in(dir.path(), &[]).unwrap(), + json!({"enabled": true, "enforced": true}) + ); + assert!(command_in(dir.path(), &["disable"]).is_err()); + } + + struct Request { + line: String, + headers: Vec<(String, String)>, + body: Value, + } + + /// Serves `/v1/bootstrap` then `/v1/logs`, running `during_bootstrap` + /// before answering the bootstrap. + fn gateway(during_bootstrap: impl Fn() + Send + 'static) -> (String, Arc>>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let origin = format!("http://{}", listener.local_addr().unwrap()); + let requests = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&requests); + thread::spawn(move || { + for stream in listener.incoming() { + let mut stream = stream.unwrap(); + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + let mut headers = Vec::new(); + loop { + let mut header = String::new(); + reader.read_line(&mut header).unwrap(); + let Some((name, value)) = header.trim_end().split_once(": ") else { + break; + }; + headers.push((name.to_ascii_lowercase(), value.to_string())); + } + let length = headers + .iter() + .find(|(name, _)| name == "content-length") + .map_or(0, |(_, value)| value.parse().unwrap()); + let mut body = vec![0; length]; + reader.read_exact(&mut body).unwrap(); + let bootstrap = line.starts_with("POST /v1/bootstrap "); + if bootstrap { + during_bootstrap(); + } + recorded.lock().unwrap().push(Request { + line: line.trim_end().to_string(), + headers, + body: serde_json::from_slice(&body).unwrap(), + }); + let response = if bootstrap { + r#"{"token":"token-1","tokenType":"Bearer","expiresInSeconds":900}"# + } else { + r#"{"accepted":true}"# + }; + write!( + stream, + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response}", + response.len() + ) + .unwrap(); + } + }); + (origin, requests) + } + + fn export_to(origin: &str, dir: &Path) -> Result<(), String> { + let build = Build { + origin: Box::leak(origin.to_string().into_boxed_str()), + ..production() + }; + let context = context(&["session"], &pocket(1.0)).unwrap(); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(export( + &client().unwrap(), + build, + dir, + record( + "berd_voice_conversation_started", + at(1), + &context.attributes(), + ), + )) + } + + #[cfg(not(feature = "telemetry-enforced"))] + #[test] + fn nothing_is_sent_or_created_without_consent() { + let dir = tempfile::tempdir().unwrap(); + let (origin, requests) = gateway(|| {}); + assert!(export_to(&origin, dir.path()).is_err()); + assert!(requests.lock().unwrap().is_empty()); + assert!(!dir.path().join(store::INSTALLATION_ID_FILE_NAME).exists()); + } + + #[test] + fn a_consented_export_bootstraps_then_uploads_the_record() { + let dir = tempfile::tempdir().unwrap(); + store::persist_telemetry_consent(dir.path(), true).unwrap(); + let (origin, requests) = gateway(|| {}); + export_to(&origin, dir.path()).unwrap(); + let id = store::load_or_create_installation_id(dir.path()).unwrap(); + let requests = requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert_eq!(requests[0].line, "POST /v1/bootstrap HTTP/1.1"); + assert_eq!(requests[0].body, json!({"installationId": id})); + assert_eq!(requests[1].line, "POST /v1/logs HTTP/1.1"); + let header = |name: &str| { + requests[1] + .headers + .iter() + .filter(|(header, _)| header == name) + .map(|(_, value)| value.as_str()) + .collect::>() + }; + assert_eq!(header("authorization"), ["Bearer token-1"]); + assert_eq!(header("x-berd-schema-version"), ["berd-call-otlp-logs-v1"]); + assert_eq!(header("content-type"), ["application/json"]); + let resource = &requests[1].body["resourceLogs"][0]["resource"]["attributes"]; + assert_eq!( + resource[4], + json!({"key": "installation.id", "value": {"stringValue": id}}) + ); + } + + #[cfg(not(feature = "telemetry-enforced"))] + #[test] + fn revoking_consent_during_bootstrap_stops_the_upload() { + let dir = tempfile::tempdir().unwrap(); + store::persist_telemetry_consent(dir.path(), true).unwrap(); + let revoke = dir.path().to_path_buf(); + let (origin, requests) = gateway(move || { + store::persist_telemetry_consent(&revoke, false).unwrap(); + }); + assert!(export_to(&origin, dir.path()).is_err()); + let requests = requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].line, "POST /v1/bootstrap HTTP/1.1"); + } + + #[test] + fn ended_attributes_append_duration_counts_and_reason() { + let context = context(&["session"], &pocket(9.0)).unwrap(); + let attributes = ended_attributes( + &context, + Duration::from_millis(1_500), + 0, + 2, + "clean-shutdown", + ); + assert_eq!( + &attributes[4..], + [ + ("duration_ms", "1500".to_string()), + ("user_utterance_count", "0".to_string()), + ("assistant_response_count", "2".to_string()), + ("end_reason", "clean-shutdown".to_string()), + ] + ); + } +} diff --git a/src-tauri/crates/berd-call/src/telemetry_store.rs b/src-tauri/crates/berd-call/src/telemetry_store.rs new file mode 100644 index 000000000..152fcfe41 --- /dev/null +++ b/src-tauri/crates/berd-call/src/telemetry_store.rs @@ -0,0 +1,126 @@ +//! Berd's persisted telemetry consent and anonymous installation id, shared by +//! Berd.app and the standalone berd-call CLI so both honor one preference. The +//! caller supplies the directory: the app passes its Tauri app-data dir, and +//! the CLI resolves the same platform path. Consent is opt-in and fail-closed. + +use std::{fs, io, path::Path}; + +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +pub const TELEMETRY_SETTINGS_FILE_NAME: &str = "telemetry-settings.json"; +pub const TELEMETRY_SETTINGS_SCHEMA_VERSION: u32 = 1; +pub const INSTALLATION_ID_FILE_NAME: &str = "telemetry-installation-id"; + +/// On-disk shape of the consent file. The schema version is pinned exactly: a +/// file from a future schema fails closed rather than guessing at semantics. +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct PersistedTelemetrySettings { + pub schema_version: u32, + pub enabled: bool, +} + +/// The persisted consent. A missing file is `Ok(false)`, the normal first-run +/// state; an unreadable, invalid or future-schema file is an error that callers +/// treat as disabled. +pub fn load_telemetry_consent(dir: &Path) -> Result { + let path = dir.join(TELEMETRY_SETTINGS_FILE_NAME); + let raw = match fs::read_to_string(&path) { + Ok(raw) => raw, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(format!("unreadable telemetry settings: {error}")), + }; + match serde_json::from_str::(&raw) { + Ok(settings) if settings.schema_version == TELEMETRY_SETTINGS_SCHEMA_VERSION => { + Ok(settings.enabled) + } + Ok(settings) => Err(format!( + "telemetry settings have unknown schema version {}", + settings.schema_version + )), + Err(error) => Err(format!("unreadable telemetry settings: {error}")), + } +} + +/// Persists consent with a write-then-rename, so a crash never leaves a torn +/// file (which would read as disabled anyway). +pub fn persist_telemetry_consent(dir: &Path, enabled: bool) -> Result<(), String> { + let body = serde_json::to_string_pretty(&PersistedTelemetrySettings { + schema_version: TELEMETRY_SETTINGS_SCHEMA_VERSION, + enabled, + }) + .map_err(|error| format!("Failed to serialize telemetry settings: {error}"))?; + write_atomically(dir, TELEMETRY_SETTINGS_FILE_NAME, &body) +} + +/// Loads the persisted installation id, generating and persisting a fresh UUID +/// when the file is missing or holds a value the gateway would reject. +pub fn load_or_create_installation_id(dir: &Path) -> Result { + if let Ok(existing) = fs::read_to_string(dir.join(INSTALLATION_ID_FILE_NAME)) { + let existing = existing.trim(); + if is_valid_installation_id(existing) { + return Ok(existing.to_string()); + } + } + let id = Uuid::new_v4().to_string(); + write_atomically(dir, INSTALLATION_ID_FILE_NAME, &id)?; + Ok(id) +} + +/// Mirrors the gateway's accepted installation-id shape +/// (`^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$`). +pub fn is_valid_installation_id(value: &str) -> bool { + let mut chars = value.chars(); + let Some(first) = chars.next() else { + return false; + }; + (8..=128).contains(&value.len()) + && first.is_ascii_alphanumeric() + && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | ':' | '-')) +} + +fn write_atomically(dir: &Path, name: &str, body: &str) -> Result<(), String> { + fs::create_dir_all(dir) + .map_err(|error| format!("Failed to create app data dir {}: {error}", dir.display()))?; + let path = dir.join(name); + let tmp = path.with_extension("tmp"); + fs::write(&tmp, body).map_err(|error| format!("Failed to write {}: {error}", tmp.display()))?; + fs::rename(&tmp, &path) + .map_err(|error| format!("Failed to persist {}: {error}", path.display())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn consent_round_trips_and_fails_closed() { + let dir = tempfile::tempdir().unwrap(); + assert_eq!(load_telemetry_consent(dir.path()), Ok(false)); + persist_telemetry_consent(dir.path(), true).unwrap(); + assert_eq!(load_telemetry_consent(dir.path()), Ok(true)); + assert_eq!( + fs::read_to_string(dir.path().join(TELEMETRY_SETTINGS_FILE_NAME)).unwrap(), + "{\n \"schemaVersion\": 1,\n \"enabled\": true\n}" + ); + for raw in [ + "not json", + r#"{"schemaVersion":2,"enabled":true}"#, + r#"{"enabled":true}"#, + ] { + fs::write(dir.path().join(TELEMETRY_SETTINGS_FILE_NAME), raw).unwrap(); + assert!(load_telemetry_consent(dir.path()).is_err(), "{raw}"); + } + } + + #[test] + fn installation_id_is_created_once_and_replaced_when_invalid() { + let dir = tempfile::tempdir().unwrap(); + let id = load_or_create_installation_id(dir.path()).unwrap(); + assert!(is_valid_installation_id(&id)); + assert_eq!(load_or_create_installation_id(dir.path()).unwrap(), id); + fs::write(dir.path().join(INSTALLATION_ID_FILE_NAME), "bad id").unwrap(); + assert_ne!(load_or_create_installation_id(dir.path()).unwrap(), id); + } +} diff --git a/src-tauri/src/commands/telemetry.rs b/src-tauri/src/commands/telemetry.rs index ed8c8fd44..e2020f58f 100644 --- a/src-tauri/src/commands/telemetry.rs +++ b/src-tauri/src/commands/telemetry.rs @@ -1,11 +1,13 @@ use crate::services::distro_bundle::{DistroBundleState, TelemetryChannel}; +use berd_call::telemetry_store::{ + self as store, load_or_create_installation_id, persist_telemetry_consent, +}; use reqwest::{ header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE}, redirect::Policy, }; use serde::{Deserialize, Serialize}; use std::{ - fs, path::{Path, PathBuf}, sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, @@ -60,20 +62,13 @@ const ALLOWED_OTEL_LOGS_HOSTS: [&str; 3] = [ "otlp.invalid.goose-internal.example", ]; -// File under the app-data dir holding the persistent anonymous installation -// id. It keys bootstrap-token issuance (and rate limiting) on the gateway and -// is stamped by the renderer as the `installation.id` resource attribute. -const INSTALLATION_ID_FILE_NAME: &str = "telemetry-installation-id"; - -// File under the app-data dir holding the user's telemetry consent. It lives -// here — not in the renderer's localStorage — precisely so this module can +// The installation id and consent files live in the app-data dir, through +// `berd_call::telemetry_store`, which the standalone berd-call CLI shares. The +// consent lives there, not in the renderer's localStorage, so this module can // enforce it natively: "disabled" must mean no bytes to the gateway at all, -// including `/v1/bootstrap` (which carries the installation id), and only the -// Rust host can guarantee that regardless of renderer timing. A missing file +// including `/v1/bootstrap` (which carries the installation id). A missing file // is DISABLED — telemetry is opt-in, and every failure mode reads as "no // consent". -const TELEMETRY_SETTINGS_FILE_NAME: &str = "telemetry-settings.json"; -const TELEMETRY_SETTINGS_SCHEMA_VERSION: u32 = 1; // Refresh the cached upload token this long before its reported expiry, so an // export near the boundary does not race server-side expiry. @@ -128,9 +123,14 @@ impl TelemetryAuthState { /// The effective telemetry consent: forced ON in enforced builds, /// otherwise the persisted user setting. Fail-closed — a missing, - /// unreadable, or unrecognized settings file means disabled. + /// unreadable, or unrecognized settings file means disabled. The file is + /// re-read on every check because `berd-call telemetry disable` writes it + /// from outside this process, and that revocation must stop this app's + /// exports too, including ones already past the entry gate. fn consent_granted(&self) -> bool { - telemetry_enforced_by_build() || self.consent_enabled.load(Ordering::Relaxed) + telemetry_enforced_by_build() + || (self.consent_enabled.load(Ordering::Relaxed) + && store::load_telemetry_consent(&self.app_data_dir).unwrap_or(false)) } /// Snapshot of the revocation counter, taken as an export begins and @@ -259,16 +259,6 @@ fn telemetry_enforced_by_build() -> bool { cfg!(feature = "block-telemetry-enforced") } -/// On-disk shape of the consent file. The schema version is pinned exactly: -/// a file from a future schema fails closed (reads as disabled) rather than -/// guessing at consent semantics this build does not know. -#[derive(Debug, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -struct PersistedTelemetrySettings { - schema_version: u32, - enabled: bool, -} - /// Wire shape of `get_telemetry_settings` / `set_telemetry_enabled`: the /// *effective* consent, i.e. what the export gate will actually apply. #[derive(Debug, Clone, Copy, Serialize)] @@ -506,95 +496,14 @@ async fn bootstrap_upload_token( }) } -/// Loads the persisted installation id, generating and persisting a fresh UUID -/// when the file is missing or holds a value the gateway would reject. -fn load_or_create_installation_id(app_data_dir: &Path) -> Result { - let path = app_data_dir.join(INSTALLATION_ID_FILE_NAME); - if let Ok(existing) = fs::read_to_string(&path) { - let existing = existing.trim(); - if is_valid_installation_id(existing) { - return Ok(existing.to_string()); - } - } - - let id = Uuid::new_v4().to_string(); - fs::create_dir_all(app_data_dir).map_err(|error| { - format!( - "Failed to create app data dir {}: {error}", - app_data_dir.display() - ) - })?; - // Write-then-rename so a crash can never leave a torn id on disk. - let tmp = path.with_extension("tmp"); - fs::write(&tmp, &id).map_err(|error| format!("Failed to write {}: {error}", tmp.display()))?; - fs::rename(&tmp, &path) - .map_err(|error| format!("Failed to persist {}: {error}", path.display()))?; - Ok(id) -} - /// Loads the persisted telemetry consent. Fail-closed: a missing file (the /// normal first-run state — telemetry is opt-in), an unreadable file, invalid /// JSON, or an unknown schema version all read as disabled. fn load_telemetry_consent(app_data_dir: &Path) -> bool { - let path = app_data_dir.join(TELEMETRY_SETTINGS_FILE_NAME); - let Ok(raw) = fs::read_to_string(&path) else { - return false; - }; - match serde_json::from_str::(&raw) { - Ok(settings) if settings.schema_version == TELEMETRY_SETTINGS_SCHEMA_VERSION => { - settings.enabled - } - Ok(settings) => { - log::warn!( - "Ignoring telemetry settings with unknown schema version {}", - settings.schema_version - ); - false - } - Err(error) => { - log::warn!("Ignoring unreadable telemetry settings: {error}"); - false - } - } -} - -/// Persists the telemetry consent with a write-then-rename, mirroring the -/// installation-id write, so a crash can never leave a torn settings file — -/// and a torn file would read as disabled anyway. The rename replaces an -/// existing destination on Windows too — `std::fs::rename`'s documented -/// contract, via `MoveFileExW`/`FileRenameInfoEx` with replace-if-exists, -/// unlike C's `rename` — so re-toggling consent needs no remove-first step. -fn persist_telemetry_consent(app_data_dir: &Path, enabled: bool) -> Result<(), String> { - let path = app_data_dir.join(TELEMETRY_SETTINGS_FILE_NAME); - fs::create_dir_all(app_data_dir).map_err(|error| { - format!( - "Failed to create app data dir {}: {error}", - app_data_dir.display() - ) - })?; - let body = serde_json::to_string_pretty(&PersistedTelemetrySettings { - schema_version: TELEMETRY_SETTINGS_SCHEMA_VERSION, - enabled, + store::load_telemetry_consent(app_data_dir).unwrap_or_else(|error| { + log::warn!("Ignoring telemetry settings: {error}"); + false }) - .map_err(|error| format!("Failed to serialize telemetry settings: {error}"))?; - let tmp = path.with_extension("tmp"); - fs::write(&tmp, body).map_err(|error| format!("Failed to write {}: {error}", tmp.display()))?; - fs::rename(&tmp, &path) - .map_err(|error| format!("Failed to persist {}: {error}", path.display()))?; - Ok(()) -} - -/// Mirrors the gateway's accepted installation-id shape -/// (`^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$`), so a corrupted file regenerates -/// instead of bootstrapping with a value the gateway will reject. -fn is_valid_installation_id(value: &str) -> bool { - let mut chars = value.chars(); - let Some(first) = chars.next() else { - return false; - }; - (8..=128).contains(&value.len()) - && first.is_ascii_alphanumeric() - && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | ':' | '-')) } /// Validates that an OTLP logs endpoint is HTTPS, targets an allowlisted @@ -671,6 +580,8 @@ fn client() -> &'static reqwest::Client { #[cfg(test)] mod tests { use super::*; + use berd_call::telemetry_store::{is_valid_installation_id, INSTALLATION_ID_FILE_NAME}; + use std::fs; use std::sync::{ atomic::{AtomicUsize, Ordering}, Arc, Mutex as StdMutex, @@ -1304,6 +1215,10 @@ mod tests { #[cfg(not(feature = "block-telemetry-enforced"))] mod consent { use super::*; + use berd_call::telemetry_store::{ + PersistedTelemetrySettings, TELEMETRY_SETTINGS_FILE_NAME, + TELEMETRY_SETTINGS_SCHEMA_VERSION, + }; use tokio::sync::Semaphore; #[test] @@ -1480,6 +1395,67 @@ mod tests { assert!(gateway.requests_to("/v1/logs").is_empty()); } + /// `berd-call telemetry disable` revokes consent by rewriting the shared + /// file from another process, so a running app must honor that at its + /// next boundary as well as at the entry gate. + #[tokio::test] + async fn an_external_revocation_stops_a_running_app_from_uploading() { + let release = Arc::new(Semaphore::new(0)); + let gate = release.clone(); + let gateway = TestGateway::spawn_async(move |request| { + let gate = gate.clone(); + async move { + if request.path == "/v1/bootstrap" { + let _permit = gate.acquire().await; + bootstrap_response("token-1") + } else { + ScriptedResponse { + status: 200, + body: "{}".to_string(), + } + } + } + }) + .await; + let (dir, auth) = consented_auth_state(); + let auth = Arc::new(auth); + let logs_url = gateway.url("/v1/logs"); + let bootstrap_url = gateway.url("/v1/bootstrap"); + let epoch = auth.revocation_epoch(); + + let export = tokio::spawn({ + let auth = auth.clone(); + async move { + export_with_bootstrap_auth( + client(), + &auth, + epoch, + &logs_url, + &bootstrap_url, + EXPORT_BODY.to_string(), + ) + .await + } + }); + + wait_for_request(&gateway, "/v1/bootstrap").await; + persist_telemetry_consent(dir.path(), false).unwrap(); + release.add_permits(1); + + let error = export.await.unwrap().unwrap_err(); + assert!(error.contains("revoked"), "unexpected error: {error}"); + assert!(gateway.requests_to("/v1/logs").is_empty()); + assert!(!auth.settings().enabled); + let error = export_otel_logs_for_state( + &auth, + "https://otlp.invalid.goose-internal.example/v1/logs", + EXPORT_BODY.to_string(), + ) + .await + .unwrap_err(); + assert!(error.contains("disabled"), "unexpected error: {error}"); + } + /// The 401 path is the widest post-gate window: it would otherwise /// re-bootstrap (a second request carrying the installation id) and /// re-send the batch after the revocation.