diff --git a/desktop/TELEMETRY.md b/desktop/TELEMETRY.md new file mode 100644 index 000000000..039e10596 --- /dev/null +++ b/desktop/TELEMETRY.md @@ -0,0 +1,51 @@ +# Desktop telemetry + +Telemetry is enabled by default. Welcome discloses it without an opt-in gate. Launch OpenBot with either `COPILOTKIT_TELEMETRY_DISABLED=true` or `DO_NOT_TRACK=1` to disable both the desktop emitter and the runtime. Both variables accept `true` or `1`. Quit and relaunch after changing the environment. Opting out removes pending desktop events and prevents recording or replay. + +The native emitter starts before credentials are available. It persists a random installation UUID and a bounded queue in the app's local data directory. It does not derive identity from the machine. The runtime receives the same UUID through `CPK_TELEMETRY_ID` and uses sampling rate 1; opt-out still takes precedence. + +## Event allowlist + +All desktop event names begin with `oss.desktop.`. The Rust `EventData` enum and `schema_json()` define the closed property schema; deserialization rejects unknown properties and enum values. + +| Event suffix | Properties | +| --- | --- | +| `step_viewed` | Setup step enum | +| `harness_chosen` | Harness enum, including `byo_url` | +| `model_chosen` | Provider and credential-path enums; custom-base-URL boolean | +| `engine_detected` | Engine enum; responding boolean | +| `engine_installed` | Engine and installer-outcome enums | +| `windows_stage` | Windows prerequisite outcome enum | +| `image_pull` | Outcome enum, milliseconds, optional observed download bytes | +| `setup_failed` | Setup step and error-class enums | +| `activated` | No properties; first successful setup Bot answer | +| `setup_abandoned` | Last setup step enum | + +Every event carries desktop distribution, numeric app version, platform, architecture, numeric OS version when available, and engine. The runtime receives the same desktop metadata through its existing `telemetryProperties` hook. No prompt, answer, credential, file name, path, hostname, email, model name, YAML value, or custom URL is accepted by the desktop schema. + +Image timing covers an explicit missing-only Compose pull, excluding startup and migrations. Byte counts are the download counters observed from Compose, deduplicated per layer; they are not an exact network total. Missing counters remain null. Older Compose providers without `pull --policy missing` retain their existing implicit pull behavior and emit no pull measurement. + +## Delivery and verification + +Events persist before sending and keep their event IDs across retries. The queue holds at most 256 events, dropping the oldest on overflow. Background delivery, bounded quit flushing, and replay on relaunch use the existing CopilotKit ingest contract. A crash before activation records abandonment on the next launch. Telemetry failures do not block setup. + +The ingest must accept the new `oss.desktop.*` namespace: [oss-path-to-production #290](https://github.com/CopilotKit/oss-path-to-production/pull/290). A successful HTTP response alone does not prove downstream acceptance; production ingest acknowledges filtered events too. Merge and deployment of that change are required for production desktop delivery. + +Regression tests live in the native telemetry and pull-metrics modules, frontend telemetry tests, and server metadata tests. For an actual local HTTP and installed-runtime check, build the native probe and run the driver from the repository root: + +```sh +cargo build --manifest-path desktop/src-tauri/Cargo.toml --release --example telemetry_probe +bun desktop/scripts/validate-telemetry.ts desktop/src-tauri/target/release/examples/telemetry_probe +``` + +The driver uses temporary data and a loopback receiver. It exercises separate-process persistence/replay, quit, activation deduplication, opt-out, and the installed runtime's identity/metadata handoff. It does not require AI credentials or send validation events to production. Windows and Linux execution remain covered by the platform CI runs; a Mac run is not proof of their native UI behavior. + +Local HTTP validation is reusable through `desktop/scripts/validate-telemetry.ts` with the separately built `telemetry_probe` example. It checks offline queue replay with stable installation/event IDs, recovered abandonment, quit flushing, activation once across process restarts, native/runtime metadata, and both `true`/`1` values of `COPILOTKIT_TELEMETRY_DISABLED` and `DO_NOT_TRACK`. Opt-out must remove queued state and identity and produce zero HTTP sends. + +The September 12 local run used installed runtime 1.70.1 and its public ESM Hono factory, following the production `eventsource` preload. Its existing Bun dependency layout needed an explicit `NODE_PATH` pointing at the existing hoist directory so a cached `gaxios` module could resolve `extend`. No dependency was installed or modified. Reproduce that adapter from the repository root: + +```sh +NODE_PATH="$PWD/node_modules/.bun/node_modules" bun --no-env-file --no-install desktop/scripts/validate-telemetry.ts /absolute/path/to/telemetry_probe +``` + +The final loopback report recorded five native requests and one runtime request with matching installation identity, with all four opt-out cases passing. This covers the emitter and installed-runtime HTTP contract; fresh dependency installation and final PostHog delivery still require their own validation. diff --git a/desktop/scripts/validate-telemetry.ts b/desktop/scripts/validate-telemetry.ts new file mode 100644 index 000000000..0346d4425 --- /dev/null +++ b/desktop/scripts/validate-telemetry.ts @@ -0,0 +1,416 @@ +/** + * Real HTTP validation of the production native emitter and the installed runtime. + * Build the native example separately, then run: + * bun --no-env-file --no-install desktop/scripts/validate-telemetry.ts /absolute/path/to/telemetry_probe + * No credentials, containers, model calls, or production telemetry endpoints are used. + */ +import assert from "node:assert/strict"; +import { existsSync } from "node:fs"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { createRequire } from "node:module"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { desktopTelemetryProperties } from "../../server/src/desktop-telemetry"; + +const root = resolve(import.meta.dir, "../.."); +const fromServer = createRequire(join(root, "server/package.json")); +const uuid = + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; +type Env = Record; +type Captured = { identity: string | null; body: Record }; +type Queued = { event_id: string; event_name: string; occurred_at_ms: number }; +type State = { + install_id: string; + queue: Queued[]; + activated: boolean; + last_step: string | null; +}; + +function object(value: unknown): Record { + assert.ok(value && typeof value === "object" && !Array.isArray(value)); + return value as Record; +} + +function lastObject(stdout: string): Record { + const line = stdout.trim().split("\n").at(-1); + assert.ok(line, "child did not return a JSON report"); + return object(JSON.parse(line)); +} + +function loopback(value: string): URL { + const url = new URL(value); + assert.equal(url.protocol, "http:"); + assert.equal(url.hostname, "127.0.0.1"); + return url; +} + +/** Fresh process: the SDK snapshots opt-out and sampling before any runtime import. */ +async function runtimeChild(endpoint: string) { + loopback(endpoint); + assert.equal(process.env.COPILOTKIT_TELEMETRY_URL, endpoint); + const fetchHttp = globalThis.fetch; + const sends: Promise[] = []; + const blocked: string[] = []; + const guardedFetch: typeof fetch = Object.assign( + ( + input: Parameters[0], + init?: Parameters[1], + ) => { + const url = new URL(input instanceof Request ? input.url : String(input)); + if (url.protocol !== "http:" || url.hostname !== "127.0.0.1") { + blocked.push(url.origin); + return Promise.reject(new Error("Validation forbids external fetches")); + } + const result = fetchHttp(input, init); + if (url.href === endpoint) sends.push(result); + return result; + }, + { preconnect: fetchHttp.preconnect }, + ); + globalThis.fetch = guardedFetch; + + const runtimePackage = fromServer("@copilotkit/runtime/package.json"); + assert.equal( + runtimePackage.version, + "1.70.1", + "recheck the wire contract before changing the installed version", + ); + // Match production-entry.ts's preload and the server's public ESM imports. Requiring the + // CJS runtime selects a different dependency graph and does not validate the shipped path. + await import(Bun.resolveSync("eventsource", join(root, "server"))); + const { CopilotRuntime } = await import( + Bun.resolveSync("@copilotkit/runtime/v2", join(root, "server")) + ); + const { createCopilotHonoHandler } = await import( + Bun.resolveSync("@copilotkit/runtime/v2/hono", join(root, "server")) + ); + const runtime = new CopilotRuntime({ + agents: {}, + telemetryProperties: desktopTelemetryProperties(), + }); + const handler = createCopilotHonoHandler({ + runtime, + basePath: "/api/copilotkit", + }); + const info = await handler.fetch( + new Request("http://127.0.0.1/api/copilotkit/info"), + ); + assert.equal(info.status, 200); + await info.text(); + // Handler creation queues instance_created in a promise continuation. Drain it before + // waiting on the real fetches; no transport is mocked and no polling timeout proves success. + await new Promise((done) => setImmediate(done)); + await Promise.all(sends); + assert.deepEqual(blocked, []); + console.log( + JSON.stringify({ + runtimeVersion: runtimePackage.version, + requests: sends.length, + }), + ); +} + +async function validate(nativePath: string) { + assert.ok( + existsSync(nativePath), + `Build the native telemetry_probe example first: ${nativePath}`, + ); + const scratch = await mkdtemp( + join(tmpdir(), "openbot-telemetry-validation-"), + ); + const received: Captured[] = []; + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + assert.equal(request.method, "POST"); + assert.equal(new URL(request.url).pathname, "/ingest"); + const raw = await request.text(); + assert.ok(Buffer.byteLength(raw) <= 16 * 1024); + received.push({ + identity: request.headers.get("X-CopilotKit-Telemetry-Id"), + body: object(JSON.parse(raw)), + }); + return Response.json({ ok: true }, { status: 202 }); + }, + }); + const endpoint = `http://127.0.0.1:${server.port}/ingest`; + // Reserve and close a second local socket: a real connection-refused flush, not a fake transport. + const offline = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch: () => new Response(), + }); + const offlineEndpoint = `http://127.0.0.1:${offline.port}/ingest`; + await offline.stop(true); + + const cleanEnv: Env = { + PATH: process.env.PATH ?? "", + NODE_ENV: "production", + NO_COLOR: "1", + }; + // NODE_PATH is an explicit caller-selected resolution aid for existing installations; + // this script never installs packages or silently rewires their dependency graph. + for (const key of ["SystemRoot", "WINDIR", "TEMP", "TMP", "NODE_PATH"]) { + const value = process.env[key]; + if (value) cleanEnv[key] = value; + } + async function processRun(command: string[], env: Env, cwd = scratch) { + const child = Bun.spawn(command, { + cwd, + env: { ...cleanEnv, ...env }, + stdout: "pipe", + stderr: "pipe", + }); + const timeout = setTimeout(() => child.kill(), 15_000); + try { + const [stdout, stderr, code] = await Promise.all([ + new Response(child.stdout).text(), + new Response(child.stderr).text(), + child.exited, + ]); + return { stdout, stderr, code }; + } finally { + clearTimeout(timeout); + } + } + function lastJson(stdout: string): Env { + const parsed = lastObject(stdout); + assert.ok( + Object.values(parsed).every((value) => typeof value === "string"), + ); + return parsed as Env; + } + async function native( + directory: string, + action: string, + env: Env = {}, + url = endpoint, + ) { + const result = await processRun([nativePath, directory, url, action], env); + assert.equal( + result.code, + 0, + `${action} failed: ${result.stderr.slice(0, 500)}`, + ); + return lastJson(result.stdout); + } + async function state(directory: string): Promise { + return JSON.parse( + await readFile(join(directory, "telemetry-state.json"), "utf8"), + ); + } + async function runtime(env: Env) { + const result = await processRun( + [ + process.execPath, + "--no-env-file", + "--no-install", + import.meta.path, + "--runtime-child", + endpoint, + ], + { ...env, COPILOTKIT_TELEMETRY_URL: endpoint }, + join(root, "server"), + ); + assert.equal( + result.code, + 0, + `runtime child failed: ${result.stderr.slice(0, 1000)}`, + ); + return lastObject(result.stdout); + } + const checks: string[] = []; + try { + const recoveredDir = join(scratch, "recovered"); + const nativeEnv = await native(recoveredDir, "record", {}, offlineEndpoint); + const original = await state(recoveredDir); + assert.match(original.install_id, uuid); + assert.equal(original.queue.length, 1); + assert.equal(original.queue[0].event_name, "oss.desktop.step_viewed"); + assert.equal(nativeEnv.CPK_TELEMETRY_ID, original.install_id); + assert.equal(nativeEnv.COPILOTKIT_TELEMETRY_SAMPLE_RATE, "1"); + const failed = await processRun( + [nativePath, recoveredDir, offlineEndpoint, "flush"], + {}, + ); + assert.notEqual(failed.code, 0, "offline connection must fail"); + const queued = await state(recoveredDir); + assert.equal(queued.install_id, original.install_id); + assert.equal(queued.queue.length, 2); + assert.equal(queued.queue[0].event_id, original.queue[0].event_id); + assert.equal(queued.queue[1].event_name, "oss.desktop.setup_abandoned"); + assert.equal(received.length, 0); + const relaunchedEnv = await native(recoveredDir, "flush"); + assert.deepEqual(relaunchedEnv, nativeEnv); + assert.deepEqual( + received.map((request) => request.body.event_id), + queued.queue.map((event) => event.event_id), + ); + assert.ok( + received.every((request) => request.identity === original.install_id), + ); + assert.deepEqual( + received.map((request) => request.body.ts), + queued.queue.map((event) => Math.floor(event.occurred_at_ms / 1000)), + ); + assert.equal((await state(recoveredDir)).queue.length, 0); + await native(recoveredDir, "flush"); + assert.equal( + received.length, + 2, + "successful flush must not replay acknowledged events", + ); + checks.push( + "offline queue replay: stable installation and event IDs; one recovered abandonment", + ); + + const quitStart = received.length; + await native(join(scratch, "quit"), "quit"); + assert.deepEqual( + received.slice(quitStart).map((request) => request.body.event), + ["oss.desktop.step_viewed", "oss.desktop.setup_abandoned"], + ); + checks.push("quit flushes current step and abandonment"); + + const activationStart = received.length; + const activatedDir = join(scratch, "activated"); + await native(activatedDir, "activate"); + await native(activatedDir, "activate"); + assert.deepEqual( + received.slice(activationStart).map((request) => request.body.event), + ["oss.desktop.activated"], + ); + assert.equal((await state(activatedDir)).activated, true); + checks.push( + "activation emitted once across duplicate calls and process relaunch", + ); + + const nativeRequests = received.length; + for (const request of received) { + assert.deepEqual(Object.keys(request.body).sort(), [ + "event", + "event_id", + "global_properties", + "package", + "properties", + "ts", + ]); + assert.match(String(request.body.event_id), uuid); + assert.deepEqual(request.body.package, { + name: "openbot-desktop", + version: "0.0.9", + }); + assert.deepEqual(object(request.body.global_properties), { + ...desktopTelemetryProperties(nativeEnv), + runtime_env: "test", + sampleRate: 1, + sampleWeight: 1, + sampleRateAdjustmentFactor: 0, + }); + } + checks.push( + "native wire envelope uses closed metadata, seconds and full sampling", + ); + + const runtimeResult = await runtime({ + ...nativeEnv, + OPENBOT_UNKNOWN: "synthetic-private-value", + }); + assert.equal(runtimeResult.requests, 1); + const runtimeRequest = received.at(-1); + assert.ok(runtimeRequest); + assert.equal(received.length, nativeRequests + 1); + assert.equal(runtimeRequest.identity, original.install_id); + assert.equal(runtimeRequest.body.event, "oss.runtime.instance_created"); + assert.deepEqual(runtimeRequest.body.package, { + name: "@copilotkit/runtime", + version: "1.70.1", + }); + assert.deepEqual(runtimeRequest.body.properties, { + actionsAmount: 0, + endpointTypes: [], + endpointsAmount: 0, + agentsAmount: 0, + "cloud.api_key_provided": false, + }); + assert.deepEqual(runtimeRequest.body.global_properties, { + ...desktopTelemetryProperties(nativeEnv), + sampleRate: 1, + sampleWeight: 1, + sampleRateAdjustmentFactor: 0, + telemetry_identified: false, + telemetry_emitter: "v2-runtime", + telemetry_transport: "lambda", + }); + assert.ok( + Number(runtimeRequest.body.ts) > 1_000_000_000 && + Number(runtimeRequest.body.ts) < 10_000_000_000, + ); + checks.push( + "installed runtime1.70.1 Hono boot event: same native UUID, closed metadata, full sampling", + ); + + for (const variable of ["COPILOTKIT_TELEMETRY_DISABLED", "DO_NOT_TRACK"]) { + for (const value of ["true", "1"]) { + const before: number = received.length; + const directory = join(scratch, `${variable}-${value}`); + await native(directory, "record", {}, offlineEndpoint); + assert.ok((await state(directory)).queue.length > 0); + const optedOutEnv = await native(directory, "quit", { + [variable]: value, + }); + assert.deepEqual(optedOutEnv, { COPILOTKIT_TELEMETRY_DISABLED: "1" }); + assert.equal( + existsSync(join(directory, "telemetry-state.json")), + false, + ); + const fresh = join(scratch, `${variable}-${value}-fresh`); + assert.deepEqual( + await native(fresh, "env", { [variable]: value }), + optedOutEnv, + ); + assert.equal(existsSync(join(fresh, "telemetry-state.json")), false); + // Test the SDK's own override too, even if a caller retained an older enabled ID. + assert.equal( + (await runtime({ ...nativeEnv, [variable]: value })).requests, + 0, + ); + assert.equal(received.length, before); + checks.push( + `${variable}=${value}: queue purged, no identity or HTTP sends, runtime also disabled`, + ); + } + } + assert.equal( + (await runtime({ COPILOTKIT_TELEMETRY_DISABLED: "1" })).requests, + 0, + ); + console.log( + JSON.stringify( + { + ok: true, + checks, + nativeRequests, + runtimeRequests: 1, + runtimeVersion: "1.70.1", + externalRequests: 0, + }, + null, + 2, + ), + ); + } finally { + await server.stop(true); + await rm(scratch, { recursive: true, force: true }); + } +} + +if (process.argv[2] === "--runtime-child") { + await runtimeChild(process.argv[3]); +} else { + const nativePath = + process.argv[2] ?? + join(root, "desktop/src-tauri/target/release/examples/telemetry_probe"); + await validate(resolve(nativePath)); +} diff --git a/desktop/src-tauri/examples/telemetry_probe.rs b/desktop/src-tauri/examples/telemetry_probe.rs new file mode 100644 index 000000000..2de34de83 --- /dev/null +++ b/desktop/src-tauri/examples/telemetry_probe.rs @@ -0,0 +1,65 @@ +//! Exercise the production emitter against an explicit local receiver, without starting OpenBot. +use openbot_desktop_lib::telemetry::{ + Architecture, Config, Context, Distribution, Engine, EventData, NumericVersion, Platform, + RuntimeEnv, Step, Telemetry, +}; + +fn main() -> Result<(), Box> { + let args: Vec = std::env::args().collect(); + if args.len() != 4 { + return Err("usage: telemetry_probe DATA_DIR http://127.0.0.1:PORT/ingest record|flush|quit|activate|env".into()); + } + let url = reqwest::Url::parse(&args[2])?; + if url.scheme() != "http" || url.host_str() != Some("127.0.0.1") { + return Err("this validation probe only sends to a local HTTP receiver".into()); + } + let telemetry = Telemetry::open( + &args[1], + Context { + distribution: Distribution::Desktop, + app_version: NumericVersion::parse("0.0.9")?, + platform: if cfg!(target_os = "windows") { + Platform::Windows + } else if cfg!(target_os = "macos") { + Platform::Macos + } else { + Platform::Linux + }, + arch: if cfg!(target_arch = "aarch64") { + Architecture::Aarch64 + } else { + Architecture::X86_64 + }, + os_version: None, + engine: Engine::Docker, + runtime_env: RuntimeEnv::Test, + }, + Config { + enabled: true, + endpoint: Some(args[2].clone()), + max_queue: 256, + }, + )?; + match args[3].as_str() { + "record" => { + telemetry.record(EventData::StepViewed { step: Step::Model })?; + // Intentionally leave the process without shutdown to exercise restart recovery. + } + "flush" => telemetry.flush()?, + "quit" => { + telemetry.record(EventData::StepViewed { + step: Step::Harness, + })?; + telemetry.shutdown()?; + } + "activate" => { + telemetry.record(EventData::Activated)?; + telemetry.record(EventData::Activated)?; + telemetry.shutdown()?; + } + "env" => {} + _ => return Err("unknown probe action".into()), + } + println!("{}", serde_json::to_string(&telemetry.runtime_env())?); + Ok(()) +} diff --git a/desktop/src-tauri/src/desktop_telemetry.rs b/desktop/src-tauri/src/desktop_telemetry.rs new file mode 100644 index 000000000..d50fb030c --- /dev/null +++ b/desktop/src-tauri/src/desktop_telemetry.rs @@ -0,0 +1,331 @@ +//! Tauri integration for the setup emitter. Network work stays off the event loop. +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use openbot_desktop_lib::{engine, quiet, telemetry}; +use tauri::Manager; + +pub struct DesktopTelemetry { + emitter: Arc, + runtime_env: Mutex>, +} + +pub fn initialize(app: &tauri::AppHandle) { + let initialized = (|| { + let context = context(&app.package_info().version.to_string())?; + let emitter = telemetry::Telemetry::open( + app.path().app_local_data_dir().ok()?.join("telemetry"), + context, + telemetry::Config { + enabled: true, + endpoint: Some( + std::env::var("COPILOTKIT_TELEMETRY_URL") + .unwrap_or_else(|_| "https://telemetry.copilotkit.ai/ingest".into()), + ), + max_queue: 256, + }, + ) + .ok()?; + let mut runtime_env = emitter.runtime_env(); + for variable in [ + "COPILOTKIT_TELEMETRY_DISABLED", + "DO_NOT_TRACK", + "COPILOTKIT_TELEMETRY_URL", + ] { + if let Ok(value) = std::env::var(variable) { + runtime_env.insert(variable.into(), value); + } + } + if emitter.install_id().is_none() { + runtime_env.insert("COPILOTKIT_TELEMETRY_DISABLED".into(), "1".into()); + } + app.manage(DesktopTelemetry { + emitter, + runtime_env: Mutex::new(runtime_env), + }); + Some(()) + })(); + if initialized.is_none() { + eprintln!("[telemetry] local setup telemetry could not initialize"); + return; + } + record( + app, + telemetry::EventData::StepViewed { + step: telemetry::Step::Welcome, + }, + ); +} + +fn context(version: &str) -> Option { + use telemetry::{Architecture, Distribution, Engine, NumericVersion, Platform, RuntimeEnv}; + let platform = match std::env::consts::OS { + "macos" => Platform::Macos, + "windows" => Platform::Windows, + "linux" => Platform::Linux, + _ => Platform::Other, + }; + let mut command = match platform { + Platform::Macos => { + let mut c = quiet::command("/usr/bin/sw_vers"); + c.arg("-productVersion"); + c + } + Platform::Windows => { + let mut c = quiet::command("cmd"); + c.args(["/C", "ver"]); + c + } + _ => { + let mut c = quiet::command("uname"); + c.arg("-r"); + c + } + }; + let os_version = command + .output() + .ok() + .filter(|output| output.status.success()) + .and_then(|output| { + String::from_utf8_lossy(&output.stdout) + .split(|c: char| !c.is_ascii_digit() && c != '.') + .find_map(|part| NumericVersion::parse(part).ok()) + }); + Some(telemetry::Context { + distribution: Distribution::Desktop, + app_version: NumericVersion::parse(version.split('-').next()?).ok()?, + platform, + arch: match std::env::consts::ARCH { + "aarch64" => Architecture::Aarch64, + "x86_64" => Architecture::X86_64, + _ => Architecture::Other, + }, + os_version, + engine: Engine::None, + runtime_env: if cfg!(debug_assertions) { + RuntimeEnv::Development + } else { + RuntimeEnv::Production + }, + }) +} + +pub fn record(app: &tauri::AppHandle, event: telemetry::EventData) { + let Some(state) = app.try_state::() else { + return; + }; + if state.emitter.record(event).is_err() { + eprintln!("[telemetry] setup event could not be saved"); + return; + } + let emitter = Arc::clone(&state.emitter); + if std::thread::Builder::new() + .name("openbot-telemetry".into()) + .spawn(move || { + if emitter.flush().is_err() { + eprintln!("[telemetry] delivery deferred until the next flush"); + } + }) + .is_err() + { + eprintln!("[telemetry] delivery deferred until the next flush"); + } +} + +pub fn observe_engine(app: &tauri::AppHandle, status: &engine::EngineStatus) { + let engine = status.address.as_ref().map(|address| address.engine); + let kind = match engine { + Some(engine::Engine::Docker) => telemetry::Engine::Docker, + Some(engine::Engine::Podman) => telemetry::Engine::Podman, + None => telemetry::Engine::None, + }; + if let Some(state) = app.try_state::() { + let _ = state.emitter.update_engine(kind); + if let Ok(mut env) = state.runtime_env.lock() { + env.insert( + "OPENBOT_ENGINE".into(), + engine.map(|value| value.binary()).unwrap_or("none").into(), + ); + } + } + record( + app, + telemetry::EventData::EngineDetected { + engine: kind, + responding: status.responding, + }, + ); +} + +pub fn runtime_env(app: &tauri::AppHandle) -> BTreeMap { + app.try_state::() + .and_then(|state| state.runtime_env.lock().ok().map(|env| env.clone())) + .unwrap_or_default() +} + +pub fn failure( + app: &tauri::AppHandle, + error_class: telemetry::SetupErrorClass, +) { + let Some(state) = app.try_state::() else { + return; + }; + let step = state + .emitter + .snapshot() + .ok() + .and_then(|state| state.last_step) + .unwrap_or(telemetry::Step::Install); + record(app, telemetry::EventData::SetupFailed { step, error_class }); +} + +pub fn pull_completed( + app: &tauri::AppHandle, + metrics: openbot_desktop_lib::pull_metrics::PullMetrics, +) { + record( + app, + telemetry::EventData::ImagePull { + outcome: metrics.outcome, + duration_ms: metrics.duration_ms, + bytes: metrics.bytes, + }, + ); +} + +pub fn shutdown(app: &tauri::AppHandle) { + if let Some(state) = app.try_state::() { + if state.emitter.shutdown().is_err() { + eprintln!("[telemetry] pending setup events retained for next launch"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::temp_root; + use tauri::Manager; + + fn context() -> telemetry::Context { + telemetry::Context { + distribution: telemetry::Distribution::Desktop, + app_version: telemetry::NumericVersion::parse("1.2.3").unwrap(), + platform: telemetry::Platform::Macos, + arch: telemetry::Architecture::Aarch64, + os_version: None, + engine: telemetry::Engine::None, + runtime_env: telemetry::RuntimeEnv::Test, + } + } + + struct Fixture { + app: tauri::App, + window: tauri::WebviewWindow, + } + + impl Fixture { + fn new() -> Self { + let data_dir = temp_root("desktop-telemetry-ipc"); + let emitter = telemetry::Telemetry::open_with_env( + data_dir, + context(), + telemetry::Config { + enabled: true, + endpoint: None, + max_queue: 256, + }, + telemetry::EnvOverride::Enabled, + ) + .unwrap(); + let app = tauri::test::mock_builder() + .manage(DesktopTelemetry { + emitter, + runtime_env: Mutex::new(BTreeMap::new()), + }) + .invoke_handler(tauri::generate_handler![crate::record_setup_event]) + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .unwrap(); + let window = tauri::WebviewWindowBuilder::new(&app, "main", Default::default()) + .build() + .unwrap(); + Self { app, window } + } + + fn invoke(&self, event: serde_json::Value) { + let _ = tauri::test::get_ipc_response( + &self.window, + tauri::webview::InvokeRequest { + cmd: "record_setup_event".into(), + callback: tauri::ipc::CallbackFn(0), + error: tauri::ipc::CallbackFn(1), + url: if cfg!(any(windows, target_os = "android")) { + "http://tauri.localhost" + } else { + "tauri://localhost" + } + .parse() + .unwrap(), + body: tauri::ipc::InvokeBody::Json(serde_json::json!({ "event": event })), + headers: Default::default(), + invoke_key: tauri::test::INVOKE_KEY.into(), + }, + ); + } + + fn queued(&self) -> Vec { + self.app + .state::() + .emitter + .snapshot() + .unwrap() + .queue + .into_iter() + .map(|event| event.data) + .collect() + } + } + + #[test] + fn setup_telemetry_ipc_accepts_only_frontend_setup_events() { + let fixture = Fixture::new(); + + fixture.invoke(serde_json::json!({ + "kind": "step_viewed", + "step": "harness" + })); + fixture.invoke(serde_json::json!({ + "kind": "harness_chosen", + "harness": "byo_url" + })); + assert_eq!(fixture.queued().len(), 2); + + fixture.invoke(serde_json::json!({ + "kind": "harness_chosen", + "harness": "byo_url", + "url": "https://example.invalid" + })); + fixture.invoke(serde_json::json!({ + "kind": "step_viewed", + "step": "credentials" + })); + fixture.invoke(serde_json::json!({ + "kind": "activated" + })); + + let queued = fixture.queued(); + assert_eq!(queued.len(), 2); + assert!(matches!( + queued[0], + telemetry::EventData::StepViewed { + step: telemetry::Step::Harness + } + )); + assert!(matches!( + queued[1], + telemetry::EventData::HarnessChosen { + harness: telemetry::Harness::ByoUrl + } + )); + } +} diff --git a/desktop/src-tauri/src/install.rs b/desktop/src-tauri/src/install.rs index 254ae1b1a..17a18594d 100644 --- a/desktop/src-tauri/src/install.rs +++ b/desktop/src-tauri/src/install.rs @@ -208,6 +208,14 @@ fn digest_of(bytes: &[u8]) -> String { /// somebody presses Start. Answers with the sentence for the step row, or a failure in both /// registers. pub fn install_engine(cache: &Path) -> Result { + install_engine_observed(cache, |_| {}) +} + +/// Observe an actual Podman installer invocation, excluding existing engines and Compose repair. +pub fn install_engine_observed( + cache: &Path, + mut installed: impl FnMut(bool), +) -> Result { let into = crate::acquire::download_dir(cache); // An engine somebody already has is theirs. This only ever adds what is missing. @@ -215,7 +223,9 @@ pub fn install_engine(cache: &Path) -> Result { return place_compose(&into); } - install_podman(&into)?; + let result = install_podman(&into); + installed(result.is_ok()); + result?; // Installed is not found. The MSI extends the *user's* PATH and this process was started with // the old one, so the engine is looked for where the installer puts it rather than on PATH. If diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 7b960bf54..48950481d 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -11,10 +11,12 @@ pub mod intelligence; pub mod plan; pub mod problem; pub mod provider; +pub mod pull_metrics; pub mod quiet; pub mod saved_intent; pub mod stack; pub mod supervise; +pub mod telemetry; pub mod tray; pub mod vault; pub mod windows; diff --git a/desktop/src-tauri/src/main.rs b/desktop/src-tauri/src/main.rs index 52d0534ef..5d20282ba 100644 --- a/desktop/src-tauri/src/main.rs +++ b/desktop/src-tauri/src/main.rs @@ -4,12 +4,15 @@ use std::path::{Path, PathBuf}; use std::sync::Mutex; +mod desktop_telemetry; + #[cfg(test)] mod test_support; use openbot_desktop_lib::{ acquire, deployment, deployment_release, engine, env as openbot_env, harness, install, - problem::Problem, provider, quiet, stack, supervise, tray, windows as win, + problem::Problem, provider, pull_metrics, quiet, stack, supervise, telemetry, tray, + windows as win, }; const QUIT_CLEANUP_NOTICE_FILE: &str = ".openbot-quit-cleanup-notice"; @@ -259,6 +262,17 @@ fn report( ok: bool, detail: impl Into, ) { + if !ok { + let error_class = match step { + "install-engine" => telemetry::SetupErrorClass::EngineInstallFailed, + "engine" | "create-machine" | "start-machine" | "health-gate" => { + telemetry::SetupErrorClass::EngineUnavailable + } + "env" | "ports" => telemetry::SetupErrorClass::InvalidConfiguration, + _ => telemetry::SetupErrorClass::Unknown, + }; + desktop_telemetry::failure(app, error_class); + } let _ = app.emit( "setup:progress", Progress { @@ -292,13 +306,46 @@ fn cleanup_root(shell: &Shell, fallback_root: &Path) -> PathBuf { } #[tauri::command] -fn detect_engine() -> engine::EngineStatus { - engine::detect() +fn detect_engine(app: tauri::AppHandle) -> engine::EngineStatus { + let status = engine::detect(); + desktop_telemetry::observe_engine(&app, &status); + status +} + +#[tauri::command] +fn windows_blocker( + app: tauri::AppHandle, +) -> Result, Problem> { + let result = win::blocker(); + if cfg!(target_os = "windows") { + use telemetry::WindowsStageOutcome as Outcome; + let outcome = match &result { + Ok(None) => Outcome::Ready, + Ok(Some(win::Blocker::WslAbsent)) => Outcome::WslAbsent, + Ok(Some(win::Blocker::WslOne)) => Outcome::WslOne, + Ok(Some(win::Blocker::WslNoKernel)) => Outcome::WslNoKernel, + Ok(Some(win::Blocker::VirtualMachinePlatformDisabled)) => { + Outcome::VirtualMachinePlatformDisabled + } + Ok(Some(win::Blocker::VirtualizationDisabled)) => Outcome::VirtualizationDisabled, + Ok(Some(win::Blocker::NotAdministrator)) => Outcome::NotAdministrator, + Err(_) => Outcome::CheckFailed, + }; + desktop_telemetry::record(&app, telemetry::EventData::WindowsStage { outcome }); + } + result } #[tauri::command] -fn windows_blocker() -> Result, Problem> { - win::blocker() +fn record_setup_event(app: tauri::AppHandle, event: serde_json::Value) { + if let Ok( + event @ (telemetry::EventData::StepViewed { .. } + | telemetry::EventData::HarnessChosen { .. } + | telemetry::EventData::ModelChosen { .. }), + ) = serde_json::from_value(event) + { + desktop_telemetry::record(&app, event); + } } #[tauri::command] @@ -327,6 +374,7 @@ async fn prepare_engine(app: tauri::AppHandle) -> Result Result { let found = engine::detect(); + desktop_telemetry::observe_engine(app, &found); let root = stack::default_root(); let existing = tauri::async_runtime::spawn_blocking(move || { ready_responding_engine_after_compose_repair( @@ -370,17 +418,33 @@ async fn engine_ready(app: &tauri::AppHandle) -> Result report(app, "install-engine", true, said), + Ok(said) => { + report(app, "install-engine", true, said); + } Err(problem) => { report(app, "install-engine", false, problem.said.clone()); return Err(problem); @@ -409,6 +473,7 @@ async fn engine_ready(app: &tauri::AppHandle) -> Result( * store, and travel from there to the processes that need them as environment, which is where * a secret can live without being written down. See `vault` for what each platform gets. */ - let (settings, secrets) = openbot_desktop_lib::vault::split(settings); + let (settings, mut secrets) = openbot_desktop_lib::vault::split(settings); /* * The credentials, plus any setting this answer dropped. * @@ -904,6 +969,8 @@ async fn start_stack_inner( &credential, )?; report(&app, "env", true, "settings written, credentials stored"); + // Set before Bun imports the runtime, and retained for supervised restarts. + secrets.extend(desktop_telemetry::runtime_env(&app)); // Said before rather than after. On a machine that has never run OpenBot this pulls five // images, and a person watching a button that says "Working" has no way to tell a download @@ -955,6 +1022,17 @@ async fn start_stack_inner( */ let bundled_bots = stack::BundledBots::for_credential(&credential); attempt.require_current()?; + stack::pull( + &found, + &root, + installed_harness, + bundled_bots, + &secrets, + |metrics| { + desktop_telemetry::pull_completed(&app, metrics); + }, + )?; + attempt.require_current()?; // Even a failed up can have started some services. Keep their root until down succeeds. *shell.containers.lock().unwrap() = Some(ContainerDeployment { root: root.clone(), @@ -2007,11 +2085,17 @@ facts about the deployment, and a window carrying them would be a second copy to */ #[tauri::command] async fn ask_the_bot( - _app: tauri::AppHandle, + app: tauri::AppHandle, root: String, question: String, ) -> Result { - ask_the_bot_inner(stack::root_from(&root), question).await + let result = ask_the_bot_inner(stack::root_from(&root), question).await; + if result.is_ok() { + desktop_telemetry::record(&app, telemetry::EventData::Activated); + } else { + desktop_telemetry::failure(&app, telemetry::SetupErrorClass::Unknown); + } + result } async fn ask_the_bot_inner(root: PathBuf, question: String) -> Result { @@ -2224,8 +2308,12 @@ async fn begin_claude_sign_in(app: tauri::AppHandle, root: String) -> Result(); exit_cleanup_with( &shell, diff --git a/desktop/src-tauri/src/pull_metrics.rs b/desktop/src-tauri/src/pull_metrics.rs new file mode 100644 index 000000000..899e078e2 --- /dev/null +++ b/desktop/src-tauri/src/pull_metrics.rs @@ -0,0 +1,334 @@ +//! Metrics for an explicit image pull, before containers or provider sign-in start. + +use std::collections::HashMap; +use std::io::Write; +use std::process::{Command, Stdio}; +use std::time::Instant; + +use serde::Deserialize; + +use crate::engine::Address; +use crate::problem::Problem; +use crate::telemetry::Outcome; + +/// No image names, layer IDs, engine output, or credentials cross this boundary. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct PullMetrics { + pub outcome: Outcome, + /// Wall time of the pull command, excluding capability detection and container startup. + pub duration_ms: u64, + /// Download counters actually observed. None means the provider supplied no byte counters. + /// This excludes extraction, cached image sizes, and unobserved bytes between progress updates. + pub bytes: Option, +} + +/// None preserves the existing implicit pull on providers without a missing-only pull policy. +/// Such a provider gets no pull metric: timing the later `up` would also measure startup. +/// Some(false) still measures explicit pulls, but the older progress format has unknown bytes. +/// Podman delegates to its selected external Compose provider through this same command path. +pub(crate) fn compose_pull_progress(engine: &Address) -> Option { + let policy_help = engine + .command() + .args(["compose", "pull", "--help"]) + .output() + .ok()?; + if !policy_help.status.success() + || !(option_advertises(&policy_help.stdout, "--policy", "missing") + || option_advertises(&policy_help.stderr, "--policy", "missing")) + { + return None; + } + Some( + engine + .command() + .args(["compose", "--help"]) + .output() + .ok() + .filter(|output| output.status.success()) + .is_some_and(|output| { + supports_json_progress(&output.stdout) || supports_json_progress(&output.stderr) + }), + ) +} + +fn supports_json_progress(help: &[u8]) -> bool { + option_advertises(help, "--progress", "json") +} + +fn option_advertises(help: &[u8], flag: &str, value: &str) -> bool { + let text = String::from_utf8_lossy(help); + let Some((_, option)) = text.split_once(flag) else { + return false; + }; + option + .split("--") + .next() + .unwrap_or_default() + .split(|character: char| !character.is_ascii_alphabetic()) + .any(|word| word == value) +} + +#[derive(Deserialize)] +struct Progress { + id: String, + parent_id: Option, + text: String, + current: Option, +} + +/// Compose's JSON writer carries Docker's byte counters as integers: +/// https://github.com/docker/compose/blob/v2.39.2/pkg/compose/pull.go#L391-L440 +/// The same layer can appear beneath multiple services. Keep its maximum download counter, +/// regardless of parent, and never add extraction progress or the advertised total size. +fn download_bytes(stdout: &[u8], stderr: &[u8]) -> Option { + let mut layers: HashMap = HashMap::new(); + for output in [stdout, stderr] { + for line in output.split(|byte| *byte == b'\n') { + let Ok(progress) = serde_json::from_slice::(line) else { + continue; + }; + if progress.text != "Downloading" + || progress.id.is_empty() + || progress.parent_id.as_deref().unwrap_or_default().is_empty() + { + continue; + } + if let Some(current) = progress.current { + layers + .entry(progress.id) + .and_modify(|maximum| *maximum = (*maximum).max(current)) + .or_insert(current); + } + } + } + if layers.is_empty() { + None + } else { + layers + .values() + .try_fold(0_u64, |sum, current| sum.checked_add(*current)) + } +} + +/// Pull a provider sign-in image without starting its CLI or creating a container. +/// A tiny stdin Compose project gives the same missing-only policy and JSON progress as setup. +/// Providers without `pull --policy missing` retain their implicit pull and emit no metric. +pub fn pull_image( + engine: &Address, + image: &str, + on_complete: impl FnOnce(PullMetrics), +) -> Result<(), Problem> { + let Some(json_progress) = compose_pull_progress(engine) else { + return Ok(()); + }; + let mut command = engine.command(); + command.arg("compose"); + if json_progress { + command.args(["--progress", "json"]); + } + command.args([ + "--project-name", + "openbot-image-pull", + "-f", + "-", + "pull", + "--policy", + "missing", + ]); + let project = serde_json::json!({"services": {"image": {"image": image}}}).to_string(); + run( + command, + Some(project.as_bytes()), + json_progress, + on_complete, + ) +} + +/// Complete one explicit pull and report its outcome before the caller can start containers. +/// A failed pull is returned as the same two-part Problem used by the existing startup path. +pub(crate) fn run( + mut command: Command, + input: Option<&[u8]>, + json_progress: bool, + on_complete: impl FnOnce(PullMetrics), +) -> Result<(), Problem> { + command + .stdin(if input.is_some() { + Stdio::piped() + } else { + Stdio::null() + }) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let started = Instant::now(); + let output = (|| { + let mut child = command.spawn()?; + if let Some(input) = input { + let result = child + .stdin + .take() + .expect("piped pull stdin") + .write_all(input); + if let Err(error) = result { + let _ = child.kill(); + let _ = child.wait(); + return Err(error); + } + } + child.wait_with_output() + })(); + let duration_ms = started.elapsed().as_millis().min(u64::MAX as u128) as u64; + let success = output.as_ref().is_ok_and(|output| output.status.success()); + let bytes = if json_progress { + output + .as_ref() + .ok() + .and_then(|output| download_bytes(&output.stdout, &output.stderr)) + } else { + None + }; + on_complete(PullMetrics { + outcome: if success { + Outcome::Success + } else { + Outcome::Failure + }, + duration_ms, + bytes, + }); + let output = output.map_err(|error| { + Problem::plain(format!("Could not pull the images OpenBot needs: {error}")) + })?; + if output.status.success() { + return Ok(()); + } + let raw = crate::quiet::said(if output.stderr.is_empty() { + &output.stdout + } else { + &output.stderr + }); + Err(Problem::with(crate::problem::said_about(&raw), raw)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn counts_each_shared_layer_once_and_ignores_extraction() { + let output = br#" +{"id":"layer-a","parent_id":"service-a","text":"Downloading","current":20,"total":100} +{"id":"layer-a","parent_id":"service-b","text":"Downloading","current":80,"total":100} +{"id":"layer-a","parent_id":"service-a","text":"Downloading","current":60,"total":100} +{"id":"layer-b","parent_id":"service-a","text":"Downloading","current":7,"total":7} +{"id":"layer-a","parent_id":"service-a","text":"Extracting","current":900,"total":900} +{"id":"layer-a","parent_id":"service-a","text":"Download complete"} +"#; + assert_eq!(download_bytes(output, b""), Some(87)); + } + + #[test] + fn missing_download_counters_are_unknown_even_after_a_successful_pull() { + // Observed with Docker 29.5.2 / hello-world:latest: download completion has no + // byte counters, while extraction reports current=1. That is not one downloaded byte. + let output = br#" +{"id":"layer-a","parent_id":"image","text":"Download complete","details":"0B","percent":100} +{"id":"layer-a","parent_id":"image","text":"Extracting","current":1} +{"id":"image","text":"Pulled"} +"#; + assert_eq!(download_bytes(output, b""), None); + assert_eq!( + download_bytes(b"", b"image Skipped - Image is already present locally"), + None + ); + } + + #[test] + fn accepts_counters_on_either_pipe_without_counting_noise_or_totals() { + let stdout = + br#"{"id":"layer-a","parent_id":"image","text":"Downloading","current":5,"total":50}"#; + let stderr = br#" +Executing external compose provider +{"id":"layer-a","parent_id":"image","text":"Downloading","current":8,"total":50} +{"id":"layer-b","parent_id":"image","text":"Downloading","total":999} +{"id":"layer-c","parent_id":"image","text":"Downloading","current":-1} +{"id":"image","text":"Downloading","current":10000} +"#; + assert_eq!(download_bytes(stdout, stderr), Some(8)); + } + + #[test] + fn json_capability_requires_the_progress_option_to_advertise_json() { + assert!(supports_json_progress( + b"--progress string Set type of progress output (auto,\n tty, plain, json, quiet)\n--project-directory string" + )); + assert!(!supports_json_progress( + b"--progress string (auto, tty, plain)\n--format string (json)" + )); + assert!(!supports_json_progress(b"--format string (json)")); + } + + #[test] + fn missing_policy_must_be_advertised_before_an_explicit_pull() { + assert!(option_advertises( + b"--policy string Apply pull policy (\"missing\"|\"always\")\n--quiet", + "--policy", + "missing" + )); + // Compose v2.20 has no --policy; leave its existing implicit pull alone. + assert!(!option_advertises( + b"--include-deps Also pull dependencies\n--quiet", + "--policy", + "missing" + )); + } + + #[cfg(unix)] + #[test] + fn reports_failure_and_observed_bytes_once_before_returning_the_problem() { + let mut command = crate::quiet::command("sh"); + command.args(["-c", "printf '%s\\n' '{\"id\":\"layer\",\"parent_id\":\"image\",\"text\":\"Downloading\",\"current\":12}' >&2; exit 1"]); + let mut reports = Vec::new(); + let result = run(command, None, true, |metrics| reports.push(metrics)); + assert!(result.is_err()); + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].outcome, crate::telemetry::Outcome::Failure); + assert_eq!(reports[0].bytes, Some(12)); + } + + #[cfg(unix)] + #[test] + fn measures_the_child_operation_and_closes_its_stdin() { + let mut command = crate::quiet::command("sh"); + command.args(["-c", "cat; sleep 0.02"]); + let mut reports = Vec::new(); + let input = br#"{"id":"layer","parent_id":"image","text":"Downloading","current":9}"#; + assert!(run(command, Some(input), true, |metrics| reports.push(metrics)).is_ok()); + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].outcome, crate::telemetry::Outcome::Success); + assert!(reports[0].duration_ms >= 10); + assert_eq!(reports[0].bytes, Some(9)); + } + + #[cfg(unix)] + #[test] + fn non_json_provider_keeps_bytes_unknown() { + let mut command = crate::quiet::command("sh"); + command.args(["-c", "cat"]); + let mut reports = Vec::new(); + let input = br#"{"id":"layer","parent_id":"image","text":"Downloading","current":9}"#; + assert!(run(command, Some(input), false, |metrics| reports.push(metrics)).is_ok()); + assert_eq!(reports[0].outcome, crate::telemetry::Outcome::Success); + assert_eq!(reports[0].bytes, None); + } + + #[test] + fn reports_spawn_failure_once() { + let command = crate::quiet::command("openbot-test-missing-pull-executable"); + let mut reports = Vec::new(); + assert!(run(command, None, true, |metrics| reports.push(metrics)).is_err()); + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].outcome, crate::telemetry::Outcome::Failure); + assert_eq!(reports[0].bytes, None); + } +} diff --git a/desktop/src-tauri/src/stack.rs b/desktop/src-tauri/src/stack.rs index f024f2728..b4a41dee7 100644 --- a/desktop/src-tauri/src/stack.rs +++ b/desktop/src-tauri/src/stack.rs @@ -179,6 +179,36 @@ fn compose_command(engine: &Address, root: &Path, secrets: &Secrets) -> Command command } +/// Pull the selected stack before `up`, including the one-shot migration and service dependencies. +/// Keep this separate from `up`: image transfer must not include container startup or migrations. +/// Older providers without a missing-only pull policy retain the existing implicit pull in `up`; +/// no pull metric is reported for that unmeasurable path. +pub fn pull( + engine: &Address, + root: &Path, + harness: bool, + bots: BundledBots, + secrets: &Secrets, + on_complete: impl FnOnce(crate::pull_metrics::PullMetrics), +) -> Result<(), Problem> { + let Some(json_progress) = crate::pull_metrics::compose_pull_progress(engine) else { + return Ok(()); + }; + let mut requested = selected_services(harness, bots); + requested.push("migrate"); + let mut command = compose_command(engine, root, secrets); + if json_progress { + command.args(["--progress", "json"]); + } + if harness { + command.args(["--profile", "harness"]); + } + command + .args(["pull", "--policy", "missing", "--include-deps"]) + .args(&requested); + crate::pull_metrics::run(command, None, json_progress, on_complete) +} + /// Raise the containers. /// /// `--no-build` is the point of the whole published-images job: a desktop install has no toolchain, diff --git a/desktop/src-tauri/src/telemetry.rs b/desktop/src-tauri/src/telemetry.rs new file mode 100644 index 000000000..aa57f29b8 --- /dev/null +++ b/desktop/src-tauri/src/telemetry.rs @@ -0,0 +1,1559 @@ +use rand::RngCore; +use reqwest::blocking::Client; +use reqwest::redirect::Policy; +use serde::{Deserialize, Serialize}; +use std::env; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::thread::sleep; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +pub const STATE_SCHEMA_VERSION: u8 = 1; +const STATE_FILE: &str = "telemetry-state.json"; +const REQUEST_TIMEOUT: Duration = Duration::from_secs(3); +const FLUSH_BUDGET: Duration = Duration::from_secs(6); + +pub type Result = std::result::Result; + +#[derive(Debug)] +pub enum TelemetryError { + Io(io::Error), + Json(serde_json::Error), + Http(reqwest::Error), + MissingEndpoint, + InvalidState(String), + Poisoned, +} + +impl From for TelemetryError { + fn from(error: io::Error) -> Self { + Self::Io(error) + } +} + +impl From for TelemetryError { + fn from(error: serde_json::Error) -> Self { + Self::Json(error) + } +} + +impl From for TelemetryError { + fn from(error: reqwest::Error) -> Self { + Self::Http(error) + } +} + +impl std::fmt::Display for TelemetryError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Io(error) => write!(formatter, "telemetry I/O error: {error}"), + Self::Json(error) => write!(formatter, "telemetry JSON error: {error}"), + Self::Http(error) => write!(formatter, "telemetry HTTP error: {error}"), + Self::MissingEndpoint => formatter.write_str("telemetry endpoint is missing"), + Self::InvalidState(error) => write!(formatter, "invalid telemetry state: {error}"), + Self::Poisoned => formatter.write_str("telemetry state lock is poisoned"), + } + } +} + +impl std::error::Error for TelemetryError {} + +#[derive(Debug, Clone)] +pub struct Config { + pub enabled: bool, + pub endpoint: Option, + pub max_queue: usize, +} + +impl Config { + pub fn disabled_for_tests() -> Self { + Self { + enabled: true, + endpoint: None, + max_queue: 256, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EnvOverride { + Process, + Enabled, + Disabled, +} + +#[derive(Debug)] +pub struct Telemetry { + state_path: PathBuf, + context: Mutex, + config: Config, + state: Mutex>, + session_last_step_viewed: Mutex>, + flush_lock: Mutex<()>, + client: Client, +} + +impl Telemetry { + pub fn open(data_dir: impl AsRef, context: Context, config: Config) -> Result> { + Self::open_with_env(data_dir, context, config, EnvOverride::Process) + } + + pub fn open_with_env( + data_dir: impl AsRef, + context: Context, + config: Config, + env_override: EnvOverride, + ) -> Result> { + let state_path = state_path(data_dir.as_ref()); + if telemetry_disabled(env_override) || !config.enabled { + let _ = fs::remove_file(&state_path); + return Ok(Arc::new(Self::disabled(state_path, context, config)?)); + } + + fs::create_dir_all(data_dir.as_ref())?; + let mut state = if state_path.exists() { + let state = serde_json::from_slice::(&fs::read(&state_path)?)?; + state.validate()?; + state + } else { + PersistedState::new(new_uuid()) + }; + + if !state.activated { + if let Some(step) = state.last_step.take() { + state.push_bounded( + Event::new(EventData::SetupAbandoned { step }, context.clone()), + normalized_max_queue(config.max_queue), + ); + } + } + write_state(&state_path, &state)?; + + Ok(Arc::new(Self { + state_path, + context: Mutex::new(context), + config, + state: Mutex::new(Some(state)), + session_last_step_viewed: Mutex::new(None), + flush_lock: Mutex::new(()), + client: client()?, + })) + } + + fn disabled(state_path: PathBuf, context: Context, config: Config) -> Result { + Ok(Self { + state_path, + context: Mutex::new(context), + config, + state: Mutex::new(None), + session_last_step_viewed: Mutex::new(None), + flush_lock: Mutex::new(()), + client: client()?, + }) + } + + pub fn install_id(&self) -> Option { + self.state + .lock() + .ok() + .and_then(|guard| guard.as_ref().map(|state| state.install_id.clone())) + } + + pub fn update_engine(&self, engine: Engine) -> Result<()> { + let mut context = self.context.lock().map_err(|_| TelemetryError::Poisoned)?; + context.engine = engine; + Ok(()) + } + + /// The exact process environment the desktop gives to Bun before runtime imports. + pub fn runtime_env(&self) -> std::collections::BTreeMap { + let mut env = std::collections::BTreeMap::new(); + let Some(id) = self.install_id() else { + env.insert("COPILOTKIT_TELEMETRY_DISABLED".into(), "1".into()); + return env; + }; + let Ok(context) = self.context.lock() else { + return env; + }; + env.insert("CPK_TELEMETRY_ID".into(), id); + env.insert("COPILOTKIT_TELEMETRY_SAMPLE_RATE".into(), "1".into()); + env.insert( + "OPENBOT_DISTRIBUTION".into(), + context.distribution.as_str().into(), + ); + env.insert("OPENBOT_VERSION".into(), context.app_version.to_string()); + env.insert("OPENBOT_PLATFORM".into(), context.platform.as_str().into()); + env.insert("OPENBOT_ARCH".into(), context.arch.as_str().into()); + env.insert("OPENBOT_ENGINE".into(), context.engine.as_str().into()); + if let Some(version) = &context.os_version { + env.insert("OPENBOT_OS_VERSION".into(), version.to_string()); + } + env + } + + pub fn record(&self, data: EventData) -> Result<()> { + let context = self + .context + .lock() + .map_err(|_| TelemetryError::Poisoned)? + .clone(); + let mut guard = self.state.lock().map_err(|_| TelemetryError::Poisoned)?; + let Some(state) = guard.as_mut() else { + return Ok(()); + }; + + match &data { + EventData::StepViewed { step } => { + let mut last_step = self + .session_last_step_viewed + .lock() + .map_err(|_| TelemetryError::Poisoned)?; + if *last_step == Some(*step) { + return Ok(()); + } + *last_step = Some(*step); + state.last_step = Some(*step); + } + EventData::Activated => { + if state.activated { + return Ok(()); + } + state.activated = true; + state.last_step = None; + *self + .session_last_step_viewed + .lock() + .map_err(|_| TelemetryError::Poisoned)? = None; + } + EventData::SetupAbandoned { .. } => { + if state.activated { + return Ok(()); + } + state.last_step = None; + *self + .session_last_step_viewed + .lock() + .map_err(|_| TelemetryError::Poisoned)? = None; + } + _ => { + *self + .session_last_step_viewed + .lock() + .map_err(|_| TelemetryError::Poisoned)? = None; + } + } + + state.push_bounded( + Event::new(data, context), + normalized_max_queue(self.config.max_queue), + ); + write_state(&self.state_path, state) + } + + pub fn flush(&self) -> Result<()> { + self.flush_with_lock_wait(Duration::ZERO) + } + + fn flush_with_lock_wait(&self, lock_wait: Duration) -> Result<()> { + let started = Instant::now(); + let flush_guard = loop { + match self.flush_lock.try_lock() { + Ok(guard) => break guard, + Err(_) if started.elapsed() < lock_wait => sleep(Duration::from_millis(10)), + Err(_) => return Ok(()), + } + }; + let _flush_guard = flush_guard; + let endpoint = match self.config.endpoint.as_ref() { + Some(endpoint) if !endpoint.trim().is_empty() => endpoint, + _ => return Ok(()), + }; + + loop { + if started.elapsed() >= FLUSH_BUDGET { + return Ok(()); + } + let (install_id, event) = { + let guard = self.state.lock().map_err(|_| TelemetryError::Poisoned)?; + let Some(state) = guard.as_ref() else { + return Ok(()); + }; + let Some(event) = state.queue.first().cloned() else { + return Ok(()); + }; + (state.install_id.clone(), event) + }; + + let response = self + .client + .post(endpoint) + .header("X-CopilotKit-Telemetry-Id", &install_id) + .json(&event.to_sink_payload()) + .send(); + match response { + Ok(response) if response.status().is_success() => { + let mut guard = self.state.lock().map_err(|_| TelemetryError::Poisoned)?; + if let Some(state) = guard.as_mut() { + if let Some(index) = state + .queue + .iter() + .position(|queued| queued.event_id == event.event_id) + { + state.queue.remove(index); + write_state(&self.state_path, state)?; + } + } + } + Ok(_) => return Ok(()), + Err(error) => return Err(TelemetryError::Http(error)), + } + } + } + + pub fn shutdown(&self) -> Result<()> { + let step = { + let guard = self.state.lock().map_err(|_| TelemetryError::Poisoned)?; + guard + .as_ref() + .and_then(|state| (!state.activated).then_some(state.last_step).flatten()) + }; + if let Some(step) = step { + self.record(EventData::SetupAbandoned { step })?; + } + self.flush_with_lock_wait(FLUSH_BUDGET) + } + + pub fn snapshot(&self) -> Result { + self.state + .lock() + .map_err(|_| TelemetryError::Poisoned)? + .clone() + .ok_or(TelemetryError::MissingEndpoint) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct SinkPayload { + pub event: String, + pub event_id: String, + pub properties: serde_json::Value, + pub global_properties: serde_json::Value, + pub package: SinkPackage, + pub ts: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct SinkPackage { + pub name: String, + pub version: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct Context { + #[serde(rename = "openbot_distribution")] + pub distribution: Distribution, + #[serde(rename = "openbot_version")] + pub app_version: NumericVersion, + #[serde(rename = "openbot_platform")] + pub platform: Platform, + #[serde(rename = "openbot_arch")] + pub arch: Architecture, + #[serde(rename = "openbot_os_version", skip_serializing_if = "Option::is_none")] + pub os_version: Option, + #[serde(rename = "openbot_engine")] + pub engine: Engine, + pub runtime_env: RuntimeEnv, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NumericVersion { + pub parts: Vec, +} + +impl NumericVersion { + pub fn new(parts: Vec) -> Result { + if (2..=4).contains(&parts.len()) { + Ok(Self { parts }) + } else { + Err(TelemetryError::InvalidState( + "numeric version must have 2 to 4 dot-separated numeric parts".to_string(), + )) + } + } + + pub fn parse(value: &str) -> Result { + if value.len() > 32 { + return Err(TelemetryError::InvalidState( + "numeric version is too long".to_string(), + )); + } + let parts = value + .split('.') + .map(|part| { + if part.is_empty() || !part.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(TelemetryError::InvalidState(format!( + "numeric version contains a non-numeric part: {value}" + ))); + } + part.parse::().map_err(|_| { + TelemetryError::InvalidState(format!( + "numeric version part is too large: {value}" + )) + }) + }) + .collect::>>()?; + Self::new(parts) + } +} + +impl std::fmt::Display for NumericVersion { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + for (index, part) in self.parts.iter().enumerate() { + if index > 0 { + formatter.write_str(".")?; + } + write!(formatter, "{part}")?; + } + Ok(()) + } +} + +impl Serialize for NumericVersion { + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + serializer.serialize_str(&self.to_string()) + } +} + +impl<'de> Deserialize<'de> for NumericVersion { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + let value = String::deserialize(deserializer)?; + Self::parse(&value).map_err(serde::de::Error::custom) + } +} + +impl Context { + fn to_global_properties(&self) -> serde_json::Value { + let mut properties = serde_json::json!({ + "openbot_distribution": self.distribution, + "openbot_version": &self.app_version, + "openbot_platform": self.platform, + "openbot_arch": self.arch, + "openbot_engine": self.engine, + "runtime_env": self.runtime_env, + "sampleRate": 1, + "sampleWeight": 1, + "sampleRateAdjustmentFactor": 0 + }); + if let Some(os_version) = &self.os_version { + properties["openbot_os_version"] = + serde_json::to_value(os_version).unwrap_or(serde_json::Value::Null); + } + properties + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct PersistedState { + pub schema_version: u8, + pub install_id: String, + pub activated: bool, + pub last_step: Option, + pub queue: Vec, +} + +impl PersistedState { + fn new(install_id: String) -> Self { + Self { + schema_version: STATE_SCHEMA_VERSION, + install_id, + activated: false, + last_step: None, + queue: Vec::new(), + } + } + + fn push_bounded(&mut self, event: Event, max_queue: usize) { + self.queue.push(event); + while self.queue.len() > max_queue { + self.queue.remove(0); + } + } + + fn validate(&self) -> Result<()> { + if self.schema_version != STATE_SCHEMA_VERSION { + return Err(TelemetryError::InvalidState(format!( + "unsupported telemetry state schema {}", + self.schema_version + ))); + } + if !is_uuid(&self.install_id) { + return Err(TelemetryError::InvalidState( + "install_id is not a UUID".to_string(), + )); + } + for event in &self.queue { + event.validate()?; + } + Ok(()) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct Event { + pub event_id: String, + pub occurred_at_ms: u64, + pub event_name: String, + pub context: Context, + pub data: EventData, +} + +impl Event { + pub fn new(data: EventData, context: Context) -> Self { + let event_name = data.event_name().to_string(); + Self { + event_id: new_uuid(), + occurred_at_ms: now_ms(), + event_name, + context, + data, + } + } + + pub fn event_name(&self) -> &str { + &self.event_name + } + + fn validate(&self) -> Result<()> { + if !is_uuid(&self.event_id) { + return Err(TelemetryError::InvalidState( + "event_id is not a UUID".to_string(), + )); + } + if self.event_name != self.data.event_name() { + return Err(TelemetryError::InvalidState(format!( + "event_name {} does not match {:?}", + self.event_name, self.data + ))); + } + Ok(()) + } + + fn to_sink_payload(&self) -> SinkPayload { + SinkPayload { + event: self.event_name.clone(), + event_id: self.event_id.clone(), + properties: serde_json::to_value(&self.data).unwrap_or_else(|_| serde_json::json!({})), + global_properties: self.context.to_global_properties(), + package: SinkPackage { + name: "openbot-desktop".to_string(), + version: self.context.app_version.to_string(), + }, + ts: self.occurred_at_ms / 1000, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum EventData { + StepViewed { + step: Step, + }, + HarnessChosen { + harness: Harness, + }, + ModelChosen { + provider: Provider, + credential_path: CredentialPath, + custom_base_url: bool, + }, + EngineDetected { + engine: Engine, + responding: bool, + }, + EngineInstalled { + engine: Engine, + outcome: EngineInstallOutcome, + }, + WindowsStage { + outcome: WindowsStageOutcome, + }, + ImagePull { + outcome: Outcome, + duration_ms: u64, + bytes: Option, + }, + SetupFailed { + step: Step, + error_class: SetupErrorClass, + }, + Activated, + SetupAbandoned { + step: Step, + }, +} + +impl EventData { + pub fn event_name(&self) -> &'static str { + match self { + Self::StepViewed { .. } => "oss.desktop.step_viewed", + Self::HarnessChosen { .. } => "oss.desktop.harness_chosen", + Self::ModelChosen { .. } => "oss.desktop.model_chosen", + Self::EngineDetected { .. } => "oss.desktop.engine_detected", + Self::EngineInstalled { .. } => "oss.desktop.engine_installed", + Self::WindowsStage { .. } => "oss.desktop.windows_stage", + Self::ImagePull { .. } => "oss.desktop.image_pull", + Self::SetupFailed { .. } => "oss.desktop.setup_failed", + Self::Activated => "oss.desktop.activated", + Self::SetupAbandoned { .. } => "oss.desktop.setup_abandoned", + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Distribution { + Desktop, +} + +impl Distribution { + fn as_str(self) -> &'static str { + "desktop" + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeEnv { + Development, + Production, + Test, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Platform { + Windows, + Macos, + Linux, + Other, +} + +impl Platform { + fn as_str(self) -> &'static str { + match self { + Self::Windows => "windows", + Self::Macos => "macos", + Self::Linux => "linux", + Self::Other => "other", + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Architecture { + X86_64, + Aarch64, + Other, +} + +impl Architecture { + fn as_str(self) -> &'static str { + match self { + Self::X86_64 => "x86_64", + Self::Aarch64 => "aarch64", + Self::Other => "other", + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Step { + Welcome, + Harness, + Model, + Install, + Ask, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Harness { + Crewai, + Llamaindex, + Agno, + Langgraph, + GoogleAdk, + PydanticAi, + MicrosoftAgentFramework, + ClaudeAgentSdk, + Strands, + Ag2, + Langroid, + Mastra, + ByoUrl, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Provider { + Openai, + Anthropic, + Compatible, + None, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum CredentialPath { + Subscription, + ApiKey, + None, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Engine { + Docker, + Podman, + None, +} + +impl Engine { + fn as_str(self) -> &'static str { + match self { + Self::Docker => "docker", + Self::Podman => "podman", + Self::None => "none", + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum EngineInstallOutcome { + Success, + Failure, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum WindowsStageOutcome { + Ready, + WslAbsent, + WslOne, + WslNoKernel, + VirtualMachinePlatformDisabled, + VirtualizationDisabled, + NotAdministrator, + CheckFailed, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Outcome { + Success, + Failure, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SetupErrorClass { + EngineUnavailable, + EngineInstallFailed, + ImagePullFailed, + InvalidConfiguration, + NetworkUnavailable, + PermissionDenied, + Unknown, +} + +pub fn schema_json() -> serde_json::Value { + serde_json::json!({ + "schema_version": STATE_SCHEMA_VERSION, + "event_name_prefix": "oss.desktop.", + "numeric_version_pattern": r"^\d+(?:\.\d+){1,3}$", + "context": { + "openbot_distribution": ["desktop"], + "openbot_version": { "pattern": r"^\d+(?:\.\d+){1,3}$" }, + "openbot_platform": ["windows", "macos", "linux", "other"], + "openbot_arch": ["x86_64", "aarch64", "other"], + "openbot_os_version": { "optional": true, "pattern": r"^\d+(?:\.\d+){1,3}$" }, + "openbot_engine": ["docker", "podman", "none"], + "runtime_env": ["development", "production", "test"] + }, + "events": { + "step_viewed": { "step": ["welcome", "harness", "model", "install", "ask"] }, + "harness_chosen": { "harness": ["crewai", "llamaindex", "agno", "langgraph", "google_adk", "pydantic_ai", "microsoft_agent_framework", "claude_agent_sdk", "strands", "ag2", "langroid", "mastra", "byo_url"] }, + "model_chosen": { "provider": ["openai", "anthropic", "compatible", "none"], "credential_path": ["subscription", "api_key", "none"], "custom_base_url": "bool" }, + "engine_detected": { "engine": ["docker", "podman", "none"], "responding": "bool" }, + "engine_installed": { "engine": ["docker", "podman", "none"], "outcome": ["success", "failure"] }, + "windows_stage": { "outcome": ["ready", "wsl_absent", "wsl_one", "wsl_no_kernel", "virtual_machine_platform_disabled", "virtualization_disabled", "not_administrator", "check_failed"] }, + "image_pull": { "outcome": ["success", "failure"], "duration_ms": "u64", "bytes": "option_u64" }, + "setup_failed": { "step": ["welcome", "harness", "model", "install", "ask"], "error_class": ["engine_unavailable", "engine_install_failed", "image_pull_failed", "invalid_configuration", "network_unavailable", "permission_denied", "unknown"] }, + "activated": {}, + "setup_abandoned": { "step": ["welcome", "harness", "model", "install", "ask"] } + } + }) +} + +pub fn state_path(data_dir: impl AsRef) -> PathBuf { + data_dir.as_ref().join(STATE_FILE) +} + +fn client() -> Result { + Ok(Client::builder() + .timeout(REQUEST_TIMEOUT) + .connect_timeout(REQUEST_TIMEOUT) + .redirect(Policy::none()) + .build()?) +} + +fn write_state(path: &Path, state: &PersistedState) -> Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + crate::env::write_private_file(path, &serde_json::to_vec(state)?)?; + Ok(()) +} + +fn normalized_max_queue(configured: usize) -> usize { + configured.clamp(1, 256) +} + +fn telemetry_disabled(env_override: EnvOverride) -> bool { + match env_override { + EnvOverride::Enabled => false, + EnvOverride::Disabled => true, + EnvOverride::Process => { + disabled_value("COPILOTKIT_TELEMETRY_DISABLED") || disabled_value("DO_NOT_TRACK") + } + } +} + +fn disabled_value(name: &str) -> bool { + env::var(name) + .ok() + .map(|value| matches!(value.trim().to_ascii_lowercase().as_str(), "1" | "true")) + .unwrap_or(false) +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) +} + +fn is_uuid(value: &str) -> bool { + let bytes = value.as_bytes(); + if bytes.len() != 36 { + return false; + } + for (index, byte) in bytes.iter().enumerate() { + if matches!(index, 8 | 13 | 18 | 23) { + if *byte != b'-' { + return false; + } + } else if !byte.is_ascii_hexdigit() { + return false; + } + } + true +} + +fn new_uuid() -> String { + let mut bytes = [0u8; 16]; + rand::rng().fill_bytes(&mut bytes); + bytes[6] = (bytes[6] & 0x0f) | 0x40; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + format!( + "{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}", + bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7], bytes[8], bytes[9], bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15] + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::path::PathBuf; + use std::sync::{mpsc, Barrier}; + use std::thread; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn temp_dir(name: &str) -> PathBuf { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let dir = std::env::temp_dir().join(format!("openbot-telemetry-{name}-{unique}")); + fs::create_dir_all(&dir).unwrap(); + dir + } + + fn context(version: &str, engine: Engine) -> Context { + Context { + distribution: Distribution::Desktop, + app_version: NumericVersion::parse(version).unwrap(), + platform: Platform::Macos, + arch: Architecture::Aarch64, + os_version: Some(NumericVersion::parse("14.5.0").unwrap()), + engine, + runtime_env: RuntimeEnv::Test, + } + } + + fn enabled(endpoint: String) -> Config { + Config { + enabled: true, + endpoint: Some(endpoint), + max_queue: 256, + } + } + + #[test] + fn numeric_version_accepts_only_normalized_numeric_versions() { + let version = NumericVersion::parse("1.2").unwrap(); + assert_eq!(version.to_string(), "1.2"); + assert_eq!(serde_json::to_value(&version).unwrap(), "1.2"); + let without_os = Context { + os_version: None, + ..context("1.2.3", Engine::None) + }; + assert!(serde_json::to_value(&without_os) + .unwrap() + .get("openbot_os_version") + .is_none()); + assert_eq!( + NumericVersion::parse("1.2.3.4").unwrap().to_string(), + "1.2.3.4" + ); + assert!(NumericVersion::parse("1").is_err()); + assert!(NumericVersion::parse("1.2.3.4.5").is_err()); + assert!(NumericVersion::parse("1.2-beta").is_err()); + assert!(NumericVersion::parse("123456789012345678901234567890123").is_err()); + } + + #[test] + fn event_schema_is_tagged_closed_and_exposes_centralized_name() { + let event = Event::new( + EventData::ModelChosen { + provider: Provider::Compatible, + credential_path: CredentialPath::ApiKey, + custom_base_url: true, + }, + context("1.2.3", Engine::None), + ); + + assert_eq!(event.event_name(), "oss.desktop.model_chosen"); + let json = serde_json::to_value(&event.data).unwrap(); + assert_eq!(json["kind"], "model_chosen"); + assert_eq!(json["provider"], "compatible"); + assert_eq!( + serde_json::to_value(EventData::HarnessChosen { + harness: Harness::MicrosoftAgentFramework + }) + .unwrap()["harness"], + "microsoft_agent_framework" + ); + assert!(serde_json::from_value::(serde_json::json!({ + "kind": "model_chosen", + "provider": "compatible", + "credential_path": "api_key", + "custom_base_url": true, + "extra": "rejected" + })) + .is_err()); + assert!(serde_json::from_value::(serde_json::json!({ + "kind": "setup_failed", + "step": "welcome", + "error_class": "raw user-facing error" + })) + .is_err()); + } + + #[test] + fn schema_lists_every_closed_event_kind_and_no_freeform_string_fields() { + let schema = schema_json(); + let events = schema["events"].as_object().unwrap(); + for sample in sample_events() { + let value = serde_json::to_value(&sample).unwrap(); + let kind = value["kind"].as_str().unwrap(); + assert!(events.contains_key(kind), "schema missing {kind}"); + assert_string_values_are_listed(&value, &events[kind]); + } + assert_eq!(schema["numeric_version_pattern"], r"^\d+(?:\.\d+){1,3}$"); + } + + fn sample_events() -> Vec { + vec![ + EventData::StepViewed { + step: Step::Welcome, + }, + EventData::HarnessChosen { + harness: Harness::ByoUrl, + }, + EventData::ModelChosen { + provider: Provider::Openai, + credential_path: CredentialPath::Subscription, + custom_base_url: false, + }, + EventData::EngineDetected { + engine: Engine::Docker, + responding: true, + }, + EventData::EngineInstalled { + engine: Engine::Podman, + outcome: EngineInstallOutcome::Failure, + }, + EventData::WindowsStage { + outcome: WindowsStageOutcome::WslNoKernel, + }, + EventData::ImagePull { + outcome: Outcome::Success, + duration_ms: 12, + bytes: Some(34), + }, + EventData::SetupFailed { + step: Step::Install, + error_class: SetupErrorClass::ImagePullFailed, + }, + EventData::Activated, + EventData::SetupAbandoned { step: Step::Ask }, + ] + } + + fn assert_string_values_are_listed(value: &serde_json::Value, schema: &serde_json::Value) { + if let Some(object) = value.as_object() { + for (key, child) in object { + if key == "kind" { + continue; + } + if let Some(string) = child.as_str() { + let allowed = schema[key] + .as_array() + .expect("string fields must be enum arrays"); + assert!( + allowed.iter().any(|candidate| candidate == string), + "{key}={string} not in schema" + ); + } else if child.is_object() { + assert_string_values_are_listed(child, &schema[key]); + } + } + } + } + + #[test] + fn disabled_mode_drops_pending_state_without_reading_or_replaying() { + let dir = temp_dir("disabled"); + let state = PersistedState { + schema_version: STATE_SCHEMA_VERSION, + install_id: "11111111-1111-4111-8111-111111111111".to_string(), + activated: false, + last_step: None, + queue: vec![Event::new( + EventData::Activated, + context("1.2.3", Engine::None), + )], + }; + fs::write(state_path(&dir), serde_json::to_vec_pretty(&state).unwrap()).unwrap(); + + let telemetry = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + enabled("http://127.0.0.1:9/ingest".to_string()), + EnvOverride::Disabled, + ) + .unwrap(); + telemetry.record(EventData::Activated).unwrap(); + telemetry.flush().unwrap(); + + assert!(telemetry.install_id().is_none()); + assert!(telemetry.install_id().is_none()); + assert!(!state_path(&dir).exists()); + } + + #[test] + fn adjacent_duplicate_step_viewed_is_suppressed_only_within_current_session() { + let dir = temp_dir("step-dedupe"); + let first = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled, + ) + .unwrap(); + first + .record(EventData::StepViewed { + step: Step::Welcome, + }) + .unwrap(); + first + .record(EventData::StepViewed { + step: Step::Welcome, + }) + .unwrap(); + assert_eq!( + first + .snapshot() + .unwrap() + .queue + .iter() + .filter(|event| matches!( + event.data, + EventData::StepViewed { + step: Step::Welcome + } + )) + .count(), + 1 + ); + first + .record(EventData::HarnessChosen { + harness: Harness::ByoUrl, + }) + .unwrap(); + first + .record(EventData::StepViewed { + step: Step::Welcome, + }) + .unwrap(); + assert_eq!( + first + .snapshot() + .unwrap() + .queue + .iter() + .filter(|event| matches!( + event.data, + EventData::StepViewed { + step: Step::Welcome + } + )) + .count(), + 2 + ); + drop(first); + + let second = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled, + ) + .unwrap(); + second + .record(EventData::StepViewed { + step: Step::Welcome, + }) + .unwrap(); + let state = second.snapshot().unwrap(); + assert_eq!( + state + .queue + .iter() + .filter(|event| matches!( + event.data, + EventData::StepViewed { + step: Step::Welcome + } + )) + .count(), + 3 + ); + assert_eq!( + state + .queue + .iter() + .filter(|event| matches!( + event.data, + EventData::SetupAbandoned { + step: Step::Welcome + } + )) + .count(), + 1 + ); + } + + #[test] + fn record_persists_before_send_and_flush_preserves_until_success() { + let dir = temp_dir("queue"); + let telemetry = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + enabled("http://127.0.0.1:9/ingest".to_string()), + EnvOverride::Enabled, + ) + .unwrap(); + + telemetry + .record(EventData::StepViewed { + step: Step::Harness, + }) + .unwrap(); + assert!(fs::read_to_string(state_path(&dir)) + .unwrap() + .contains("step_viewed")); + assert!(telemetry.flush().is_err()); + assert_eq!( + serde_json::from_str::(&fs::read_to_string(state_path(&dir)).unwrap()) + .unwrap() + .queue + .len(), + 1 + ); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/ingest", listener.local_addr().unwrap()); + let server = thread::spawn(move || { + let first = one_response(listener.try_clone().unwrap()); + let second = one_response(listener); + format!("{first}\n{second}") + }); + let replay = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + enabled(endpoint), + EnvOverride::Enabled, + ) + .unwrap(); + replay.flush().unwrap(); + let request = server.join().unwrap(); + assert!(request.contains("oss.desktop.step_viewed")); + assert!(request + .to_ascii_lowercase() + .contains("x-copilotkit-telemetry-id")); + assert!(request.contains("oss.desktop.setup_abandoned")); + assert!(request.contains("openbot_distribution")); + assert_eq!( + serde_json::from_str::(&fs::read_to_string(state_path(&dir)).unwrap()) + .unwrap() + .queue + .len(), + 0 + ); + } + + #[test] + fn concurrent_flush_does_not_drop_unsent_second_event() { + let dir = temp_dir("concurrent"); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/ingest", listener.local_addr().unwrap()); + let telemetry = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + enabled(endpoint), + EnvOverride::Enabled, + ) + .unwrap(); + telemetry + .record(EventData::StepViewed { + step: Step::Welcome, + }) + .unwrap(); + telemetry + .record(EventData::StepViewed { + step: Step::Harness, + }) + .unwrap(); + + let server = thread::spawn(move || one_response(listener)); + let barrier = Arc::new(Barrier::new(2)); + let a = Arc::clone(&telemetry); + let barrier_a = Arc::clone(&barrier); + let first = thread::spawn(move || { + barrier_a.wait(); + let _ = a.flush(); + }); + let b = Arc::clone(&telemetry); + let second = thread::spawn(move || { + barrier.wait(); + let _ = b.flush(); + }); + first.join().unwrap(); + second.join().unwrap(); + let _ = server.join().unwrap(); + + let state = telemetry.snapshot().unwrap(); + assert_eq!(state.queue.len(), 1); + assert!(matches!( + state.queue[0].data, + EventData::StepViewed { + step: Step::Harness + } + )); + } + + fn one_response(listener: TcpListener) -> String { + let (mut stream, _) = listener.accept().unwrap(); + let mut buffer = [0u8; 8192]; + let read = stream.read(&mut buffer).unwrap(); + stream.write_all(b"HTTP/1.1 202 Accepted\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}").unwrap(); + String::from_utf8_lossy(&buffer[..read]).into_owned() + } + + #[test] + fn shutdown_waits_for_active_flush_and_returns_only_after_abandonment_flushes() { + let dir = temp_dir("shutdown-active-flush"); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/ingest", listener.local_addr().unwrap()); + let telemetry = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + enabled(endpoint), + EnvOverride::Enabled, + ) + .unwrap(); + telemetry + .record(EventData::StepViewed { + step: Step::Welcome, + }) + .unwrap(); + + let (first_read_tx, first_read_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let server = thread::spawn(move || { + let (mut first_stream, _) = listener.accept().unwrap(); + let mut first_buffer = [0u8; 8192]; + let first_read = first_stream.read(&mut first_buffer).unwrap(); + first_read_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + first_stream.write_all(b"HTTP/1.1 202 Accepted\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}").unwrap(); + + let (mut second_stream, _) = listener.accept().unwrap(); + let mut second_buffer = [0u8; 8192]; + let second_read = second_stream.read(&mut second_buffer).unwrap(); + second_stream.write_all(b"HTTP/1.1 202 Accepted\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}").unwrap(); + format!( + "{}\n{}", + String::from_utf8_lossy(&first_buffer[..first_read]), + String::from_utf8_lossy(&second_buffer[..second_read]) + ) + }); + + let flushing = Arc::clone(&telemetry); + let flush_thread = thread::spawn(move || flushing.flush().unwrap()); + first_read_rx.recv().unwrap(); + + let shutting_down = Arc::clone(&telemetry); + let shutdown_thread = thread::spawn(move || shutting_down.shutdown().unwrap()); + sleep(Duration::from_millis(50)); + assert!( + !shutdown_thread.is_finished(), + "shutdown returned while active flush was still blocked" + ); + release_tx.send(()).unwrap(); + shutdown_thread.join().unwrap(); + assert_eq!(telemetry.snapshot().unwrap().queue.len(), 0); + flush_thread.join().unwrap(); + let requests = server.join().unwrap(); + + assert!(requests.contains("oss.desktop.step_viewed")); + assert!(requests.contains("oss.desktop.setup_abandoned")); + } + + #[test] + fn pending_events_keep_record_time_context() { + let dir = temp_dir("context"); + let telemetry = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled, + ) + .unwrap(); + telemetry + .record(EventData::EngineDetected { + engine: Engine::None, + responding: false, + }) + .unwrap(); + telemetry.update_engine(Engine::Docker).unwrap(); + let state = telemetry.snapshot().unwrap(); + assert_eq!(state.queue[0].context.engine, Engine::None); + } + + #[test] + fn activation_and_shutdown_abandonment_are_deduped_from_persisted_state() { + let dir = temp_dir("activation"); + let telemetry = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled, + ) + .unwrap(); + telemetry + .record(EventData::StepViewed { step: Step::Ask }) + .unwrap(); + telemetry.record(EventData::Activated).unwrap(); + telemetry.record(EventData::Activated).unwrap(); + telemetry.shutdown().unwrap(); + let state = telemetry.snapshot().unwrap(); + assert_eq!( + state + .queue + .iter() + .filter(|event| matches!(event.data, EventData::Activated)) + .count(), + 1 + ); + assert!(!state + .queue + .iter() + .any(|event| matches!(event.data, EventData::SetupAbandoned { .. }))); + } + + #[test] + fn quit_abandonment_is_not_duplicated_on_next_launch() { + let dir = temp_dir("quit-reopen"); + let first = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled, + ) + .unwrap(); + first + .record(EventData::StepViewed { step: Step::Ask }) + .unwrap(); + first.shutdown().unwrap(); + drop(first); + + let second = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled, + ) + .unwrap(); + let state = second.snapshot().unwrap(); + assert_eq!( + state + .queue + .iter() + .filter(|event| matches!(event.data, EventData::SetupAbandoned { step: Step::Ask })) + .count(), + 1 + ); + } + + #[test] + fn next_launch_records_abandoned_step_after_crash_before_activation() { + let dir = temp_dir("abandoned"); + let first = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled, + ) + .unwrap(); + first + .record(EventData::StepViewed { + step: Step::Install, + }) + .unwrap(); + drop(first); + + let second = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled, + ) + .unwrap(); + let state = second.snapshot().unwrap(); + assert!(state.queue.iter().any(|event| matches!( + event.data, + EventData::SetupAbandoned { + step: Step::Install + } + ))); + } + + #[test] + fn state_deserialize_refuses_unknown_fields_and_untyped_events() { + let dir = temp_dir("schema"); + let raw = serde_json::json!({ + "schema_version": 1, + "install_id": "11111111-1111-4111-8111-111111111111", + "activated": false, + "last_step": null, + "queue": [{ + "event_id": "22222222-2222-4222-8222-222222222222", + "occurred_at_ms": 1, + "event_name": "oss.desktop.activated", + "context": context("1.2.3", Engine::None), + "data": { "kind": "activated" }, + "unknown": true + }] + }); + fs::write(state_path(&dir), serde_json::to_vec_pretty(&raw).unwrap()).unwrap(); + assert!(Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config::disabled_for_tests(), + EnvOverride::Enabled + ) + .is_err()); + } + + #[test] + fn queue_is_bounded_to_maximum() { + let dir = temp_dir("bounded"); + let telemetry = Telemetry::open_with_env( + &dir, + context("1.2.3", Engine::None), + Config { + enabled: true, + endpoint: None, + max_queue: 2, + }, + EnvOverride::Enabled, + ) + .unwrap(); + telemetry + .record(EventData::StepViewed { + step: Step::Welcome, + }) + .unwrap(); + telemetry + .record(EventData::StepViewed { + step: Step::Harness, + }) + .unwrap(); + telemetry + .record(EventData::StepViewed { step: Step::Model }) + .unwrap(); + let state = telemetry.snapshot().unwrap(); + assert_eq!(state.queue.len(), 2); + assert!(matches!( + state.queue[0].data, + EventData::StepViewed { + step: Step::Harness + } + )); + } +} diff --git a/desktop/src/App.test.tsx b/desktop/src/App.test.tsx index 52b11bd22..fc1060d8f 100644 --- a/desktop/src/App.test.tsx +++ b/desktop/src/App.test.tsx @@ -2,6 +2,7 @@ import { afterAll, afterEach, beforeAll, expect, mock, test } from "bun:test"; import { GlobalRegistrator } from "@happy-dom/global-registrator"; import { act, cleanup, render, waitFor } from "@testing-library/react"; import userEvent from "@testing-library/user-event"; +import { StrictMode } from "react"; type Invoke = (command: string, args?: unknown) => Promise; type Deferred = { @@ -49,6 +50,87 @@ async function renderApp() { return view; } +function setupEvents() { + return invokeCalls + .filter((call) => call.command === "record_setup_event") + .map((call) => call.args); +} + +test("setup discloses default telemetry without a consent gate and deduplicates viewed steps", async () => { + useRootConfigurationSetup("/tmp/private-setup-root", async () => + emptyConfiguration(), + ); + const previous = invokeHandler; + invokeHandler = async (command, args) => { + if (command === "record_setup_event") return null; + return previous(command, args); + }; + let view!: ReturnType; + await act(async () => { + view = render( + + + , + ); + }); + + expect(view.getByText(/COPILOTKIT_TELEMETRY_DISABLED=1/)).toBeTruthy(); + expect(view.getByText(/DO_NOT_TRACK=1/)).toBeTruthy(); + expect(view.queryByRole("checkbox")).toBeNull(); + expect(view.queryByRole("switch")).toBeNull(); + expect(setupEvents()).toEqual([ + { event: { kind: "step_viewed", step: "welcome" } }, + ]); + await userEvent.click(view.getByRole("button", { name: "Set up OpenBot" })); + await userEvent.click(view.getByRole("button", { name: "Continue" })); + expect(setupEvents()).toEqual([ + { event: { kind: "step_viewed", step: "welcome" } }, + { event: { kind: "step_viewed", step: "harness" } }, + { event: { kind: "harness_chosen", harness: "langgraph" } }, + { event: { kind: "step_viewed", step: "model" } }, + ]); + await userEvent.click(view.getByRole("button", { name: "Back" })); + expect(setupEvents().at(-1)).toEqual({ + event: { kind: "step_viewed", step: "harness" }, + }); +}); + +test("setup records only model categories and reaches Ask when telemetry is unavailable", async () => { + useCompatibleEndpointSetup({}); + const previous = invokeHandler; + invokeHandler = async (command, args) => { + if (command === "record_setup_event") throw new Error("offline"); + return previous(command, args); + }; + const privateUrl = "https://private-model.example/v1"; + const privateKey = "synthetic-secret-endpoint-key"; + const view = await enterCompatibleEndpoint(privateUrl, privateKey); + await userEvent.click(view.getByRole("button", { name: "Continue" })); + expect(setupEvents()).toContainEqual({ + event: { + kind: "model_chosen", + provider: "compatible", + credential_path: "api_key", + custom_base_url: true, + }, + }); + await userEvent.click(view.getByRole("button", { name: "Start OpenBot" })); + await view.findByRole("button", { name: "Ask" }); + expect(setupEvents().at(-1)).toEqual({ + event: { kind: "step_viewed", step: "ask" }, + }); + const serialized = JSON.stringify(setupEvents()); + for (const privateValue of [ + privateUrl, + privateKey, + "local-model", + "/tmp/openbot-app-test", + ]) { + expect(serialized).not.toContain(privateValue); + } + expect(view.queryByText(/Nothing here leaves this computer/)).toBeNull(); +}); + type StartStackPayload = { root?: unknown; apiKey?: unknown; diff --git a/desktop/src/App.tsx b/desktop/src/App.tsx index d21081f2f..39ff0fcce 100644 --- a/desktop/src/App.tsx +++ b/desktop/src/App.tsx @@ -15,6 +15,12 @@ import { } from "./ProviderPicker"; import { Welcome } from "./Welcome"; import { isHttpEndpointUrl } from "./http-endpoint-url"; +import { + harnessChoiceEvent, + modelChoiceEvent, + recordSetupEvent, + type SetupStep, +} from "./telemetry"; type EngineStatus = { engine: "docker" | "podman" | null; @@ -129,14 +135,22 @@ export function App() { setSigningIn(false); } } - const [step, setStep] = useState< - "welcome" | "harness" | "model" | "install" | "ask" - >("welcome"); + const [step, setStep] = useState("welcome"); const [apiUrl, setApiUrl] = useState(MANAGED_INTELLIGENCE_API_URL); const [wsUrl, setWsUrl] = useState(MANAGED_INTELLIGENCE_GATEWAY_WS_URL); const [steps, setSteps] = useState([]); const [busy, setBusy] = useState(false); const [running, setRunning] = useState(false); + const visibleSetupStep = + blockerFailure || blocker || (running && step !== "ask") ? null : step; + const lastViewedStep = useRef(null); + useEffect(() => { + if (visibleSetupStep === lastViewedStep.current) return; + lastViewedStep.current = visibleSetupStep; + if (visibleSetupStep !== null) { + recordSetupEvent({ kind: "step_viewed", step: visibleSetupStep }); + } + }, [visibleSetupStep]); const configuredRunRef = useRef(0); /* * A failure, in both registers. @@ -414,6 +428,9 @@ export function App() { chosen={harness} onChoose={setHarness} onContinue={() => { + recordSetupEvent( + harnessChoiceEvent(harness?.id ?? DEFAULT_HARNESS), + ); setHarness((choice) => choice?.id === "byo-url" ? { ...choice, agentUrl: choice.agentUrl?.trim() } @@ -460,6 +477,7 @@ export function App() { root={root} chosen={model} onChoose={(choice) => { + recordSetupEvent(modelChoiceEvent(choice)); setModel(choice); setStep("install"); }} diff --git a/desktop/src/Ask.tsx b/desktop/src/Ask.tsx index c63a70024..65a6e0eab 100644 --- a/desktop/src/Ask.tsx +++ b/desktop/src/Ask.tsx @@ -126,8 +126,9 @@ export function Ask({

- Nothing here leaves this computer except the question, which goes to the - AI provider you connected. + Your question goes to the AI provider you connected. Setup telemetry + records whether your Bot answered, without including your question or + its answer.

); diff --git a/desktop/src/Welcome.tsx b/desktop/src/Welcome.tsx index 3037c0617..e0f6d8279 100644 --- a/desktop/src/Welcome.tsx +++ b/desktop/src/Welcome.tsx @@ -26,6 +26,14 @@ export function Welcome({ onStart }: { onStart: () => void }) { Takes a few minutes. OpenBot installs what it needs and asks you to sign in to the AI plan you already have.

+

+ OpenBot sends setup and usage statistics to CopilotKit by default: a + random installation ID, setup steps, platform and engine details, + download performance, and Bot and connection choices. Telemetry does not + include prompts, files, credentials, email addresses, or server URLs. To + turn it off, launch OpenBot with COPILOTKIT_TELEMETRY_DISABLED=1 or + DO_NOT_TRACK=1. +

); } diff --git a/desktop/src/telemetry.test.ts b/desktop/src/telemetry.test.ts new file mode 100644 index 000000000..8681d5dac --- /dev/null +++ b/desktop/src/telemetry.test.ts @@ -0,0 +1,85 @@ +import { expect, test } from "bun:test"; +import { harnessChoiceEvent, modelChoiceEvent } from "./telemetry"; + +test("harness telemetry admits catalogue enums and excludes URLs and unknown IDs", () => { + expect(harnessChoiceEvent("byo-url")).toEqual({ + kind: "harness_chosen", + harness: "byo_url", + }); + expect(harnessChoiceEvent("claude-agent-sdk")).toEqual({ + kind: "harness_chosen", + harness: "claude_agent_sdk", + }); + for (const unknown of [ + "https://private-agent.example", + "future-harness", + "__proto__", + "toString", + ]) { + expect(harnessChoiceEvent(unknown)).toBeNull(); + } +}); + +test.each(["openai", "anthropic"])( + "%s subscription and API-key telemetry contain only closed categories", + (provider) => { + expect( + modelChoiceEvent({ + provider, + login: "plan", + token: "synthetic-private-token", + }), + ).toEqual({ + kind: "model_chosen", + provider, + credential_path: "subscription", + custom_base_url: false, + }); + expect( + modelChoiceEvent({ + provider, + login: "api-key", + apiKey: "synthetic-private-key", + saved: true, + }), + ).toEqual({ + kind: "model_chosen", + provider, + credential_path: "api_key", + custom_base_url: false, + }); + }, +); + +test("a custom endpoint becomes a boolean without exporting its URL, model, or credentials", () => { + expect( + modelChoiceEvent({ + provider: "openai-compatible", + login: "endpoint", + baseUrl: "https://private-model.example/v1", + containerBaseUrl: "http://private-container:9000/v1", + model: "private-model-name", + apiKey: "synthetic-private-key", + }), + ).toEqual({ + kind: "model_chosen", + provider: "compatible", + credential_path: "api_key", + custom_base_url: true, + }); +}); + +test("a skipped model is explicit and unsupported providers are not exported", () => { + expect(modelChoiceEvent(null)).toEqual({ + kind: "model_chosen", + provider: "none", + credential_path: "none", + custom_base_url: false, + }); + expect( + modelChoiceEvent({ provider: "private-provider", login: "api-key" }), + ).toBeNull(); + expect( + modelChoiceEvent({ provider: "openai", login: "endpoint" }), + ).toBeNull(); +}); diff --git a/desktop/src/telemetry.ts b/desktop/src/telemetry.ts new file mode 100644 index 000000000..8af42614d --- /dev/null +++ b/desktop/src/telemetry.ts @@ -0,0 +1,80 @@ +import { invoke } from "@tauri-apps/api/core"; +import type { ModelChoice } from "./ProviderPicker"; + +export type SetupStep = "welcome" | "harness" | "model" | "install" | "ask"; + +const HARNESSES = { + crewai: "crewai", + llamaindex: "llamaindex", + agno: "agno", + langgraph: "langgraph", + "google-adk": "google_adk", + "pydantic-ai": "pydantic_ai", + "microsoft-agent-framework": "microsoft_agent_framework", + "claude-agent-sdk": "claude_agent_sdk", + strands: "strands", + ag2: "ag2", + langroid: "langroid", + mastra: "mastra", + "byo-url": "byo_url", +} as const; + +type Harness = (typeof HARNESSES)[keyof typeof HARNESSES]; + +export type SetupEvent = + | { kind: "step_viewed"; step: SetupStep } + | { kind: "harness_chosen"; harness: Harness } + | { + kind: "model_chosen"; + provider: "openai" | "anthropic" | "compatible" | "none"; + credential_path: "subscription" | "api_key" | "none"; + custom_base_url: boolean; + }; + +export function harnessChoiceEvent(id: string): SetupEvent | null { + // Match values rather than indexing with an arbitrary string, including prototype keys. + for (const [known, harness] of Object.entries(HARNESSES)) { + if (id === known) return { kind: "harness_chosen", harness }; + } + return null; +} + +export function modelChoiceEvent( + choice: ModelChoice | null, +): SetupEvent | null { + if (choice === null) { + return { + kind: "model_chosen", + provider: "none", + credential_path: "none", + custom_base_url: false, + }; + } + const { provider, login } = choice; + if ( + (provider === "openai" || provider === "anthropic") && + (login === "plan" || login === "api-key") + ) { + return { + kind: "model_chosen", + provider, + credential_path: login === "plan" ? "subscription" : "api_key", + custom_base_url: Boolean(choice.baseUrl?.trim()), + }; + } + if (provider === "openai-compatible" && login === "endpoint") { + return { + kind: "model_chosen", + provider: "compatible", + credential_path: "api_key", + custom_base_url: Boolean(choice.baseUrl?.trim()), + }; + } + return null; +} + +export function recordSetupEvent(event: SetupEvent | null): void { + if (event === null) return; + // Native code owns validation, opt-out, and delivery. Analytics never gate setup. + void invoke("record_setup_event", { event }).catch(() => undefined); +} diff --git a/server/src/copilot.ts b/server/src/copilot.ts index 2cb9e1411..ceb434923 100644 --- a/server/src/copilot.ts +++ b/server/src/copilot.ts @@ -25,6 +25,7 @@ import { } from "./channels/attachment-parts"; import type { AgentFetch, StallGuard } from "./channels/stall-guard"; import type { DeploymentConfig } from "./config"; +import { desktopTelemetryProperties } from "./desktop-telemetry"; import type { SelectableSkill, Selection } from "./plugins/selection"; import { latestUserText, @@ -2063,9 +2064,10 @@ export function mountCopilotRuntime( licenseToken: intelligence.licenseToken, // Carried on the events the runtime already sends, so OpenBot's traffic is separable from any // other deployment's. Adds no events of its own. - ...(config.accessibility - ? { telemetryProperties: { accessibility_title: "OpenBot" } } - : {}), + telemetryProperties: { + ...(config.accessibility ? { accessibility_title: "OpenBot" } : {}), + ...desktopTelemetryProperties(), + }, /* * What lets a Bot answer with an interface it wrote itself. * diff --git a/server/src/desktop-telemetry.test.ts b/server/src/desktop-telemetry.test.ts new file mode 100644 index 000000000..077b02fe3 --- /dev/null +++ b/server/src/desktop-telemetry.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, test } from "bun:test"; +import { desktopTelemetryProperties } from "./desktop-telemetry"; + +describe("desktop runtime metadata", () => { + test("leaves ordinary server deployments untagged", () => { + expect(desktopTelemetryProperties({})).toEqual({}); + expect( + desktopTelemetryProperties({ OPENBOT_DISTRIBUTION: "server" }), + ).toEqual({}); + }); + + test("carries only the shell's bounded metadata", () => { + expect( + desktopTelemetryProperties({ + OPENBOT_DISTRIBUTION: "desktop", + OPENBOT_VERSION: "0.0.9", + OPENBOT_PLATFORM: "macos", + OPENBOT_ARCH: "aarch64", + OPENBOT_OS_VERSION: "15.6.1", + OPENBOT_ENGINE: "podman", + OPENAI_API_KEY: "synthetic-secret", + CPK_TELEMETRY_ID: "identity-belongs-in-the-transport", + HOME: "/Users/private-name", + OPENBOT_BASE_URL: "https://private.example", + }), + ).toEqual({ + openbot_distribution: "desktop", + openbot_version: "0.0.9", + openbot_platform: "macos", + openbot_arch: "aarch64", + openbot_os_version: "15.6.1", + openbot_engine: "podman", + }); + }); + + test.each([ + "/Users/private-name", + "private.example", + "1.2-private-name", + "1.2\nsecret", + "1.2\n", + "1.2.3.4.5", + "1".repeat(40), + ])("rejects arbitrary text in every metadata field: %s", (value) => { + expect( + desktopTelemetryProperties({ + OPENBOT_DISTRIBUTION: "desktop", + OPENBOT_VERSION: value, + OPENBOT_OS_VERSION: value, + OPENBOT_PLATFORM: value, + OPENBOT_ARCH: value, + OPENBOT_ENGINE: value, + }), + ).toEqual({ openbot_distribution: "desktop" }); + }); +}); diff --git a/server/src/desktop-telemetry.ts b/server/src/desktop-telemetry.ts new file mode 100644 index 000000000..a20ad59d6 --- /dev/null +++ b/server/src/desktop-telemetry.ts @@ -0,0 +1,38 @@ +/** Only the desktop shell's closed metadata may join the runtime's existing events. */ +export function desktopTelemetryProperties( + env: Record = process.env, +): Record { + if (env.OPENBOT_DISTRIBUTION !== "desktop") return {}; + + const properties: Record = { + openbot_distribution: "desktop", + }; + for (const [input, output, allowed] of [ + [ + "OPENBOT_PLATFORM", + "openbot_platform", + ["macos", "windows", "linux", "other"], + ], + ["OPENBOT_ARCH", "openbot_arch", ["aarch64", "x86_64", "other"]], + ["OPENBOT_ENGINE", "openbot_engine", ["docker", "podman", "none"]], + ] as const) { + const value = env[input]; + if (value && allowed.some((item) => item === value)) + properties[output] = value; + } + for (const [input, output] of [ + ["OPENBOT_VERSION", "openbot_version"], + ["OPENBOT_OS_VERSION", "openbot_os_version"], + ] as const) { + const value = env[input]; + if ( + value && + value.length <= 32 && + value.trim() === value && + /^\d+(?:\.\d+){1,3}$/.test(value) + ) { + properties[output] = value; + } + } + return properties; +}